node.h

Go to the documentation of this file.

#pragma once
 
#include "librats/util/rats_export.h"
#include "librats/transport/connection.h"      // ConnectionDelegate
#include "librats/transport/reactor_pool.h"
#include "librats/core/address.h"
#include "librats/peer/peer_table.h"
#include "librats/peer/peer_id.h"
#include "librats/peer/peer_info.h"
#include "librats/security/identity.h"
#include "librats/security/handshaker.h"  // SecurityProvider
#include "librats/node/circuit_service.h"
#include "librats/node/config.h"
#include "librats/node/dial_service.h"
#include "librats/node/dialer.h"
#include "librats/node/nat_status.h"
#include "librats/node/node_context.h"    // NodeContext, EventBus, ServiceRegistry
#include "librats/peer/peer.h"
#include "librats/node/peer_network.h"
#include "librats/wire/message_router.h"
 
#include <atomic>
#include <condition_variable>
#include <functional>
#include <memory>
#include <mutex>
#include <optional>
#include <string>
#include <string_view>
#include <thread>
#include <type_traits>
#include <vector>
 
namespace librats {
 
class NetworkMonitor;  // util/network_monitor.h — owned via unique_ptr, included in node.cpp
class MessageJson;     // subsystems/message_json.h — reached via json() (json.h stays out of node.h)
 
class RATS_API Node final : public ConnectionDelegate,
                            public PeerNetwork,
                            public DialService,
                            public CircuitService {
public:
    explicit Node(NodeConfig config);
    ~Node() override;
 
    Node(const Node&) = delete;
    Node& operator=(const Node&) = delete;
 
    template <class T>
    T* add_subsystem(std::unique_ptr<T> subsystem) {
        static_assert(std::is_base_of<Subsystem, T>::value, "T must derive from Subsystem");
        T* raw = subsystem.get();
        subsystems_.push_back(std::move(subsystem));  // upcast to unique_ptr<Subsystem>
        return raw;
    }
 
    bool start();
    void stop();
 
    const PeerId& local_id() const noexcept override { return identity_.id; }
    uint16_t      listen_port() const noexcept override { return listen_port_; }
 
    uint8_t       transports() const noexcept override { return transports_; }
 
    const std::string& protocol() const noexcept override { return config_.protocol; }
 
    // — node-scoped coordination, shared by subsystems and the app (see NodeContext) —
    //   events()   : fire-and-forget notifications, one→many (host events, …)
    //   services() : targeted synchronous calls by capability interface, one→one
    EventBus&        events()   noexcept { return events_; }
    ServiceRegistry& services() noexcept { return services_; }
 
    // — connections —
    void connect(const Address& address) override;
    void connect(const std::string& host, uint16_t port);
 
    size_t                  peer_count() const noexcept { return peers_.size(); }
    std::vector<PeerInfo>   peers() const override { return peers_.snapshot(); }
    std::optional<Peer> peer(const PeerId& id);
 
    std::vector<Address> observed_addresses() const;
 
    const NatStatus& nat_status() const noexcept { return nat_status_; }
 
    // — DialService: dial one endpoint over one wire, bypassing the transport race
    //   (see node/dial_service.h; used by hole punching) —
    bool dial_direct(const Address& addr, TransportKind kind,
                     const DialProfile& profile) override;
 
    // — CircuitService: make a relayed byte stream an ordinary peer connection
    //   (see node/circuit_service.h; used by the relay module) —
    std::optional<PeerRoute> adopt_circuit(const PeerId& carrier, std::unique_ptr<Link> link,
                                           ConnRole role, bool connected) override;
    void                     wake_circuit(PeerRoute route, uint32_t events) override;
    void                     close_circuit(PeerRoute route, CloseReason reason) override;
 
    // — peer admission limit (0 = unlimited; guards inbound, not our own dials) —
    size_t max_peers() const noexcept { return max_peers_.load(std::memory_order_relaxed); }
    void   set_max_peers(size_t n) noexcept { max_peers_.store(n, std::memory_order_relaxed); }
    bool   peer_limit_reached() const noexcept {
        const size_t cap = max_peers_.load(std::memory_order_relaxed);
        return cap != 0 && peers_.size() >= cap;
    }
 
    // — application messaging (raw bytes on a named channel) —
    bool send(const PeerId& to, std::string_view channel, ByteView payload);
    bool broadcast(std::string_view channel, ByteView payload);
 
    bool peer_writable(const PeerId& id) const override;
 
    // — events (register before start(); invoked on a reactor thread). Multiple
    //   listeners are supported, so subsystems and the app can both subscribe. —
    void on_peer_connected(PeerNetwork::PeerEventHandler cb) override { peer_connected_.push_back(std::move(cb)); }
    void on_peer_disconnected(PeerNetwork::PeerDisconnectHandler cb) override { peer_disconnected_.push_back(std::move(cb)); }
    void on_dial_failed(PeerNetwork::DialFailedHandler cb) override { dial_failed_.push_back(std::move(cb)); }
    void on_peer_writable(PeerNetwork::PeerEventHandler cb) override { peer_writable_.push_back(std::move(cb)); }
    void on(std::string_view channel, MessageRouter::Handler cb) { router_.on_channel(channel, std::move(cb)); }
 
    // — typed lookup of an attached subsystem (nullptr if none of that type) —
    //   reaches a module's own API without threading a pointer from add_subsystem:
    //     if (auto* j = node.json()) j->on("chat", …);
    template <class T>
    T* subsystem() noexcept {
        for (auto& s : subsystems_) if (auto* p = dynamic_cast<T*>(s.get())) return p;
        return nullptr;
    }
    MessageJson* json() noexcept;
 
    // — PeerNetwork (for subsystems) —
    bool                send(const PeerId& to, MessageType type, ByteView payload) override;
    bool                broadcast(MessageType type, ByteView payload) override;
    std::vector<PeerId> connected_peers() const override;
    void                on(MessageType type, PeerNetwork::MessageHandler cb) override { router_.on_type(type, std::move(cb)); }
 
private:
    friend class Peer;
 
    // ConnectionDelegate (reactor thread)
    bool admit_inbound() override;
    void on_established(Connection& conn) override;
    void on_frame(Connection& conn, const Frame& frame) override;
    void on_closed(Connection& conn, CloseReason reason) override;
    void on_writable_changed(Connection& conn, bool writable) override;
    void on_dial_aborted(uint8_t reactor_index, ConnId id,
                         const std::string& host, uint16_t port) override;
 
    bool open_listeners();
    void report_dial_failed(const std::string& host, uint16_t port);
 
    Peer make_peer(const PeerId& id, PeerRoute route) { return Peer(id, route, *this); }
    void route_send(PeerRoute route, FrameHeader header, Bytes payload,
                    std::shared_ptr<std::atomic<size_t>> owed = nullptr);
    size_t send_low_water() const noexcept;
    void route_close(PeerRoute route);
 
    // — identify: how peers learn each other's dialable addresses (reactor thread) —
    void                 send_identify(Connection& conn);            
    void                 handle_identify(Connection& conn, const Frame& frame);  
    std::vector<Address> advertised_addresses() const;              
    void                 rebuild_advertised_addresses(const std::vector<std::string>& local_ips);
    void                 record_observed_address(const Address& addr);
    bool                 is_own_endpoint(const Address& addr) const;
 
    void start_network_monitor();   
    void stop_network_monitor();    
    void maintenance_loop();        
 
    NodeConfig                        config_;
    Identity                          identity_;
    std::unique_ptr<SecurityProvider> security_;
    PeerTable                     peers_;
    MessageRouter                     router_;
    EventBus                          events_;      
    ServiceRegistry                   services_;    
    std::unique_ptr<ReactorPool>      reactors_;
    std::unique_ptr<Dialer>           dialer_;      
 
    std::vector<std::unique_ptr<Subsystem>> subsystems_;
 
    // Host network-change watch. The monitor signals on its own thread; the
    // maintenance thread does the (possibly blocking) EventBus emit off it, so
    // subscribers may run slow recovery without stalling change detection.
    std::unique_ptr<NetworkMonitor> monitor_;
    std::thread                     maintenance_thread_;
    std::mutex                      maintenance_mutex_;
    std::condition_variable         maintenance_cv_;
    std::vector<std::string>        pending_addresses_;
    bool                            maintenance_pending_ = false;
    bool                            maintenance_stop_    = false;
 
    socket_t            listen_socket_ = RATS_INVALID_SOCKET;  
    socket_t            udp_socket_    = RATS_INVALID_SOCKET;  
    uint16_t            listen_port_   = 0;   
    uint8_t             transports_    = 0;   
    std::atomic<bool>   running_{false};
    std::atomic<size_t> max_peers_{0};  
 
    std::vector<PeerNetwork::PeerEventHandler>      peer_connected_;
    std::vector<PeerNetwork::PeerDisconnectHandler> peer_disconnected_;
    std::vector<PeerNetwork::DialFailedHandler>     dial_failed_;
    std::vector<PeerNetwork::PeerEventHandler>      peer_writable_;
 
    // Our own addresses as peers observe us (their reported IP + our listen port).
    mutable std::mutex   observed_mutex_;
    std::vector<Address> observed_addresses_;
 
    // The datagram half of the same knowledge, which is a different thing: what a
    // peer sees on a UDP link is our NAT's mapping of the one shared socket, port
    // included, and several peers' views of it are what say whether that mapping is
    // stable enough to punch through. Published as ExternalAddressService.
    NatStatus            nat_status_;
 
    // The dialable addresses we advertise to peers in identify. Derived from local
    // interfaces (and, in future, promoted observed addresses). Rebuilt once at
    // start() and on NetworkMonitor changes — never re-enumerated per connection,
    // since interface enumeration is a syscall and the send path is hot.
    mutable std::mutex   advertised_mutex_;
    std::vector<Address> advertised_addresses_;
};
 
} // namespace librats