proto_server.h

Go to the documentation of this file.

#pragma once
 
#include <uwebsockets/App.h>
#include <atomic>
#include <thread>
 
#include "incoming_machine_message.h"
#include "outgoing_machine_message.h"
 
#include "core/containers/thread_safe_queue.h"

class ProtoServer {
 public:
  struct ConnectionData {
    std::string id;
  };

  explicit ProtoServer(
      mm::core::containers::ThreadSafeQueue<IncomingMachineMessage>&
          incomingMachineMessageQueue,
      mm::core::containers::ThreadSafeQueue<OutgoingMachineMessage>&
          outgoingMachineMessageQueue,
      uint16_t reqResPort = 63524, uint16_t pubSubPort = 63525);

  ~ProtoServer();

  void start();

  void stop();
 
 private:
  void handleUpgrade(uWS::HttpResponse<false>* res, uWS::HttpRequest* req,
                     us_socket_context_t* context,
                     bool checkDuplicateClientIds = false);

  void runReqRes();

  void runPubSub();

  std::atomic<bool> running_ = false;

  uint16_t reqResPort_ = 63524;
  std::thread reqResThread_;
  uWS::Loop* reqResLoop_ = nullptr;
  us_listen_socket_t* reqResListenToken_ = nullptr;
  std::unordered_map<std::string,
                     uWS::WebSocket<false, true, ProtoServer::ConnectionData>*>
      reqResConnectionIdToWebSocketMap_;

  uint16_t pubSubPort_ = 63525;
  std::thread pubSubThread_;
  uWS::Loop* pubSubLoop_ = nullptr;
  us_listen_socket_t* pubSubListenToken_ = nullptr;
  std::unordered_map<std::string,
                     uWS::WebSocket<false, true, ProtoServer::ConnectionData>*>
      pubSubConnectionIdToWebSocketMap_;

  mm::core::containers::ThreadSafeQueue<IncomingMachineMessage>&
      incomingMachineMessageQueue_;
  mm::core::containers::ThreadSafeQueue<OutgoingMachineMessage>&
      outgoingMachineMessageQueue_;

  std::thread serverThread_;

  void runServer();

  void sendOutgoingMachineMessages();

  std::unordered_map<std::string,
                     std::tuple<std::string, std::set<std::string>>>
      monitoringSubscriptionMap_;
};