FesapiHdfProxy.h

/*-----------------------------------------------------------------------
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements.  See the NOTICE file
distributed with this work for additional information
regarding copyright ownership.  The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"; you may not use this file except in compliance
with the License.  You may obtain a copy of the License at
 
  http://www.apache.org/licenses/LICENSE-2.0
 
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied.  See the License for the
specific language governing permissions and limitations
under the License.
-----------------------------------------------------------------------*/
#pragma once
 
#include <set>
#include <type_traits>
 
#include <fesapi/common/HdfProxyFactory.h>
 
#include "../AbstractSession.h"
#include "../protocolHandlers/GetFullDataArrayHandlers.h"
 
namespace ETP_NS
{
    class FETPAPI_DLL_IMPORT_OR_EXPORT FesapiHdfProxy : public EML2_NS::AbstractHdfProxy
    {
    public:

        FesapiHdfProxy(AbstractSession* session, COMMON_NS::DataObjectRepository * repo, const std::string & guid, const std::string & title, const std::string & packageDirAbsolutePath, const std::string & externalFilePath, COMMON_NS::DataObjectRepository::openingMode hdfPermissionAccess) :
            EML2_NS::AbstractHdfProxy(packageDirAbsolutePath, externalFilePath, hdfPermissionAccess), session_(session), compressionLevel(0) {
            xmlNs_ = repo->getDefaultEmlVersion() == COMMON_NS::DataObjectRepository::EnergisticsStandard::EML2_3 ? "eml23" : "eml20";
            initGsoapProxy(repo, guid, title, 20);
        }

        FesapiHdfProxy(AbstractSession* session, gsoap_resqml2_0_1::_eml20__EpcExternalPartReference* fromGsoap) :
            EML2_NS::AbstractHdfProxy(fromGsoap), session_(session), compressionLevel(0), xmlNs_("eml20") {}

        FesapiHdfProxy(AbstractSession* session, const COMMON_NS::DataObjectReference& dor) :
            EML2_NS::AbstractHdfProxy(dor), session_(session), compressionLevel(0) {
            std::string ct = dor.getContentType();
            if (ct.find("2.0") != std::string::npos) {
                xmlNs_ = "eml20";
            }
            else if (ct.find("2.3") != std::string::npos) {
                xmlNs_ = "eml23";
            }
            else {
                throw std::invalid_argument("The HDF Proxy DataObjectReference does not look to be eml20 or eml23 neither : " + ct);
            }
        }
 

        ~FesapiHdfProxy() = default;
 
        AbstractSession* getSession() { return session_; }
 
        void setSession(AbstractSession* session) { session_ = session; }

        void open() final {}

        bool isOpened() const final { return session_ != nullptr && !session_->isWebSocketSessionClosed();  }

        void close() final {}
 
        /*
        * Get the used (native) datatype in a dataset
        */
        COMMON_NS::AbstractObject::numericalDatatypeEnum getNumericalDatatype(const std::string & groupName) final;

        int getHdfDatatypeClassInDataset(const std::string & datasetName) final;

        void writeItemizedListOfList(const std::string & groupName,
            const std::string & name,
            COMMON_NS::AbstractObject::numericalDatatypeEnum cumulativeLengthDatatype,
            const void * cumulativeLength,
            uint64_t cumulativeLengthSize,
            COMMON_NS::AbstractObject::numericalDatatypeEnum elementsDatatype,
            const void * elements,
            uint64_t elementsSize) final;

        std::vector<uint32_t> getElementCountPerDimension(const std::string & datasetName) final;

        void setCompressionLevel(unsigned int newCompressionLevel) final { if (newCompressionLevel > 9) compressionLevel = 9; else compressionLevel = newCompressionLevel; }

        void writeArrayNd(const std::string & groupName,
            const std::string & name,
            COMMON_NS::AbstractObject::numericalDatatypeEnum datatype,
            const void * values,
            const uint64_t * numValuesInEachDimension,
            unsigned int numDimensions) final;

        void createArrayNd(
            const std::string& groupName,
            const std::string& name,
            COMMON_NS::AbstractObject::numericalDatatypeEnum datatype,
            const uint64_t* numValuesInEachDimension,
            unsigned int numDimensions) final;

        void writeArrayNdSlab(
            const std::string& groupName,
            const std::string& name,
            COMMON_NS::AbstractObject::numericalDatatypeEnum datatype,
            const void* values,
            const uint64_t* numValuesInEachDimension,
            const uint64_t* offsetValuesInEachDimension,
            unsigned int numDimensions) final;

