Line data    Source code

       1             : #include "felix-server/Receiver.hpp"
       2             : 
       3             : #include <algorithm>
       4             : 
       5             : #include <netio3-backend/Netio3Backend.hpp>
       6             : #include <netio3/NetioPublisher.hpp>
       7             : 
       8             : #include "felix-server/Exceptions.hpp"
       9             : #include "felix-server/Utility.hpp"
      10             : 
      11           8 : felix_server::Receiver::Receiver(Settings settings,
      12             :                                  std::shared_ptr<netio3::BaseEventLoop> evloop,
      13           8 :                                  std::span<const std::uint64_t> tags) :
      14          16 :   m_settings{std::move(settings)},
      15           8 :   m_bus_writer{
      16           8 :     m_settings.bus_path,
      17           8 :     m_settings.bus_group,
      18             :   },
      19           8 :   m_receiver{{
      20           8 :                .backend_type = m_settings.network_type,
      21           8 :                .backend_mode = utility::get_network_mode(m_settings.network_type),
      22             :              },
      23           8 :              std::move(evloop)}
      24             : {
      25           8 :   m_receiver.set_on_buffer_cb(m_settings.on_buffer);
      26           8 :   m_receiver.set_on_data_cb(m_settings.on_msg);
      27           8 :   m_port = m_receiver.listen(
      28             :     {
      29           8 :       m_settings.ip,
      30           8 :       m_settings.port,
      31             :     },
      32             :     {
      33           8 :       m_settings.buffer_size,
      34           8 :       m_settings.buffer_count,
      35             :     });
      36             :   try {
      37           7 :     declare(tags);
      38           2 :   } catch (const std::exception& e) {
      39           2 :     throw BusError(e);
      40           2 :   }
      41          14 : }
      42             : 
      43           7 : void felix_server::Receiver::declare(std::span<const std::uint64_t> tags)
      44             : {
      45           7 :   auto info = std::vector<felixbus::FelixBusInfo>{};
      46           7 :   info.reserve(tags.size());
      47           7 :   std::ranges::transform(tags, std::back_inserter(info), [this](const auto& tag) {
      48             :     return felixbus::FelixBusInfo{
      49             :       .fid = tag,
      50          13 :       .ip = m_settings.ip,
      51          13 :       .port = m_port,
      52          13 :       .unbuffered = not m_settings.buffered,
      53             :       .pubsub = false,
      54          13 :       .tcp = m_settings.network_type == netio3::NetworkType::ASYNCMSG,
      55             :       .stream = false,
      56          13 :       .netio_pages = m_settings.buffer_count,
      57          13 :       .netio_pagesize = m_settings.buffer_size,
      58          26 :     };
      59          39 :   });
      60           7 :   m_bus_writer.publish(info, m_settings.bus_filename);
      61           7 : }