Fast and easy-to-use event-driven network library.

This is especially important since this crate is meant to be used as a library.

For a example an nmap scan while termchat is active will crash here https://github.com/lemunozm/message-io/blob/master/src/network_adapter.rs#L348

message-io doesn't seem to work on 32-bit systems. Some of the constants in resource_id.rs (namely RESOURCE_TYPE_BIT and ADAPTER_ID_MASK_OVER_ID) overflow in 32-bit systems.

Perhaps the constants can be updated to make use of std::mem::size_of, or the type changed from usize to u64?

In main I start two tasks via tokio. One is a graphql server and the other is a service running node_listener.for_each(... to handle messages

    // Create and start discovery service
    // thread::spawn(move || {
    let t1 = tokio::task::spawn_blocking(move || {
        debug!("Creating Discovery Service");
        let dsserver = DiscoveryServer::new();
        if let Ok(discovery) = dsserver {
            debug!("Starting discovery service");
            let _ = discovery.run();
        }
    });

    tokio::join!(async move {
        let _ = graphql_server.run().await;
    });

When I want to terminate, the discovery service stays and can only be killed brutally sudo kill -9 PID Is there a clean way to structure the code such that it can terminate gently?

original discussion https://github.com/lemunozm/termchat/pull/11

Sending multiple small data:

• using this branch https://github.com/sigmaSd/termchat/tree/send_in_chunks (sends file in small chunks)
• send a 10mb file with ?send
• try to send any message after -> no message is received
• debugging this lead me to encoding here https://github.com/lemunozm/message-io/blob/master/src/encoding.rs#L100 decoded_data at some point becomes always None

Sending big data:

• using this branch https://github.com/sigmaSd/termchat/tree/send_files (send the whole file in one send)
• send 10mb file
• receiver will crash here https://github.com/lemunozm/message-io/blob/master/src/network.rs#L112 ``Err value: Io(Kind(UnexpectedEof))
• the issue might be here https://github.com/lemunozm/message-io/blob/master/src/encoding.rs#L11 the PADDING can be too small for the data length

Let's say a client sends a bogus packet, or some error occurs, or the server wants to ban or rate-limit a client, would it be possible to add a way to close an endpoint? That is, cut the connection. The API could look like Network::close(Endpoint). The NetEvent::Disconnected event would be sent to the server, and if a new connection attempt is done, the NetEvent::Connected would be sent again.

I'm using message-io as part of my game's networking. Using the previous API I was doing something like this (pseudo code)

let mut game_state = GameState::new();
let mut networking = Networking::new(); // connects to the server

loop {
    let new_inputs = networking.handle_packets();
    game_state.update(new_inputs);
    draw(&game_state);
}


// handle_packets being something like this
fn handle_packets(&mut self) -> Option<Inputs> {
    while let Ok(packet) = self.network.try_receive() {
         // handle packet
    }
    self.new_inputs
}

However, it looks like using the new API I cannot do this as the for_each loop completely takes control of the thread used.

My guess is that I now have to communicate between my game state and the networking code using channels.. I liked that it was abstracted away.

Hi, I have a question , most likely is a simple/silly one but I'm new in Rust so I hope you can answer me:

I run the ping-pong example and it work. But in order to use it in my project I need to access to client/server endpoint from outside listener loop because I want to keep just one connection all the time between client and server.

Outside the ping-pong example I have my project (I use a game engine which use winit as crate) where I have my UI with a button that I made in order to send a message when I click on it. I was going to use handler.network().send(server_id, &output_data); But I have no access to the server_id / endpoint because the point where my button trigger is inside at winit crate loop (aka UI).

Because the UI loop consume(move) the data - it use a closure - it give an error when I try to save endpoint in a external struct because Endpoint doesn't have Clone trait.

But I'm not confident to make changes yet in your code :).

So my question is: which solution/option do you suggest? I tried my best to explain my problem but tell me if you need more info.

Hi,

I am looking into creating a game server for my classical RPG project at Eldiron.com.

Question: From your perspective what would be a good design implementation to have one thread on the server communicate with the clients and another thread performing the game ticks on the characters ? Is there some kind of best practice for this ?

Thanks!

I'm trying to implement message-io as a tcp client within the actix framework. This is my current implementation:

use actix::Message;
use actix::{
    Actor, ActorContext, AsyncContext, Context, Handler, Running, StreamHandler, WrapFuture,
};
use log::info;
use message_io::network::{NetEvent, Transport};
use message_io::node::{
    self, NodeEvent, NodeHandler, NodeListener, NodeTask, StoredNetEvent, StoredNodeEvent,
};

pub struct TcpClientActor {
    handler: NodeHandler<String>,
    listener: Option<NodeListener<String>>,
}

impl TcpClientActor {
    pub fn new() -> Self {
        let (handler, listener) = node::split::<String>();
        TcpClientActor {
            handler,
            listener: Some(listener),
        }
    }
}

impl Actor for TcpClientActor {
    type Context = Context<Self>;

    fn started(&mut self, ctx: &mut Self::Context) {
        let message = serde_json::json!({
            "op": "subscribe",
            "topic": "/client_count"
        });
        // Listen for TCP, UDP and WebSocket messages at the same time.
        let (server, socket_address) = self
            .handler
            .network()
            .connect(Transport::Tcp, "192.168.43.217:9092")
            .unwrap();
    }

    fn stopping(&mut self, ctx: &mut Self::Context) -> Running {
        ctx.stop();
        Running::Stop
    }
}

#[derive(Message)]
#[rtype(result = "()")]
pub struct Listen {}

impl Handler<Listen> for TcpClientActor {
    type Result = ();

    fn handle(&mut self, msg: Listen, ctx: &mut Self::Context) -> Self::Result {
        let listener = self.listener.take().unwrap();
        listener
            .for_each_async(move |event| {
                match event {
                    NodeEvent::Network(net_event) => match net_event {
                        NetEvent::Connected(_endpoint, _ok) => {
                            info!("Connected");
                            // handler.signals().send(message.to_string());
                        }
                        NetEvent::Accepted(_, _) => unreachable!(), // Only generated by listening
                        NetEvent::Message(_endpoint, data) => {
                            println!("Received: {}", String::from_utf8_lossy(data));
                        }
                        NetEvent::Disconnected(_endpoint) => (),
                    },
                    NodeEvent::Signal(signal) => match signal {
                        _ => {
                            // computed every second
                            info!("Signal Received: {}", signal);
                            // handler.signals().send_with_timer(signal, Duration::from_secs(1));
                        }
                    },
                }
            })
            .wait();
    }
}

The problem with the current implementation is that whenever the listener.for_each_async() method runs, the thread gets blocked and the entire program freezes. I though for_each_async was non-blocking. Am I misinterpreting something here? Thank you

Ps: If you need any information regarding the actix framework or anything else, just let me know and I'll write it down. I didn't want to overwhelm the issue with information that might not be needed.

Error message:

error[E0433]: failed to resolve: could not find `tcp` in `super`
   --> /home/redact/.cargo/registry/src/github.com-1ecc6299db9ec823/message-io-0.14.1/src/adapters/ws.rs:183:45
    |
183 |                     let tcp_status = super::tcp::check_stream_ready(&stream.0);
    |                                             ^^^ could not find `tcp` in `super`

This might be intended, but in that case there should be websocket = ["tcp"]. It also doesn't make sense for the web either. Maybe have two different websocket features? websocket and websocket-web.

hi @lemunozm ! excellent work on this crate, very exciting!

just wanted to share that while i was reviewing and running some of the examples, i found this issue where the second client to attempt running the multicast example panics after connecting, with an Address already in use error.

Hello, after adding message-io to new project and compiling, i got the following message:

warning: the following packages contain code that will be rejected by a future version of Rust: ntapi v0.3.7 note: To solve this problem, you can try the following approaches:

• Some affected dependencies have newer versions available. You may want to consider updating them to a newer version to see if the issue has been fixed.

ntapi v0.3.7 has the following newer versions available: 0.4.0

• If the issue is not solved by updating the dependencies, a fix has to be implemented by those dependencies. You can help with that by notifying the maintainers of this problem (e.g. by creating a bug report) or by proposing a fix to the maintainers (e.g. by creating a pull request):

  • [email protected]
  • Repository: https://github.com/MSxDOS/ntapi
  • Detailed warning command: cargo report future-incompatibilities --id 4 --package [email protected]

• If waiting for an upstream fix is not an option, you can use the [patch] section in Cargo.toml to use your own version of the dependency. For more information, see: https://doc.rust-lang.org/cargo/reference/overriding-dependencies.html#the-patch-section

note: this report can be shown with cargo report future-incompatibilities --id 4

It seems that just updating the dependency would fix this issue, in case it does not i would make a issue in their repo :)