        std::set<int64_t> async_writeArrayNdSlab(
            const std::string& groupName,
            const std::string& name,
            COMMON_NS::AbstractObject::numericalDatatypeEnum datatype,
            const void* values,
            const uint64_t* numValuesInEachDimension,
            const uint64_t* offsetValuesInEachDimension,
            unsigned int numDimensions);

        void writeGroupAttributes(const std::string&,
            const std::vector<std::string>&,
            const std::vector<std::string>&) final {
            throw std::logic_error("Group attributes are not supported in ETP1.2");
        }

        void writeGroupAttribute(const std::string&,
            const std::string&,
            const std::vector<std::string>&) final {
            throw std::logic_error("Group attributes are not supported in ETP1.2");
        }

        void writeGroupAttributes(const std::string&,
            const std::vector<std::string>&,
            const std::vector<double>&) {
            throw std::logic_error("Group attributes are not supported in ETP1.2");
        }

        void writeGroupAttributes(const std::string&,
            const std::vector<std::string>&,
            const std::vector<int>&) final {
            throw std::logic_error("Group attributes are not supported in ETP1.2");
        }

        void writeDatasetAttributes(const std::string&,
            const std::vector<std::string>&,
            const std::vector<std::string>&) final {
            throw std::logic_error("Dataset attributes are not supported in ETP1.2");
        }

        void writeDatasetAttribute(const std::string&,
            const std::string&,
            const std::vector<std::string>&) final {
            throw std::logic_error("Dataset attributes are not supported in ETP1.2");
        }

        void writeDatasetAttributes(const std::string&,
            const std::vector<std::string>&,
            const std::vector<double>&) final {
            throw std::logic_error("Dataset attributes are not supported in ETP1.2");
        }

        void writeDatasetAttributes(const std::string&,
            const std::vector<std::string>&,
            const std::vector<int>&) final {
            throw std::logic_error("Dataset attributes are not supported in ETP1.2");
        }
 
        std::string readStringAttribute(const std::string&,
            const std::string&) const final {
            throw std::logic_error("Attributes are not supported in ETP1.2");
        }
 
        std::vector<std::string> readStringArrayAttribute(const std::string&,
            const std::string&) const final {
            throw std::logic_error("Attributes are not supported in ETP1.2");
        }
 
        double readDoubleAttribute(const std::string&,
            const std::string&) const final {
            throw std::logic_error("Attributes are not supported in ETP1.2");
        }
 
        int64_t readInt64Attribute(const std::string&,
            const std::string&) const final {
            throw std::logic_error("Attributes are not supported in ETP1.2");
        }

        void readArrayNdOfDoubleValues(const std::string & datasetName, double* values) final { readArrayNdOfValues(datasetName, values); }

        void readArrayNdOfDoubleValues(
            const std::string & datasetName,
            double* values,
            uint64_t const * numValuesInEachDimension,
            uint64_t const * offsetInEachDimension,
            unsigned int numDimensions) final;

        void readArrayNdOfDoubleValues(
            const std::string & datasetName, double* values,
            uint64_t const * blockCountPerDimension,
            uint64_t const * offsetInEachDimension,
            uint64_t const * strideInEachDimension,
            uint64_t const * blockSizeInEachDimension,
            unsigned int numDimensions) final;
 
        void selectArrayNdOfValues(
            const std::string & datasetName,
            uint64_t const * blockCountPerDimension,
            uint64_t const * offsetInEachDimension,
            uint64_t const * strideInEachDimension,
            uint64_t const * blockSizeInEachDimension,
            unsigned int numDimensions,
            bool newSelection,
            hdf5_hid_t & dataset,
            hdf5_hid_t & filespace) final;

        void readArrayNdOfDoubleValues(
            hdf5_hid_t dataset,
            hdf5_hid_t filespace,
            void* values,
            uint64_t slabSize) final;

        void readArrayNdOfFloatValues(const std::string & datasetName, float* values) final { readArrayNdOfValues(datasetName, values); }

        void readArrayNdOfFloatValues(
            const std::string & datasetName,
            float* values,
            uint64_t const * numValuesInEachDimension,
            uint64_t const * offsetInEachDimension,
            unsigned int numDimensions) final;

        void readArrayNdOfInt64Values(const std::string & datasetName, int64_t* values) final { readArrayNdOfValues(datasetName, values); }

        void readArrayNdOfInt64Values(
            const std::string & datasetName,
            int64_t* values,
            uint64_t const * numValuesInEachDimension,
            uint64_t const * offsetInEachDimension,
            unsigned int numDimensions)  final;

        void readArrayNdOfUInt64Values(const std::string & datasetName, uint64_t* values) final { readArrayNdOfValues(datasetName, values); }

