Tsyncp: Channel-like primitives (mpsc, broadcast, etc) over TCP

Hello.

tsyncp is a simple but extendable library that provides channel-like APIs (mpsc, broadcast, etc) over TCP.

github | crate.io | examples

Channel APIs are a pretty common pattern for message-passing between threads and tasks. For example, std has std::sync::mpsc for passing messages between threads:

use std::sync::mpsc;

let (tx, rx) = mpsc::channel();
std::thread::spawn(move|| {
    tx.send(10).unwrap();
});

and tokio has tokio::sync::mpsc for passing messages between tasks:

use tokio::sync::mpsc;

let (tx, mut rx) = mpsc::channel(100);

tokio::spawn(async move {
    for i in 0..10 {
        if let Err(_) = tx.send(i).await {
            println!("receiver dropped");
            return;
        }
    }
});

I thought it would be pretty convenient to have something similar for message passing over the network, so I decided write one over TCP.

Building and using the APIs are similar to tokio's implementation.

Sender can be used as below:

let mut tx: mpsc::JsonSender<Dummy> = mpsc::sender_to("localhost::11114").await?;

let dummy = Dummy { ... };

tx.send(dummy).await?;

Receiver can be used as below:

use tsyncp::mpsc;

let mut rx: mpsc::JsonReceiver<Dummy> = mpsc::receiver_on("localhost:11114").await?;

// accept a sender connection
rx.accept().await?;

while let Some(Ok(item)) = rx.recv().await {
    println("received {item:?}!");
    ...
}

Future Chaining:

Tsyncp uses future-chaining to keep the simplicity and allow optional extendability.

To extend use-cases of the APIs, you can chain future-methods to allow more advanced functionalities.

For example, you can extend the receiver like this:

let mut rx: mpsc::JsonReceiver<Dummy> = mpsc::receiver_on("localhost:11114")
    .limit(10)            // set connection limit to 10.
    .accept()             // accept before returning Receiver.
    .to_limit()           // accept until limit is reached.
    .await?;

it let Some(Ok((item, addr))) = rx.recv().with_addr().await {
    println!("received {item:?} from {addr}!");
    ...
}

Above code will initialize the Receiver with connection limit 10, accept connections up to the limt; and it will receive data along with the address it came from.

The API doc has a detailed guide on how to build, use, and extend these primitives, so if you think this is interesting, please check them out. You can also try out the chat\_server and chat\_client examples in the repository, along with other examples.

This library provides 5 types of channels:

• mpsc
• broadcast
• barrier
• channel
• multi\_channel

This library supports 3 different encoding/decoding choices (json, protobuf, bincode), but users can also implement their own encoding/decoding schemes pretty easily.

Also, if anyone wants to review the code and give feedback of implementations, module organizations, documentations, etc., I would be very very grateful, as I just wrote this over a few weekends by myself. For areas that need reviews, I would say, are implementations of stream pool and length-delimited codec. Not sure if this library is worth contributing to, but I will be eternally grateful if anyone wanted to contribute.

If people are interested in the library, I will continue working on it: extensive testing, benchmarking and proper CI flow, more primitives, provide options for Tls or noise encrypting, etc.

Thank you!