data_publisher.hpp

#pragma once
 
#include <atomic>
#include <optional>
#include <thread>
 
#include "spdlog/spdlog.h"
#include "zmq.hpp"
 
#include "synapse-app-sdk/middleware/conversions.hpp"
 
namespace synapse {

enum class PublisherType : uint8_t {
  kUnknown = 0, 
  ZMQ_IPC = 1,  
  ZMQ_TCP = 2,  
  UDP = 3       
};

class DataPublisher {
 public:
  DataPublisher(const PublisherType& publisher_type, zmq::context_t& zmq_context)
      : publisher_type_(publisher_type),
        zmq_context_(zmq_context),
        publisher_socket_(zmq_context_, zmq::socket_type::pub) {}

  ~DataPublisher() {
    if (connected_) {
      publisher_socket_.close();
    }
  }

  bool setup(const std::string& endpoint) {
    try {
      publisher_socket_.bind(endpoint);
      connected_ = true;
      return true;
    } catch (const std::exception& e) {
      spdlog::error("Failed to bind data publisher to: {}, why: {}", endpoint, e.what());
      return false;
    }
  }

  template <typename T>
  bool try_publish(const T& message) {
    if (!connected_) {
      spdlog::warn("Could not publish, did you connect?");
      return false;
    }
    // Send the topic data first
    const std::string kControllerZMQTopic = "controller/output";
    zmq::message_t topic_message(kControllerZMQTopic.begin(), kControllerZMQTopic.end());
    const auto topic_send_ret = publisher_socket_.send(topic_message, zmq::send_flags::sndmore);
    if (!topic_send_ret.has_value() || topic_send_ret.value() == 0) {
      spdlog::error("Failed to publish topic data for controller node");
      return false;
    }
 
    // TODO: We will need to abstract this out for different publishers
    zmq::message_t out_message = protobuf_to_zmq_message(message);
    if (out_message.size() == 0) {
      spdlog::error("Failed to convert protobuf to zmq message for sending");
      return false;
    }
 
    auto result = publisher_socket_.send(out_message, zmq::send_flags::dontwait);
    return result.has_value();
  }
 
 private:
  PublisherType publisher_type_;

  std::atomic<bool> connected_{false};

  zmq::context_t& zmq_context_;

  zmq::socket_t publisher_socket_;
};
}  // namespace synapse