cm_reqrep.h

/*
 * Developer : Alexey Zakharov (alexey.zakharov@vectioneer.com)
 * All rights reserved. Copyright (c) 2019 VECTIONEER.
 */
 
#ifndef CM_REQREP_H
#define CM_REQREP_H
 
#include "cm_basereqrep.h"
#include "cm_rpcmotorcortex.h"
#include "cm_groupmanager.h"
#include "cm_paramtree.h"
#include "cm_conndata.h"
#include "utl_realtime.h"
#include "utl_buffer.h"
 
#include <string>
#include <vector>
#include <nng/nng.h>
#include <nng/supplemental/tls/tls.h>
 
namespace mcx {
 
namespace parameter_server {
class Parameter;
}
 
namespace comm {
 
class AuthorizationBase;
 
struct PipeEvent {
  nng_pipe_ev event;
  nng_pipe pipe;
};
 
using RPCProcessorMap = std::map<uint32_t, RPCProcessor*>;
 
// The server keeps a list of work items, sorted by expiration time,
// so that we can use this to set the timeout to the correct value for
// use in poll.
struct Work {
  enum {
    INIT, RECV, SEND, FINISHED
  } state;
  nng_aio* aio;
  nng_msg* msg;
  nng_ctx ctx;
  utils::MpmcQueue<PipeEvent>* pipe_event_queue;
  AuthorizationBase* authorization_ptr;
  RPCProcessorMap* rpc_processor_map;
  GroupManager* group_manager;
  utils::PageFaults page_faults;
};
 
class RequestReply : public BaseRequestReply {
 
  static constexpr auto SOCKET_EVENT_QUEUE = 128;
 
public:
  explicit RequestReply(ProtocolType protocol_type = ProtocolType::SINGLE_TIMESTAMP);
 
  RequestReply(const RequestReply& orig) = delete;
 
  ~RequestReply() override;
 
  bool addRPCProcessor(const char* name, RPCProcessor* rpc_processor, bool force = false);
 
  bool configure(parameter_server::Parameter* root, const std::string& path);
 
  bool start(const ConnectionData& conn_data);
 
  void iterate(unsigned int timeout_micro_s = 50);
 
  void stop();
 
private:
 
  static void pipeEvent(nng_pipe pipe, nng_pipe_ev pipe_ev, void* ptr);
 
  // needed for older g++ in yocto
  static void pipeEvent(nng_pipe pipe, int pipe_ev, void* ptr);
 
  static bool process(Work* work);
 
  static void serverCb(void* arg);
 
  static struct Work* allocWork(nng_socket sock);
 
 
#ifdef __clang__
#pragma clang diagnostic push
#pragma clang diagnostic ignored "-Wunused-private-field"
#endif
  struct utils::PageFaults page_faults_{};
#ifdef __clang__
#pragma clang diagnostic pop
#endif
 
  nng_socket sock_{};
  union {
    nng_listener listener;
    nng_dialer dialer;
  } endpoint_;
  bool is_listener_{};
  nng_tls_config* tls_cfg_{nullptr};
  utils::MpmcQueue<PipeEvent> pipe_event_queue_{SOCKET_EVENT_QUEUE};
  std::vector<struct Work*> works_;
  std::map<uint32_t, RPCProcessor*> rpc_processor_map_;
 
  AuthorizationBase* authorization_ptr_{};
  ParameterTree parameter_tree_;
  RPCMotorcortex rpc_motorcortex_;
 
};
 
}  // namespace comm
 
}  // namespace mcx
 
#endif /*CM_REQREP_H*/