proto_server.h

Go to the documentation of this file.

#pragma once
 
#include <uwebsockets/App.h>
#include <atomic>
#include <mutex>
#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();

  void clearMonitoringSubscriptions();
 
 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_;
  std::atomic<uWS::Loop*> reqResLoop_ = nullptr;
  std::atomic<us_listen_socket_t*> reqResListenToken_ = nullptr;
  std::unordered_map<std::string,
                     uWS::WebSocket<false, true, ProtoServer::ConnectionData>*>
      reqResConnectionIdToWebSocketMap_;
  std::mutex reqResConnectionIdToWebSocketMapMutex_;

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

  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_;
  std::mutex monitoringSubscriptionMapMutex_;
};