ZMQTxInputModule.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/core/Module.h>
#include <framework/datastore/StoreObjPtr.h>
#include <framework/dataobjects/EventMetaData.h>
#include <framework/core/RandomGenerator.h>
#include <framework/pcore/zmq/sockets/ZMQClient.h>
#include <framework/pcore/zmq/utils/StreamHelper.h>
 
#include <framework/pcore/zmq/processModules/ProcessedEventsBackupList.h>
#include <deque>
 
namespace Belle2 {
  class ZMQTxInputModule : public Module {
  public:
    ZMQTxInputModule();
    void event() override;
    void terminate() override;
    void initialize() override;
    //    void beginRun() override;
    void endRun() override;
 
  private:
    std::deque<unsigned int> m_nextWorker;
    std::vector<unsigned int> m_workers;
    ProcessedEventsBackupList m_procEvtBackupList;
 
    unsigned int m_param_maximalWaitingTime;
    unsigned int m_param_workerProcessTimeout;
    bool m_param_useEventBackup;
 
    bool m_firstEvent = true;
 
    std::string m_param_socketName;
    std::string m_param_xpubProxySocketName;
    std::string m_param_xsubProxySocketName;
    int m_param_compressionLevel = 0;
    bool m_param_handleMergeable = true;
 
    ZMQClient m_zmqClient;
    StreamHelper m_streamer;
 
    StoreObjPtr<EventMetaData> m_eventMetaData;
    StoreObjPtr<RandomGenerator> m_randomgenerator;
 
    void checkWorkerProcTimeout();
  };
}