AbstractSession.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 <iostream>
#include <mutex>
#include <queue>
#include <unordered_map>
#include <utility>
 
#include <boost/asio.hpp>
#include <boost/beast/core.hpp>
#include <boost/beast/websocket.hpp>
#include <boost/uuid/uuid_io.hpp>
#include <boost/uuid/nil_generator.hpp>
 
#include "../nsDefinitions.h"
 
#include "EtpHelpers.h"
#include "ProtocolHandlers/CoreHandlers.h"
#include "ProtocolHandlers/DiscoveryHandlers.h"
#include "ProtocolHandlers/StoreHandlers.h"
#include "ProtocolHandlers/StoreNotificationHandlers.h"
#include "ProtocolHandlers/DataArrayHandlers.h"
#include "ProtocolHandlers/TransactionHandlers.h"
#include "ProtocolHandlers/DataspaceHandlers.h"
#include "ProtocolHandlers/DataspaceOSDUHandlers.h"
 
using tcp = boost::asio::ip::tcp;               // from <boost/asio/ip/tcp.hpp>
namespace websocket = boost::beast::websocket;  // from <boost/beast/websocket.hpp>
 
namespace ETP_NS
{
    class AbstractSession : public std::enable_shared_from_this<AbstractSession>
    {
    public:
 
        virtual ~AbstractSession() = default;
 
        const boost::uuids::uuid& getIdentifier() {
            if (isEtpSessionClosed()) {
                identifier = boost::uuids::nil_uuid();
            }
 
            return identifier;
        }
 
        void setTimeOut(double timeOut) {
            _timeOut = timeOut;
        }
 
        double getTimeOut() const {
            return _timeOut;
        }
 
        std::unordered_map<int64_t, Energistics::Etp::v12::Datatypes::Object::SubscriptionInfo> subscriptions;
 
        FETPAPI_DLL_IMPORT_OR_EXPORT void setCoreProtocolHandlers(std::shared_ptr<CoreHandlers> coreHandlers) {
            setProtocolHandlers(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::Core), coreHandlers);
        }
 
