/// Creates a struct for the NetworkThread and binds to a socket without actually spawning a thread.
/// The `input_queue` is used to send DispatchEvents to the thread but they won't be read unless
/// the `dispatcher_registration` is activated to wake up the thread.
/// `network_thread_sender` is used to confirm shutdown of the thread.
Creates a struct for the NetworkThread and binds to a socket without actually spawning a thread.
The `input_queue` is used to send DispatchEvents to the thread but they won't be read unless
the `dispatcher_registration` is activated to wake up the thread.
`network_thread_sender` is used to confirm shutdown of the thread.
Creates a struct for the NetworkThread and binds to a socket without actually spawning a thread . The ` input_queue ` is used to send DispatchEvents to the thread but they won ' t be read unless the ` dispatcher_registration ` is activated to wake up the thread . ` network_thread_sender ` is used to confirm shutdown of the thread .
pub fn new(
        log: KompactLogger,
        addr: SocketAddr,
        lookup: Arc<ArcSwap<ActorStore>>,
        input_queue: Recv<DispatchEvent>,
        shutdown_promise: KPromise<()>,
        dispatcher_ref: DispatcherRef,
        network_config: NetworkConfig,
    ) -> (NetworkThread, Waker) {
        // Set-up the Listener
        debug!(
            log,
            "NetworkThread starting, trying to bind listener to address {}", &addr
        );
        match bind_with_retries(&addr, MAX_BIND_RETRIES, &log) {
            Ok(mut tcp_listener) => {
                let actual_addr = tcp_listener.local_addr().expect("could not get real addr");
                let logger = log.new(o!("addr" => format!("{}", actual_addr)));
                let mut udp_socket =
                    UdpSocket::bind(actual_addr).expect("could not bind UDP on TCP port");
                // Set up polling for the Dispatcher and the listener.
                let poll = Poll::new().expect("failed to create Poll instance in NetworkThread");
                // Register Listener
                let registry = poll.registry();
                registry
                    .register(&mut tcp_listener, TCP_SERVER, Interest::READABLE)
                    .expect("failed to register TCP SERVER");
                registry
                    .register(
                        &mut udp_socket,
                        UDP_SOCKET,
                        Interest::READABLE | Interest::WRITABLE,
                    )
                    .expect("failed to register UDP SOCKET");
                // Create waker for Dispatch
                let waker = Waker::new(poll.registry(), DISPATCHER)
                    .expect("failed to create Waker for DISPATCHER");
                let mut buffer_pool = BufferPool::with_config(
                    &network_config.get_buffer_config(),
                    &network_config.get_custom_allocator(),
                );
                let encode_buffer = EncodeBuffer::with_config(
                    &network_config.get_buffer_config(),
                    &network_config.get_custom_allocator(),
                );
                let udp_buffer = buffer_pool
                    .get_buffer()
                    .expect("Could not get buffer for setting up UDP");
                let udp_state =
                    UdpState::new(udp_socket, udp_buffer, logger.clone(), &network_config);
                let channel_map: FxHashMap<SocketAddr, TcpChannel> = FxHashMap::default();
                let token_map: FxHashMap<Token, SocketAddr> = FxHashMap::default();
                (
                    NetworkThread {
                        log: logger,
                        addr: actual_addr,
                        lookup,
                        tcp_listener: Some(tcp_listener),
                        udp_state: Some(udp_state),
                        poll,
                        channel_map,
                        token_map,
                        token: START_TOKEN,
                        input_queue,
                        buffer_pool,
                        sent_bytes: 0,
                        received_bytes: 0,
                        sent_msgs: 0,
                        stopped: false,
                        shutdown_promise: Some(shutdown_promise),
                        dispatcher_ref,
                        network_config,
                        retry_queue: VecDeque::new(),
                        out_of_buffers: false,
                        encode_buffer,
                    },
                    waker,
                )
            }
            Err(e) => {
                panic!(
                    "NetworkThread failed to bind to address: {:?}, addr {:?}",
                    e, &addr
                );
            }
        }
    }
pub fn new ( log : KompactLogger , addr : SocketAddr , lookup : Arc < ArcSwap < ActorStore > > , input_queue : Recv < DispatchEvent > , shutdown_promise : KPromise < ( ) > , dispatcher_ref : DispatcherRef , network_config : NetworkConfig , ) -> ( NetworkThread , Waker ) { // Set-up the Listener debug ! ( log " " addr ) ; match bind_with_retries ( & addr , MAX_BIND_RETRIES , & log ) { Ok ( mut tcp_listener ) => { let actual_addr = tcp_listener . local_addr ( ) . expect ( " " ) ; let logger = log . new ( o ! ( " " format ( " " actual_addr ) ) ) ; let mut udp_socket = UdpSocket :: bind ( actual_addr ) . expect ( " " ) ; // Set up polling for the Dispatcher and the listener. let poll = Poll :: new ( ) . expect ( " " ) ; // Register Listener let registry = poll . registry ( ) ; registry . register ( & mut tcp_listener , TCP_SERVER , Interest :: READABLE ) . expect ( " " ) ; registry . register ( & mut udp_socket , UDP_SOCKET , Interest :: READABLE | Interest :: WRITABLE , ) . expect ( " " ) ; // Create waker for Dispatch let waker = Waker :: new ( poll . registry ( ) , DISPATCHER ) . expect ( " " ) ; let mut buffer_pool = BufferPool :: with_config ( & network_config . get_buffer_config ( ) , & network_config . get_custom_allocator ( ) , ) ; let encode_buffer = EncodeBuffer :: with_config ( & network_config . get_buffer_config ( ) , & network_config . get_custom_allocator ( ) , ) ; let udp_buffer = buffer_pool . get_buffer ( ) . expect ( " " ) ; let udp_state = UdpState :: new ( udp_socket , udp_buffer , logger . clone ( ) , & network_config ) ; let channel_map : FxHashMap < SocketAddr , TcpChannel > = FxHashMap :: default ( ) ; let token_map : FxHashMap < Token , SocketAddr > = FxHashMap :: default ( ) ; ( NetworkThread { log : logger , addr : actual_addr , lookup , tcp_listener : Some ( tcp_listener ) , udp_state : Some ( udp_state ) , poll , channel_map , token_map , token : START_TOKEN , input_queue , buffer_pool , sent_bytes : 0 , received_bytes : 0 , sent_msgs : 0 , stopped : false , shutdown_promise : Some ( shutdown_promise ) , dispatcher_ref , network_config , retry_queue : VecDeque :: new ( ) , out_of_buffers : false , encode_buffer , } , waker , ) } Err ( e ) => { panic ! ( " " e addr ) ; } } }
Creates a struct for the NetworkThread and binds to a socket without actually spawning a thread.
Creates a struct for the NetworkThread and binds to a socket without actually spawning a thread .
log
KompactLogger
addr
SocketAddr
lookup
Arc<ArcSwap<ActorStore>>
input_queue
Recv<DispatchEvent>
shutdown_promise
KPromise<()>
dispatcher_ref
DispatcherRef
network_config
NetworkConfig