        void readArrayNdOfIntValues(const std::string & datasetName, int* values) final { readArrayNdOfValues(datasetName, values); }

        void readArrayNdOfIntValues(
            const std::string & datasetName,
            int* values,
            uint64_t const * numValuesInEachDimension,
            uint64_t const * offsetInEachDimension,
            unsigned int numDimensions
        ) final;

        void readArrayNdOfUIntValues(const std::string & datasetName, unsigned int* values) final { readArrayNdOfValues(datasetName, values); }

        void readArrayNdOfShortValues(const std::string & datasetName, short* values) final { readArrayNdOfValues(datasetName, values); }

        void readArrayNdOfUShortValues(const std::string & datasetName, unsigned short* values) final { readArrayNdOfValues(datasetName, values); }

        void readArrayNdOfInt8Values(const std::string & datasetName, int8_t* values) final { readArrayNdOfValues(datasetName, values); }

        void readArrayNdOfUInt8Values(const std::string & datasetName, uint8_t* values) final { readArrayNdOfValues(datasetName, values); }

        bool exist(const std::string & absolutePathInHdfFile) const final;

        bool isCompressed(const std::string &) final { return false; }
        
        std::vector<uint32_t> getElementCountPerChunkDimension(const std::string&) final { return {}; }

        std::string getXmlNamespace() const { return xmlNs_; }
 
    private:
        AbstractSession* session_;
        unsigned int compressionLevel;
        std::string xmlNs_;
        size_t maxArraySize_{ 12000000 }; // Bytes
 
        Energistics::Etp::v12::Datatypes::DataArrayTypes::DataArrayIdentifier buildDataArrayIdentifier(const std::string & datasetName) const;
        std::shared_ptr<Energistics::Etp::v12::Protocol::DataArray::GetDataArrays> buildGetDataArraysMessage(const std::string & datasetName) const;
        std::shared_ptr<Energistics::Etp::v12::Protocol::DataArray::GetDataArrayMetadata> buildGetDataArrayMetadataMessage(const std::string & datasetName) const;
 
        Energistics::Etp::v12::Datatypes::DataArrayTypes::DataArrayMetadata getDataArrayMetadata(const std::string & datasetName) const;
 
