TcpCore.h

Go to the documentation of this file.

 
#ifndef __ROGUE_INTERFACES_STREAM_TCP_CORE_H__
#define __ROGUE_INTERFACES_STREAM_TCP_CORE_H__
#include "rogue/Directives.h"
 
#include <stdint.h>
 
#include <atomic>
#include <memory>
#include <string>
#include <thread>
 
#include "rogue/Logging.h"
#include "rogue/interfaces/stream/Frame.h"
#include "rogue/interfaces/stream/Master.h"
#include "rogue/interfaces/stream/Slave.h"
 
namespace rogue {
namespace interfaces {
namespace stream {
 
class TcpCore : public rogue::interfaces::stream::Master, public rogue::interfaces::stream::Slave {
  protected:
    // Inbound Address
    std::string pullAddr_;
 
    // Outbound Address
    std::string pushAddr_;
 
    void* zmqCtx_  = nullptr;
    void* zmqPull_ = nullptr;
    void* zmqPush_ = nullptr;
 
    bool server_ = false;
 
    // Thread background
    void runThread();
 
    // Rebuild zmqPush_ after a failed multipart send.  Caller must hold bridgeMtx_.
    bool rebuildPushSocket();
 
    // Log
    std::shared_ptr<rogue::Logging> bridgeLog_;
 
    std::unique_ptr<std::thread> thread_;
    std::atomic<bool> threadEn_{false};
 
    // Lock
    std::mutex bridgeMtx_;
 
  public:
    static std::shared_ptr<rogue::interfaces::stream::TcpCore> create(const std::string& addr, uint16_t port, bool server);
 
    static void setup_python();
 
    TcpCore(const std::string& addr, uint16_t port, bool server);
 
    ~TcpCore();
 
    [[deprecated("Use stop() instead")]] void close();
 
    void stop();
 
    void acceptFrame(std::shared_ptr<rogue::interfaces::stream::Frame> frame);
};
 
typedef std::shared_ptr<rogue::interfaces::stream::TcpCore> TcpCorePtr;
 
}  // namespace stream
}  // namespace interfaces
};  // namespace rogue
 
#endif