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 "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/CoreOSDUHandlers.h"
#include "protocolHandlers/StoreOSDUHandlers.h"
#include "protocolHandlers/DataspaceOSDUHandlers.h"
 
namespace beast = boost::beast;         // from <boost/beast.hpp>
namespace websocket = beast::websocket;  // from <boost/beast/websocket.hpp>
using tcp = boost::asio::ip::tcp;               // from <boost/asio/ip/tcp.hpp>
 
namespace ETP_NS
{
    class AbstractSession : public std::enable_shared_from_this<AbstractSession>
    {
    public:
 
        virtual ~AbstractSession() = default;

        const boost::uuids::uuid& getIdentifier() {
            std::lock_guard<std::mutex> lock(identifierMutex);
            return identifier;
        }

        void setTimeOut(double timeOut) {
            _timeOut = timeOut;
        }

        double getTimeOut() const {
            return _timeOut;
        }

        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 setCoreOSDUProtocolHandlers(std::shared_ptr<CoreOSDUHandlers> coreOSDUHandlers) {
            setProtocolHandlers(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::CoreOSDU), coreOSDUHandlers);
        }

        FETPAPI_DLL_IMPORT_OR_EXPORT void setStoreOSDUProtocolHandlers(std::shared_ptr<StoreOSDUHandlers> storeOSDUHandlers) {
            setProtocolHandlers(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::StoreOSDU), storeOSDUHandlers);
        }

        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);
        }

        int64_t send(std::shared_ptr<EtpMessage> message,
            int64_t correlationId = 0, int32_t messageFlags = 0, const std::vector<int64_t>& msgIdAliases = {})
        {
            return sendWithSpecificHandler(message, protocolHandlers.at(message->messageHeader.protocol), correlationId, messageFlags, msgIdAliases);
        }

        int64_t sendAndBlock(std::shared_ptr<EtpMessage> message,
                int64_t correlationId = 0, int32_t messageFlags = 0, const std::vector<int64_t>& msgIdAliases = {})
        {
            int64_t msgId = send(message, correlationId, messageFlags, msgIdAliases);
            // The correlationId of the first message MUST be set to 0 and the correlationId of all successive
            // messages in the same multipart request or notification MUST be set to the messageId of the first
            // message of the multipart request or notification.
            // If the request message is itself multipart, the correlationId of each message of the multipart
            // response MUST be set to the messageId of the FIRST message in the multipart request.
 
            auto t_start = std::chrono::high_resolution_clock::now();
            while (isMessageStillProcessing(correlationId == 0 ? msgId : correlationId)) {
                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));
                }
            }
 
            return msgId;
        }

        int64_t sendWithSpecificHandler(std::shared_ptr<EtpMessage> message, std::shared_ptr<ETP_NS::ProtocolHandlers> specificHandler,
            int64_t correlationId = 0, int32_t messageFlags = 0, const std::vector<int64_t>& msgIdAliases = {})
        {
            message->messageHeader.correlationId = correlationId;
            message->messageHeader.messageFlags = messageFlags;
            message->messageHeader.messageId = messageId.fetch_add(2);
 
            const std::lock_guard<std::mutex> sendingQueueLock(sendingQueueMutex);
 
            // Push the message into the queue
            sendingQueue.push(std::make_tuple(message, specificHandler, msgIdAliases));
            fesapi_log(message->to_string());
            
            if (message->messageHeader.protocol == static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::CoreOSDU)
                && message->messageHeader.messageType == Energistics::Etp::v12::Protocol::CoreOSDU::ResumeSession::messageTypeId) {
                // Handle previous messages which have not been processed but written in the previous sendingQueue.
                std::queue< std::tuple<std::shared_ptr<EtpMessage>, std::shared_ptr<ETP_NS::ProtocolHandlers>, std::vector<int64_t>> > queueWithPreviousInformation;
                // Put ResumeSession at first position whatever opens
                queueWithPreviousInformation.push(std::make_tuple(message, specificHandler, msgIdAliases));
                // Reorder message according to their id
                int64_t previouslyDuplicatedId = -1;
                for (size_t i = 0; i < specificProtocolHandlers.size(); ++i) {
                    int64_t lowestMsgId = (std::numeric_limits<int64_t>::max)();
                    for (auto keyVal : specificProtocolHandlers) {
                        if (keyVal.first > previouslyDuplicatedId && keyVal.first < lowestMsgId) {
                            lowestMsgId = keyVal.first;
                        }
                    }
                    previouslyDuplicatedId = lowestMsgId;
                    auto val = specificProtocolHandlers[lowestMsgId];
                    std::get<0>(val)->messageHeader.messageId = messageId.fetch_add(2);
                    std::get<2>(val).push_back(lowestMsgId);
                    queueWithPreviousInformation.push(val);
                    fesapi_log(std::get<0>(val)->to_string());
                }
                // The only remaining item must be the resumeSession which is already at the first position of queueWithPreviousInformation
                while (sendingQueue.size() != 1) {
                    std::get<0>(sendingQueue.front())->messageHeader.messageId = messageId.fetch_add(2);
                    queueWithPreviousInformation.push(sendingQueue.front());
                    fesapi_log(std::get<0>(sendingQueue.front())->to_string());
                    sendingQueue.pop();
                }
                if (queueWithPreviousInformation.size() > 1) {
                    std::swap(sendingQueue, queueWithPreviousInformation);
                }
                do_write();
            }
            else if (sendingQueue.size() == 1) {
                // Send the message directly if the sending queue was empty.
                do_write();
            }
 
            return message->messageHeader.messageId;
        }

        int64_t sendWithSpecificHandlerAndBlock(std::shared_ptr<EtpMessage> message, std::shared_ptr<ETP_NS::ProtocolHandlers> specificHandler,
            int64_t correlationId = 0, int32_t messageFlags = 0, const std::vector<int64_t>& msgIdAliases = {})
        {
            const int64_t msgId = sendWithSpecificHandler(message, specificHandler, correlationId, messageFlags, msgIdAliases);
            // The correlationId of the first message MUST be set to 0 and the correlationId of all successive
            // messages in the same multipart request or notification MUST be set to the messageId of the first
            // message of the multipart request or notification.
            // If the request message is itself multipart, the correlationId of each message of the multipart
            // response MUST be set to the messageId of the FIRST message in the multipart request.
 
            const auto t_start = std::chrono::high_resolution_clock::now();
            while (isMessageStillProcessing(correlationId == 0 ? msgId : correlationId)) {
                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));
                }
            }
 
            // If the message has not been answered correctly
            if (isEtpSessionClosed() && !isCloseRequested()) {
                // Wait for a reconnection
                while (isEtpSessionClosed() && !isCloseRequested()) {}
                // Check if reconnection is successfull
                if (isEtpSessionClosed()) {
                    throw std::runtime_error("The ETP session could not be opened in order to send again the message.");
                }
                else {
                    return sendWithSpecificHandlerAndBlock(message, specificHandler, correlationId, messageFlags);
                }
            }
 
            return msgId;
        }

        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_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 isCloseRequested() const { return isCloseRequested_; }

        FETPAPI_DLL_IMPORT_OR_EXPORT bool isWebSocketSessionClosed() const { return webSocketSessionClosed;  }

        FETPAPI_DLL_IMPORT_OR_EXPORT bool isMessageStillProcessing(int64_t msgId) {
            std::scoped_lock lock(sendingQueueMutex, specificProtocolHandlersMutex);
            return (!sendingQueue.empty() && std::get<0>(sendingQueue.front())->messageHeader.messageId <= msgId) || specificProtocolHandlers.count(msgId) > 0;
        }
 
        virtual void setMaxWebSocketMessagePayloadSize(uint64_t value) = 0;
        uint64_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(std::make_shared<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 (!webSocketSessionClosed) {
                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; }
 
        /****************
        *** 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> lockDataspaces(const std::map<std::string, std::string>& dataspaceUris, bool lock);

        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> copyToDataspace(const std::map<std::string, std::string>& sourceDataobjectUris, 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);
 
        /*********************
        ***** STORE OSDU *****
        **********************/

        FETPAPI_DLL_IMPORT_OR_EXPORT std::vector<std::string> copyDataObjectsByValue(const std::string& sourceDataobjectUri, int32_t sourcesDepth = 0, const std::vector<std::string>& dataObjectTypes = {});
 
        /****************
        ** 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 blocks the current thread until a response has been received from the store.
        * @return Failure message or empty string if success
        */
        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:
        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::tuple<std::shared_ptr<EtpMessage>, std::shared_ptr<ETP_NS::ProtocolHandlers>, std::vector<int64_t>>> specificProtocolHandlers;
        std::mutex specificProtocolHandlersMutex;
        uint64_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<std::shared_ptr<EtpMessage>, std::shared_ptr<ETP_NS::ProtocolHandlers>, std::vector<int64_t> > > sendingQueue;
        std::mutex sendingQueueMutex;
        std::atomic<int64_t> messageId;
        boost::uuids::uuid identifier{ boost::uuids::nil_uuid() };
        std::mutex identifierMutex;
        bool isCloseRequested_{ false };
        size_t reconnectionTryCount_ = 0;
 
        AbstractSession() = default;
 
        virtual boost::asio::io_context& getIoContext() = 0;
 
        void flushReceivingBuffer() {
            receivedBuffer.consume(receivedBuffer.size());
        }
 
        void setEtpSessionClosed(bool etpSessionClosed_) {
            etpSessionClosed = etpSessionClosed_;
            reconnectionTryCount_ = 0;
        }

        virtual void do_write() = 0;

        Energistics::Etp::v12::Datatypes::MessageHeader decodeMessageHeader(avro::DecoderPtr decoder);
 
        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::CoreOSDUHandlers> getCoreOSDUProtocolHandlers() {
            auto it = protocolHandlers.find(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::CoreOSDU));
            return it == protocolHandlers.end()
                ? nullptr
                : std::dynamic_pointer_cast<CoreOSDUHandlers>(it->second);
        }
 
        std::shared_ptr<ETP_NS::StoreOSDUHandlers> getStoreOSDUProtocolHandlers() {
            auto it = protocolHandlers.find(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::StoreOSDU));
            return it == protocolHandlers.end()
                ? nullptr
                : std::dynamic_pointer_cast<StoreOSDUHandlers>(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;
        }
 
        friend void CoreHandlers::decodeMessageBody(const Energistics::Etp::v12::Datatypes::MessageHeader& mh, avro::DecoderPtr d);
        friend void CoreHandlers::on_ProtocolException(const Energistics::Etp::v12::Protocol::Core::ProtocolException& msg, int64_t correlationId);
        friend void CoreOSDUHandlers::decodeMessageBody(const Energistics::Etp::v12::Datatypes::MessageHeader& mh, avro::DecoderPtr d);
    };
}