        template<typename T> void readArrayNdOfValues(const std::string & datasetName, T* values)
        {
            // First get metadata about the data array
            const Energistics::Etp::v12::Datatypes::DataArrayTypes::DataArrayMetadata daMetadata = getDataArrayMetadata(datasetName);
            const size_t valueCount = std::accumulate(daMetadata.dimensions.begin(), daMetadata.dimensions.end(), 1, std::multiplies<int64_t>());
 
            size_t valueSize = 1;
            switch (daMetadata.transportArrayType) {
            case Energistics::Etp::v12::Datatypes::AnyArrayType::bytes:
            case Energistics::Etp::v12::Datatypes::AnyArrayType::arrayOfBoolean: break;
            case Energistics::Etp::v12::Datatypes::AnyArrayType::arrayOfInt: valueSize = 6; break; // 25% more because of zig zag encoding worst case scenario
            case Energistics::Etp::v12::Datatypes::AnyArrayType::arrayOfFloat: valueSize = 4; break;
            case Energistics::Etp::v12::Datatypes::AnyArrayType::arrayOfLong:  valueSize = 10; break; // 25% more because of zig zag encoding worst case scenario
            case Energistics::Etp::v12::Datatypes::AnyArrayType::arrayOfDouble: valueSize = 8; break;
            default: throw std::logic_error("Array of strings are not implemented yet");
            }
            const size_t wholeSize = valueCount * valueSize;
 
            // maxAllowedDataArraySize is the maximum serialized size of the array (including avro extra longs for array blocks)
            const size_t maxAllowedDataArraySize = session_->getMaxWebSocketMessagePayloadSize()
                - sizeof(Energistics::Etp::v12::Datatypes::MessageHeader)
                - (daMetadata.dimensions.size() * 2 + 1) * 8; // *2 because the array can contain a maximum of element count block, +1 for the length of the last block : https://avro.apache.org/docs/1.10.2/spec.html#binary_encode_complex
 
            // Now get values of the data array
            auto specializedHandler = std::make_shared<GetFullDataArrayHandlers<T>>(session_, values);
            if (wholeSize + (valueCount + 1) * 8 <= maxAllowedDataArraySize) { // There can be valueCount array block and there is the length of the last array block
                // Get all values at once
                session_->sendWithSpecificHandlerAndBlock(
                    buildGetDataArraysMessage(datasetName),
                    specializedHandler,
                    0, 0x02);
            }
            else {
                // Get all values using several data subarrays allowing more granular streaming
                std::vector<int64_t> counts(daMetadata.dimensions.size(), 1);
 
                // Compute the dimensions of the subArrays to get
                size_t subArrayValueCount = 1;
                for (int64_t dimIndex = daMetadata.dimensions.size() - 1; dimIndex >= 0; --dimIndex) {
                    int64_t maxCountOnDim = daMetadata.preferredSubarrayDimensions.empty()
                        ? daMetadata.dimensions[dimIndex]
                        : daMetadata.preferredSubarrayDimensions[dimIndex];
                    subArrayValueCount *= maxCountOnDim;
                    int64_t allowedCountOnDim = maxCountOnDim;
                    while (subArrayValueCount * valueSize + (subArrayValueCount + 1) * 8 > maxAllowedDataArraySize) {
                        subArrayValueCount /= allowedCountOnDim;
                        allowedCountOnDim /= 2;
                        subArrayValueCount *= allowedCountOnDim;
                    }
                    counts[dimIndex] = allowedCountOnDim;
                    if (allowedCountOnDim != maxCountOnDim) {
                        break;
                    }
                }
 
                // Build the message
                auto msg = std::make_shared<Energistics::Etp::v12::Protocol::DataArray::GetDataSubarrays>();
                size_t subArrayIndex = 0;
                std::vector<int64_t> starts(daMetadata.dimensions.size(), 0);
                std::vector<int64_t> currentCounts = counts;
                bool hasParsedAllArray = false;
                while (!hasParsedAllArray) {
                    std::string subArrayIndexStr = std::to_string(subArrayIndex);
                    msg->dataSubarrays[subArrayIndexStr].uid = buildDataArrayIdentifier(datasetName);
                    msg->dataSubarrays[subArrayIndexStr].counts = currentCounts;
                    msg->dataSubarrays[subArrayIndexStr].starts = starts;
 
                    // next sub array to get
                    ++subArrayIndex;
                    hasParsedAllArray = true;
                    for (int64_t dimIndex = daMetadata.dimensions.size() - 1; dimIndex >= 0; --dimIndex) {
                        if (starts[dimIndex] + currentCounts[dimIndex] < daMetadata.dimensions[dimIndex]) {
                            starts[dimIndex] += currentCounts[dimIndex];
                            if (starts[dimIndex] + currentCounts[dimIndex] > daMetadata.dimensions[dimIndex]) {
                                currentCounts[dimIndex] = daMetadata.dimensions[dimIndex] - starts[dimIndex];
                            }
 
                            for (size_t dimIndex2 = dimIndex + 1; dimIndex2 < daMetadata.dimensions.size(); ++dimIndex2) {
                                starts[dimIndex2] = 0;
                                currentCounts[dimIndex2] = counts[dimIndex2];
                            }
                            hasParsedAllArray = false;
                            break;
                        }
                    }
 
                    specializedHandler->setDataSubarrays(subArrayIndexStr, msg->dataSubarrays[subArrayIndexStr]);
                }
 
                // Send message
                session_->sendWithSpecificHandlerAndBlock(
                    msg,
                    specializedHandler,
                    0, 0x02);
            }
        }
    };
 
    class FETPAPI_DLL_IMPORT_OR_EXPORT FesapiHdfProxyFactory : public COMMON_NS::HdfProxyFactory
    {
    public:
        FesapiHdfProxyFactory(AbstractSession* session) : session_(session) {}

        EML2_NS::AbstractHdfProxy* make(const COMMON_NS::DataObjectReference& dor) {
            FesapiHdfProxy* result = new FesapiHdfProxy(session_, dor);
            return result;
        }

        EML2_NS::AbstractHdfProxy* make(gsoap_resqml2_0_1::_eml20__EpcExternalPartReference* fromGsoap) {
            FesapiHdfProxy* result = new FesapiHdfProxy(session_, fromGsoap);
            return result;
        }
 
        EML2_NS::AbstractHdfProxy* make(COMMON_NS::DataObjectRepository * repo, const std::string & guid, const std::string & title,
            const std::string & packageDirAbsolutePath, const std::string & externalFilePath,
            COMMON_NS::DataObjectRepository::openingMode hdfPermissionAccess = COMMON_NS::DataObjectRepository::openingMode::READ_ONLY) {
            FesapiHdfProxy* result = new FesapiHdfProxy(session_, repo, guid, title, packageDirAbsolutePath, externalFilePath, hdfPermissionAccess);
            return result;
        }
 
    private:
        AbstractSession* session_;
    };
}