I would like the ability to provide an alternate thread name for the threads in for_each() and for_each_async(). If this is a change you are open to incorporating, I can make the change and put up a PR. Any interest?

I can't manage to establish wss connections using message-io version 0.14.2 on macOS Big Sur. Example:

use message_io::network::{NetEvent, RemoteAddr, Transport};
use message_io::node::{self, NodeEvent};

fn main() {
    connect("ws://echo.websocket.org".to_string());
    connect("wss://echo.websocket.org".to_string());
}

fn connect(url: String) {
    let (handler, listener) = node::split::<()>();
    handler
        .network()
        .connect(Transport::Ws, RemoteAddr::Str(url.clone()))
        .unwrap();
    listener.for_each(move |event| match event {
        NodeEvent::Network(NetEvent::Connected(e, success)) => {
            println!("{} success={} {:?}", url, success, e);
            handler.stop();
        }
        _ => panic!(),
    });
}

Output:

ws://echo.websocket.org success=true Endpoint { resource_id: [3.R.0], addr: 174.129.224.73:80 }
wss://echo.websocket.org success=false Endpoint { resource_id: [3.R.0], addr: 174.129.224.73:443 }

However, I can connect with wss using websocat from the same machine:

$ echo foo | websocat wss://echo.websocket.org
foo

It looks like it's hitting this code path in the ws adapter:

WS client handshake error: WebSocket protocol error: Handshake not finished

Would appreciate guidance if I'm doing something incorrectly. Thanks!

Currently, WebSocket is working for client/server side. However, the browsers do not allow to create tcp connections direclty (which is the current implementation based). Instead, the web-sys must be used.

• Use a different WebSocket implementation if the target is wasm.
• wasm example of a client.

This PR handles the Nagle algorithm in FramedTcp allowing to send a message immediately without sending partially (whatever possible). In other words: it disabled the algorithm until the message is fully in the OS sending buffer, and then enables it to send the full message without waiting for another possible message.

This protocol is message-oriented. This means that receiving less than a message has no meaning for the receiver who needs to wait for the remaining data to produce a meaningful message (which implies allocating in a Decoder). For that reason, sending partial data before having the whole message does not improve anything. What's more, it saturates the network.

Although this feature implies a latency reduction since once the complete message is OS output buffer it is sent, could also imply an impact on the throughput. Anyway, the user of a FramedTcp transport would want latency versus throughput. If throughput is the target, they will probably send big messages that will be less impacted by this change, or even more probably they prefer to choose TCP transport that fits better for throughput.

NOTE: Conceptually it should work, but how to evaluate this improvement?