app.hpp

#pragma once
 
#include "synapse-app-sdk/middleware/data_publisher.hpp"
#include "synapse-app-sdk/middleware/data_reader.hpp"
#include "synapse-app-sdk/middleware/taps.hpp"
#include "synapse-app-sdk/utils/performance/function_profiler.hpp"
#include "synapse-app-sdk/utils/time/time.hpp"
#include "synapse-app-sdk/utils/config/config.hpp"
#include "synapse-app-sdk/utils/logging/synapse_log_sink.hpp"
#include <atomic>
#include <csignal>
#include <filesystem>
#include <fstream>
#include <iostream>
#include <memory>
 
#include "api/synapse.pb.h"
 
namespace synapse {
const std::string kDefaultDeviceConfigPath = "/opt/scifi/config/device.json";

class App {
 public:
  App();

  explicit App(const std::string& configuration_path);

  virtual ~App() = default;

  template <typename T>
  bool create_tap(const std::string& name) {
    static_assert(std::is_base_of<google::protobuf::Message, T>::value,
                  "T must be a protobuf message type");
    // Get the type name before creating the tap
    try {
      const google::protobuf::Descriptor* descriptor = T::descriptor();
      if (descriptor == nullptr) {
        spdlog::warn("Nullptr while creating tap descriptor: {}", name);
        return false;
      }
      const std::string type_name = descriptor->full_name();
      spdlog::info("Creating tap: {}, type: {}", name, type_name);
      return tap_manager_.add_tap(name, type_name).has_value();
    } catch (const std::exception& e) {
      spdlog::warn("Failed to create tap with name: {}, why: {}", name, e.what());
      return false;
    }
  }

  template <typename T>
  bool publish_tap(const std::string& name, const T& message) {
    static_assert(std::is_base_of<google::protobuf::Message, T>::value,
                  "T must be a protobuf message type");
    return tap_manager_.send_message(name, message);
  }

  template <typename T>
  bool create_consumer_tap(const std::string& name,
                           const std::function<void(const T& message)>& callback) {
    static_assert(std::is_base_of<google::protobuf::Message, T>::value,
                  "T must be a protobuf message type");
 
    auto callback_wrapper = [callback, name](zmq::message_t message) {
      auto message_proto = parse_protobuf_message<T>(std::move(message));
      if (message_proto.has_value() && callback) {
        callback(message_proto.value());
      } else {
        spdlog::warn("Failed to parse protobuf message for consumer tap: {}", name);
      }
    };
    return tap_manager_.add_consumer_tap(name, T::descriptor()->full_name(), callback_wrapper)
        .has_value();
  }

  virtual bool setup() = 0;

  virtual bool start();

  virtual void stop();
 
 protected:
  std::atomic<bool> node_running_{false};

  zmq::context_t zmq_context_;

  std::shared_ptr<synapse::ZMQDataReader> data_reader_{nullptr};

  std::shared_ptr<synapse::DataPublisher> data_publisher_{nullptr};

  TapManager tap_manager_;

  std::shared_ptr<synapse::log::SynapseLogSink<std::mutex>> log_sink_{nullptr};

  std::optional<uint64_t> log_subscription_id_{std::nullopt};

  virtual bool setup_reader(const uint32_t node_id);

  virtual bool setup_publisher(const PublisherType& publisher_type,
                               const std::string& publisher_endpoint);

  synapse::DeviceConfiguration device_configuration_;

  bool load_device_configuration(const std::string& path);

  synapse::FunctionProfilerManager function_profiler_manager_;

  std::thread function_profiling_thread_;

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

  bool enable_function_profiling(const std::chrono::milliseconds& period);

  void start_profile(const std::string& name);

  void stop_profile(const std::string& name);

  void print_profile(const std::string& name);

  bool get_app_config(std::function<bool(const synapse::ApplicationNodeConfig&)> validator,
                      synapse::ApplicationNodeConfig& output_config) {
    // First, make sure we have a node of the correct type
    const auto app_node_configs =
        get_nodes_for_type(device_configuration_, synapse::NodeType::kApplication);
    if (app_node_configs.size() != 1) {
      spdlog::error("Expected exactly one application node, got {}", app_node_configs.size());
      spdlog::error("Did you forget to add the application node to the device configuration?");
      return false;
    }
    const auto& node_config = app_node_configs[0];
 
    if (!node_config.has_application()) {
      spdlog::error("Configuration does not have application parameters");
      return false;
    }
    const auto& app_node_config = node_config.application();
    if (!validator(app_node_config)) {
      spdlog::error("Application node configuration is not valid");
      return false;
    }
 
    output_config.CopyFrom(app_node_config);
    spdlog::info("Application node configuration: {}", output_config.DebugString());
 
    return true;
  }
 
  void disable_log_streaming() { should_stream_logs_ = false; }

  std::thread main_loop_thread_;

  virtual void main() = 0;
 
 private:
  void setup_logging();

  bool setup_log_streaming();

  std::atomic<bool> should_stream_logs_{true};
};

class AppRunner {
 public:
  static AppRunner& instance() {
    static AppRunner instance;
    return instance;
  }

  AppRunner(const AppRunner&) = delete;

  AppRunner& operator=(const AppRunner&) = delete;

  AppRunner(AppRunner&&) = delete;

  AppRunner& operator=(AppRunner&&) = delete;

  void stop() noexcept {
    if (app_) {
      app_->stop();
    }
    running_ = false;
  }

  template <typename T>
  int run() {
    std::signal(SIGINT, AppRunner::signal_handler);
 
    app_ = std::make_unique<T>();
    if (!app_->setup()) {
      spdlog::error("Failed to setup Synapse application");
      return 1;
    }
 
    if (!app_->start()) {
      spdlog::error("Failed to start Synapse application");
      return 1;
    }
 
    running_ = true;
    while (running_) {
      if (signal_received_) {
        stop();
      }
      std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
 
    return 0;
  }

  void handle_signal(int signal) { signal_received_ = true; }
 
 private:
  AppRunner() = default;

  ~AppRunner() = default;

  static void signal_handler(int signal) { AppRunner::instance().handle_signal(signal); }

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

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

  std::unique_ptr<App> app_ = nullptr;
};

template <typename T>
int Entrypoint() {
  return AppRunner::instance().run<T>();
}
 
}  // namespace synapse