ZMQMessageFactory.h

/**************************************************************************
 * basf2 (Belle II Analysis Software Framework)                           *
 * Author: The Belle II Collaboration                                     *
 *                                                                        *
 * See git log for contributors and copyright holders.                    *
 * This file is licensed under LGPL-3.0, see LICENSE.md.                  *
 **************************************************************************/
#pragma once
 
#include <framework/logging/Logger.h>
#include <framework/pcore/zmq/messages/ZMQNoIdMessage.h>
#include <framework/pcore/zmq/messages/ZMQIdMessage.h>
#include <framework/pcore/zmq/messages/ZMQDefinitions.h>
#include <memory>
#include <string>
#include <zmq.hpp>
 
namespace Belle2 {
  class ZMQMessageFactory {
  public:
    static auto createMessage(const std::string& msgIdentity,
                              const EMessageTypes msgType,
                              const std::unique_ptr<EvtMessage>& eventMessage)
    {
      return createMessage<ZMQIdMessage>(msgIdentity, msgType, eventMessage);
    }
 
    static auto createMessage(const std::string& msgIdentity,
                              const EMessageTypes msgType,
                              const std::string& msgData = "")
    {
      return createMessage<ZMQIdMessage>(msgIdentity, msgType, msgData);
    }
 
    static auto createMessage(const std::string& msgIdentity,
                              const EMessageTypes msgType,
                              zmq::message_t msgData)
    {
      return createMessage<ZMQIdMessage>(msgIdentity, msgType, std::move(msgData));
    }
 
    static auto createMessage(const std::string& msgIdentity,
                              std::unique_ptr<ZMQNoIdMessage> shortMessage)
    {
      return createMessage<ZMQIdMessage>(msgIdentity, std::move(shortMessage->getMessagePart<0>()),
                                         std::move(shortMessage->getMessagePart<1>()));
    }
 
    static auto createMessage(const EMessageTypes msgType,
                              const std::string& msgData = "")
    {
      return createMessage<ZMQNoIdMessage>(msgType, msgData);
    }
 
    static auto createMessage(const EMessageTypes msgType,
                              int msgData)
    {
      return createMessage<ZMQNoIdMessage>(msgType, msgData);
    }
 
    static auto createMessage(const EMessageTypes msgType,
                              const StoreObjPtr<EventMetaData>& evtMetaData)
    {
      return createMessage<ZMQNoIdMessage>(msgType, evtMetaData);
    }
 
    static auto createMessage(const EMessageTypes msgType,
                              zmq::message_t msgData)
    {
      return createMessage<ZMQNoIdMessage>(msgType, std::move(msgData));
    }
 
    static auto createMessage(const EMessageTypes msgType,
                              zmq::message_t msgData,
                              zmq::message_t additionalData)
    {
      return createMessage<ZMQNoIdMessage>(msgType, std::move(msgData), std::move(additionalData));
    }
 
    static auto createMessage(zmq::message_t msgType,
                              zmq::message_t msgData)
    {
      return createMessage<ZMQNoIdMessage>(std::move(msgType), std::move(msgData));
    }
 
    static auto createMessage(const EMessageTypes msgType,
                              const std::unique_ptr<EvtMessage>& eventMessage)
    {
      return createMessage<ZMQNoIdMessage>(msgType, eventMessage);
    }
 
    static auto createMessage(const EMessageTypes msgType,
                              const std::unique_ptr<EvtMessage>& eventMessage,
                              zmq::message_t additionalData)
    {
      return createMessage<ZMQNoIdMessage>(msgType, eventMessage, std::move(additionalData));
    }
 
    static auto stripIdentity(std::unique_ptr<ZMQIdMessage> message)
    {
      return createMessage<ZMQNoIdMessage>(std::move(message->getMessagePart<1>()),
                                           std::move(message->getMessagePart<2>()),
                                           std::move(message->getMessagePart<3>()));
    }
 
    template <class AMessage>
    static std::unique_ptr<AMessage> fromSocket(const std::unique_ptr<zmq::socket_t>& socket)
    {
      auto newMessage = std::unique_ptr<AMessage>(new AMessage());
      auto& messageParts = newMessage->getMessageParts();
      for (unsigned int i = 0; i < AMessage::c_messageParts; i++) {
        B2ASSERT("The next part does not belong to the same message",
                 socket->get(zmq::sockopt::rcvmore) == 1 or i == 0);
        auto received = socket->recv(messageParts[i], zmq::recv_flags::none);
        B2ASSERT("No message received", received);
      }
      B2ASSERT("There should not be more than the retrieved parts", socket->get(zmq::sockopt::rcvmore) == 0);
      return newMessage;
    }
 
  private:
    template <class T, class... Args>
    static std::unique_ptr<T> createMessage(Args&& ... args)
    {
      return std::unique_ptr<T>(new T(std::forward<Args>(args)...));
    }
  };
}