        FETPAPI_DLL_IMPORT_OR_EXPORT void setDiscoveryProtocolHandlers(std::shared_ptr<DiscoveryHandlers> discoveryHandlers) {
            setProtocolHandlers(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::Discovery), discoveryHandlers);
        }
 
        FETPAPI_DLL_IMPORT_OR_EXPORT void setStoreProtocolHandlers(std::shared_ptr<StoreHandlers> storeHandlers) {
            setProtocolHandlers(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::Store), storeHandlers);
        }
 
        FETPAPI_DLL_IMPORT_OR_EXPORT void setStoreNotificationProtocolHandlers(std::shared_ptr<ETP_NS::StoreNotificationHandlers> storeNotificationHandlers) {
            setProtocolHandlers(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::StoreNotification), storeNotificationHandlers);
        }
 
        FETPAPI_DLL_IMPORT_OR_EXPORT void setDataArrayProtocolHandlers(std::shared_ptr<DataArrayHandlers> dataArrayHandlers) {
            setProtocolHandlers(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::DataArray), dataArrayHandlers);
        }
 
        FETPAPI_DLL_IMPORT_OR_EXPORT void setTransactionProtocolHandlers(std::shared_ptr<TransactionHandlers> transactionHandlers) {
            setProtocolHandlers(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::Transaction), transactionHandlers);
        }
 
        FETPAPI_DLL_IMPORT_OR_EXPORT void setDataspaceProtocolHandlers(std::shared_ptr<DataspaceHandlers> dataspaceHandlers) {
            setProtocolHandlers(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::Dataspace), dataspaceHandlers);
        }
 
        FETPAPI_DLL_IMPORT_OR_EXPORT void setDataspaceOSDUProtocolHandlers(std::shared_ptr<DataspaceOSDUHandlers> dataspaceOsduHandlers) {
            setProtocolHandlers(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::DataspaceOSDU), dataspaceOsduHandlers);
        }
 
        template<typename T> int64_t send(const T & mb, int64_t correlationId = 0, int32_t messageFlags = 0)
        {
            return sendWithSpecificHandler(mb, protocolHandlers.at(mb.protocolId), correlationId, messageFlags);
        }
 
        template<typename T> void sendAndBlock(const T & mb, int64_t correlationId = 0, int32_t messageFlags = 0)
        {
            int64_t msgId = send(mb, correlationId, messageFlags);
 
            auto t_start = std::chrono::high_resolution_clock::now();
            while (isMessageStillProcessing(msgId)) {
                if (std::chrono::duration<double, std::milli>(std::chrono::high_resolution_clock::now() - t_start).count() > _timeOut) {
                    throw std::runtime_error("Time out waiting for a response of message id " + std::to_string(msgId));
                }
            }
        }
 
        template<typename T> int64_t sendWithSpecificHandler(const T & mb, std::shared_ptr<ETP_NS::ProtocolHandlers> specificHandler, int64_t correlationId = 0, int32_t messageFlags = 0)
        {
            // Encode the message into AVRO format
            auto queueItem = encode(mb, correlationId, messageFlags);
 
            const std::lock_guard<std::mutex> sendingQueueLock(sendingQueueMutex);
            // Set the handlers which are going to be called for the response to this sent message
            std::get<2>(queueItem) = specificHandler;
            // Push the message into the queue
            sendingQueue.push(queueItem);
            fesapi_log("*************************************************");
            fesapi_log("Message Header put in the queue : ");
            fesapi_log("protocol :", std::to_string(mb.protocolId));
            fesapi_log("type :" , std::to_string(mb.messageTypeId));
            fesapi_log("id :" , std::to_string(std::get<0>(queueItem)));
            fesapi_log("correlation id :" , std::to_string(correlationId));
            fesapi_log("flags :" , std::to_string(messageFlags));
            fesapi_log("Whole message size :" , std::to_string(std::get<1>(queueItem).size()) , "bytes.");
            fesapi_log("*************************************************");
 
            // Send the message directly if the sending queue was empty.
            if (sendingQueue.size() == 1) {
                do_write();
            }
 
            return std::get<0>(queueItem);
        }
 
        virtual void do_close() = 0;
 
        virtual void do_read() = 0;
 
        FETPAPI_DLL_IMPORT_OR_EXPORT void on_read(boost::system::error_code ec, std::size_t bytes_transferred);
 
        void on_write(boost::system::error_code ec, std::size_t) {
            if(ec) {
                std::cerr << "on_write : " << ec.message() << std::endl;
            }
 
            // Remove the sent message from the queue
            const std::lock_guard<std::mutex> sendingQueueLock(sendingQueueMutex);
            sendingQueue.pop();
 
            do_write();
        }
 
        void on_close(boost::system::error_code ec) {
            if(ec) {
                std::cerr << "on_close : " << ec.message() << std::endl;
            }
            if (!etpSessionClosed) {
                std::cerr << "Websocket session is going to be closed BUT ETP SESSION HAS NOT BEEN CLOSED YET!!!" << std::endl;
            }
 
            // If we get here then the connection is closed gracefully
            webSocketSessionClosed = true;
        }
 
        FETPAPI_DLL_IMPORT_OR_EXPORT bool isWebSocketSessionClosed() const { return webSocketSessionClosed;  }
 
        FETPAPI_DLL_IMPORT_OR_EXPORT bool isMessageStillProcessing(int64_t msgId) {
            const std::lock_guard<std::mutex> sendingQueueLock(sendingQueueMutex);
            const std::lock_guard<std::mutex> specificProtocolHandlersLock(specificProtocolHandlersMutex);
            return (!sendingQueue.empty() && std::get<0>(sendingQueue.front()) <= msgId) || specificProtocolHandlers.count(msgId) > 0;
        }
        //FETPAPI_DLL_IMPORT_OR_EXPORT bool isMessageStillProcessing(int64_t msgId) const { return specificProtocolHandlers.count(msgId) > 0; }
 
        virtual void setMaxWebSocketMessagePayloadSize(int64_t value) = 0;
        int64_t getMaxWebSocketMessagePayloadSize() const { return maxWebSocketMessagePayloadSize; }
 
        /****************
        ***** CORE ******
        ****************/
 
        FETPAPI_DLL_IMPORT_OR_EXPORT void close() {
            isCloseRequested = true;
            sendingQueueMutex.lock();
            specificProtocolHandlersMutex.lock();
            if (specificProtocolHandlers.empty() && sendingQueue.empty()) {
                etpSessionClosed = true;
                sendingQueueMutex.unlock();
                specificProtocolHandlersMutex.unlock();
                send(Energistics::Etp::v12::Protocol::Core::CloseSession(), 0, 0x02);
            }
            else {
                sendingQueueMutex.unlock();
                specificProtocolHandlersMutex.unlock();
            }
        }
 
        FETPAPI_DLL_IMPORT_OR_EXPORT void closeAndBlock() {
            close();
            auto t_start = std::chrono::high_resolution_clock::now();
            while (!isEtpSessionClosed()) {
                if (std::chrono::duration<double, std::milli>(std::chrono::high_resolution_clock::now() - t_start).count() > _timeOut) {
                    throw std::runtime_error("Time out waiting for closing");
                }
            }
        }
 
        FETPAPI_DLL_IMPORT_OR_EXPORT bool isEtpSessionClosed() const { return webSocketSessionClosed || etpSessionClosed; }
 
        void setEtpSessionClosed(bool etpSessionClosed_) { etpSessionClosed = etpSessionClosed_; }
 
        /****************
        *** DATASPACE ***
        ****************/
 
        FETPAPI_DLL_IMPORT_OR_EXPORT std::vector<Energistics::Etp::v12::Datatypes::Object::Dataspace> getDataspaces(int64_t storeLastWriteFilter = -1);
 
        FETPAPI_DLL_IMPORT_OR_EXPORT std::vector<std::string> putDataspaces(const std::map<std::string, Energistics::Etp::v12::Datatypes::Object::Dataspace>& dataspaces);
 
        FETPAPI_DLL_IMPORT_OR_EXPORT std::vector<std::string> deleteDataspaces(const std::map<std::string, std::string>& dataspaceUris);
 
        /*********************
        *** DATASPACE OSDU ***
        **********************/
 
        FETPAPI_DLL_IMPORT_OR_EXPORT std::vector<Energistics::Etp::v12::Datatypes::Object::Dataspace> getDataspaceInfo(const std::map<std::string, std::string>& dataspaceUris);
 
        FETPAPI_DLL_IMPORT_OR_EXPORT std::vector<std::string> copyDataspacesContent(const std::map<std::string, std::string>& sourceDataspaceUris, const std::string& targetDataspaceUri);
 
        FETPAPI_DLL_IMPORT_OR_EXPORT std::vector<std::string> lockDataspaces(const std::map<std::string, std::string>& dataspaceUris, bool lock);
 
        FETPAPI_DLL_IMPORT_OR_EXPORT std::vector<std::string> copyToDataspace(const std::map<std::string, std::string>& sourceUris, const std::string& targetDataspaceUri);
 
        /****************
        *** DISCOVERY ***
        ****************/
 
        FETPAPI_DLL_IMPORT_OR_EXPORT std::vector<Energistics::Etp::v12::Datatypes::Object::Resource> getResources(
            const Energistics::Etp::v12::Datatypes::Object::ContextInfo& context,
            const Energistics::Etp::v12::Datatypes::Object::ContextScopeKind& scope,
            int64_t storeLastWriteFilter = -1,
            bool countObjects = false);
 
        FETPAPI_DLL_IMPORT_OR_EXPORT std::vector<Energistics::Etp::v12::Datatypes::Object::DeletedResource> getDeletedResources(
            const std::string& dataspaceUri,
            int64_t deleteTimeFilter = -1,
            const std::vector<std::string>& dataObjectTypes = {});
 
        /****************
        ***** STORE *****
        ****************/
 
        FETPAPI_DLL_IMPORT_OR_EXPORT std::map<std::string, Energistics::Etp::v12::Datatypes::Object::DataObject> getDataObjects(const std::map<std::string, std::string>& uris);
 
        FETPAPI_DLL_IMPORT_OR_EXPORT std::vector<std::string> putDataObjects(const std::map<std::string, Energistics::Etp::v12::Datatypes::Object::DataObject>& dataObjects);
 
        FETPAPI_DLL_IMPORT_OR_EXPORT std::vector<std::string> deleteDataObjects(const std::map<std::string, std::string>& uris);
 
        /****************
        ** TRANSACTION **
        ****************/
 
        FETPAPI_DLL_IMPORT_OR_EXPORT std::string startTransaction(std::vector<std::string> dataspaceUris = {}, bool readOnly = false);
 
        FETPAPI_DLL_IMPORT_OR_EXPORT std::string rollbackTransaction();
 
        /*
        * A customer sends to a store to cancel a transaction. The store MUST disregard any requests or data sent 
        * with that transaction. The current transaction (the one being canceled) MUST NOT change the state of 
        * the store.
        * It actually sends a message and block the current thread untill a response has been received from the store.
        */
        FETPAPI_DLL_IMPORT_OR_EXPORT std::string commitTransaction();
 
        /***************************/
        // LOGGING
        /***************************/
 
        void setVerbose(bool verbose) {
            _verbose = verbose;
        }
 
        //terminating log
        void fesapi_log() { std::cout << std::endl; }
 
        template<typename First, typename ...Rest>
        void fesapi_log(First && first, Rest && ...rest)
        {
            if (_verbose) {
                std::cout << std::forward<First>(first) << " ";
                fesapi_log(std::forward<Rest>(rest)...);
            }
        }
 
    protected:
        boost::beast::flat_buffer receivedBuffer;
        std::unordered_map<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type, std::shared_ptr<ETP_NS::ProtocolHandlers>> protocolHandlers;
        std::unordered_map<int64_t, std::shared_ptr<ETP_NS::ProtocolHandlers>> specificProtocolHandlers;
        std::mutex specificProtocolHandlersMutex;
        int64_t maxWebSocketMessagePayloadSize{ 16000000 };
        std::atomic<bool> webSocketSessionClosed{ true };
        std::atomic<bool> etpSessionClosed{ true };
        std::atomic<double> _timeOut{ 30000 };
        std::atomic<bool> _verbose{ false };
        std::queue< std::tuple<int64_t, std::vector<uint8_t>, std::shared_ptr<ETP_NS::ProtocolHandlers>> > sendingQueue;
        std::mutex sendingQueueMutex;
        std::atomic<int64_t> messageId;
        boost::uuids::uuid identifier;
        bool isCloseRequested{ false };
 
        AbstractSession() = default;
 
        virtual boost::asio::io_context& getIoContext() = 0;
 
        void flushReceivingBuffer() {
            receivedBuffer.consume(receivedBuffer.size());
        }
 
        virtual void do_write() = 0;
 
        Energistics::Etp::v12::Datatypes::MessageHeader decodeMessageHeader(avro::DecoderPtr decoder);
 
        template<typename T> std::tuple<int64_t, std::vector<uint8_t>, std::shared_ptr<ETP_NS::ProtocolHandlers>> encode(const T & mb, int64_t correlationId = 0, int32_t messageFlags = 0)
        {
            // Build message header
            Energistics::Etp::v12::Datatypes::MessageHeader mh;
            mh.protocol = mb.protocolId;
            mh.messageType = mb.messageTypeId;
            mh.correlationId = correlationId;
            mh.messageId = messageId.fetch_add(2);
            mh.messageFlags = messageFlags;
 
            avro::OutputStreamPtr out = avro::memoryOutputStream();
            avro::EncoderPtr e = avro::binaryEncoder();
            e->init(*out);
            avro::encode(*e, mh);
            avro::encode(*e, mb);
            e->flush();
            const int64_t byteCount = e->byteCount();
 
            if (byteCount < maxWebSocketMessagePayloadSize) {
                return std::make_tuple(mh.messageId, *avro::snapshot(*out).get(), nullptr);
            }
            else {
                messageId -= 2;
                if (correlationId != 0) {
                    return encode(EtpHelpers::buildSingleMessageProtocolException(17, "I try to send you a too big message response of protocol "
                        + std::to_string(mb.protocolId) + " and type id " + std::to_string(mb.messageTypeId) + " and size " + std::to_string(byteCount)
                        + " bytes according to our negotiated size capability which is " + std::to_string(maxWebSocketMessagePayloadSize) + " bytes."), correlationId, 0x02);
                }
                else {
                    return std::make_tuple(-1, std::vector<uint8_t>{}, nullptr); // You cannot send a message which is too big. Please use message part or chunk or whatever else.
                }
            }
        }
 
        std::shared_ptr<ETP_NS::CoreHandlers> getCoreProtocolHandlers() {
            auto it = protocolHandlers.find(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::Core));
            return it == protocolHandlers.end()
                ? nullptr
                : std::dynamic_pointer_cast<CoreHandlers>(it->second);
        }
 
        std::shared_ptr<ETP_NS::DiscoveryHandlers> getDiscoveryProtocolHandlers() {
            auto it = protocolHandlers.find(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::Discovery));
            return it == protocolHandlers.end()
                ? nullptr
                : std::dynamic_pointer_cast<DiscoveryHandlers>(it->second);
        }
 
        std::shared_ptr<ETP_NS::StoreHandlers> getStoreProtocolHandlers() {
            auto it = protocolHandlers.find(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::Store));
            return it == protocolHandlers.end()
                ? nullptr
                : std::dynamic_pointer_cast<StoreHandlers>(it->second);
        }
 
        std::shared_ptr<ETP_NS::StoreNotificationHandlers> getStoreNotificationProtocolHandlers() {
            auto it = protocolHandlers.find(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::StoreNotification));
            return it == protocolHandlers.end()
                ? nullptr
                : std::dynamic_pointer_cast<StoreNotificationHandlers>(it->second);
        }
 
        std::shared_ptr<ETP_NS::DataArrayHandlers> getDataArrayProtocolHandlers() {
            auto it = protocolHandlers.find(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::DataArray));
            return it == protocolHandlers.end()
                ? nullptr
                : std::dynamic_pointer_cast<DataArrayHandlers>(it->second);
        }
 
        std::shared_ptr<ETP_NS::TransactionHandlers> getTransactionProtocolHandlers() {
            auto it = protocolHandlers.find(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::Transaction));
            return it == protocolHandlers.end()
                ? nullptr
                : std::dynamic_pointer_cast<TransactionHandlers>(it->second);
        }
 
        std::shared_ptr<ETP_NS::DataspaceHandlers> getDataspaceProtocolHandlers() {
            auto it = protocolHandlers.find(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::Dataspace));
            return it == protocolHandlers.end()
                ? nullptr
                : std::dynamic_pointer_cast<DataspaceHandlers>(it->second);
        }
 
        std::shared_ptr<ETP_NS::DataspaceOSDUHandlers> getDataspaceOSDUProtocolHandlers() {
            auto it = protocolHandlers.find(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::DataspaceOSDU));
            return it == protocolHandlers.end()
                ? nullptr
                : std::dynamic_pointer_cast<DataspaceOSDUHandlers>(it->second);
        }
 
        void setProtocolHandlers(std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type protocolId, std::shared_ptr<ProtocolHandlers> coreHandlers) {
            // Verify that we don't modify handlers which could be in use on the io context thread
            if (!isWebSocketSessionClosed()) {
                throw std::logic_error("You cannot set some protocol handlers once the session is running.");
            }
 
            protocolHandlers[protocolId] = coreHandlers;
        }
    };
}