Program Listing for File netio3_receiver.cpp

↰ Return to documentation for file (network/netio3_receiver.cpp)

#include "network/netio3_receiver.hpp"

#include "felix/felix_fid.h"
#include "netio3-backend/Netio3Backend.hpp"

#include "fromhost_message.hpp"
#include "netio3_utility.hpp"

Netio3Receiver::Netio3Receiver(const std::string &ip, uint32_t port, NetworkMode network_mode,
    unsigned int netio_pn, unsigned int netio_ps, bool buffered, Netio3EventLoop& evloop) :
    m_ip{ip},
    m_port{static_cast<uint16_t>(port)},
    m_num_buffers{netio_pn},
    m_buffer_size{netio_ps},
    m_evloop(&evloop),
    m_receiver{netio3::NetioReceiverConfig{
                   network::utility::get_network_type(network_mode),
                   network::utility::get_network_mode(network_mode),
                   netio3::ThreadSafetyModel::UNSAFE},
               m_evloop->get_evloop()},
    m_buffered{buffered},
    m_tcp{network_mode == NetworkMode::tcp}
{
    init_listen_socket();
}

void Netio3Receiver::init_listen_socket()
{
    m_receiver.set_on_connection_established_cb([this] (const netio3::EndPointAddress& address) { m_on_conn_open(std::format("{}:{}", address.address(), address.port())); });
    m_receiver.set_on_connection_closed_cb([this] (const netio3::EndPointAddress& address) { m_on_conn_close(std::format("{}:{}", address.address(), address.port())); });
    m_receiver.set_on_data_cb([this] (std::uint64_t fid, std::span<const std::uint8_t> data, std::uint8_t) { m_on_msg( {{ .elink=get_elink(fid), .payload=data, .status = ToFlxMessage::Status::MessageOk }}); });
    m_port = m_receiver.listen({m_ip, static_cast<unsigned short>(m_port)}, {.buf_size=m_buffer_size, .num_buf=m_num_buffers});
    // LOG_INFO("Initialised netio3 listen_socket. Port %u PROVIDER: %s", m_port, m_socket.fi->fabric_attr->prov_name);
}