2 unstable releases
1.0.0 |
|
---|---|
0.7.0 | Oct 24, 2024 |
0.6.0 | Oct 17, 2024 |
0.5.0 |
|
0.3.0 |
|
#45 in WebSocket
158 downloads per month
83KB
1.5K
SLoC
socket-flow
Straightforward async Websockets library for Rust! With a lot of examples available!
Introduction
This library is supposed to offer a simple implementation for websockets, so end-user could use this to wrap a websockets server/client into their application, offering a smooth way of setting it up into his code.
It's an async library based on tokio runtime, which uses a tokio TcpStream behind the scenes, using that as the starting point to implement the standards of WebSocket Protocol RFC, performing handshakes, reading frames, parsing masks, handling opcodes and internal payload.
Can be used as a client or server,
returning a WSConnection
, which implements the Stream
trait,
so you can continuously consume incoming messages, or send messages.
The motivation behind this, was to offer a simple way of having a WebSockets connection over your application, using as a
reference wide established libraries, like tungstenite-rs
and tokio-tungstenite
Features
Most of all WebSockets RFC features are implemented, like:
- Handshake process, key parsing and generation
- OpCodes handling, like
Text
,Binary
,Ping
,Pong
andContinue
- Multiple subscriptions
- Scalability
- Error handling
- It passes the autobahn-test-suite
- TLS Support
Features to be added:
- Compression
Usage
Add this in your Cargo.toml
:
[dependencies]
socket-flow = "*"
Examples of usage
This repo has different examples and flexible ways of using its dependencies to design the code as end-user needs.
We have the option of configuring all from scratch, creating the TcpListener, and managing the websockets connections, and we also have a plug-and-play option, which you can generate a Websockets server, with fewer lines of code.
Plug and play server
This is a very practical example, because you can have a server with just calling start_server
function, which returns
an EventStream
, for consuming server events, like new connections, messages, errors and disconnections.
You can also find in: Example
The start_server
function also accepts a rustls::ServerConfig
for enabling TLS in your server.
use futures::StreamExt;
use log::*;
use socket_flow::event::{Event, ID};
use socket_flow::server::start_server;
use socket_flow::split::WSWriter;
use std::collections::HashMap;
#[tokio::main]
async fn main() {
env_logger::init();
let port: u16 = 8080;
match start_server(port).await {
Ok(mut event_receiver) => {
let mut clients: HashMap<ID, WSWriter> = HashMap::new();
info!("Server started on address 127.0.0.1:{}", port);
while let Some(event) = event_receiver.next().await {
match event {
Event::NewClient(id, client_conn) => {
info!("New client {} connected", id);
clients.insert(id, client_conn);
}
Event::NewMessage(client_id, message) => {
info!("Message from client {}: {:?}", client_id, message);
let ws_writer = clients.get_mut(&client_id).unwrap();
ws_writer.send_message(message).await.unwrap();
}
Event::Disconnect(client_id) => {
info!("Client {} disconnected", client_id);
clients.remove(&client_id);
}
Event::Error(client_id, error) => {
error!("Error occurred for client {}: {:?}", client_id, error);
}
}
}
}
Err(err) => {
eprintln!("Could not start the server due to: {:?}", err);
}
}
}
For running this example, you can clone the repo and execute:
cargo run --color=always --package socket-flow --example simple_server
Echo server
Here is an echo-server example that you can also find in: Example
use futures::StreamExt;
use log::*;
use socket_flow::handshake::accept_async;
use socket_flow::stream::SocketFlowStream;
use std::net::SocketAddr;
use tokio::net::{TcpListener, TcpStream};
async fn handle_connection(_: SocketAddr, stream: TcpStream) {
match accept_async(SocketFlowStream::Plain(stream)).await {
Ok(mut ws_connection) => {
while let Some(result) = ws_connection.next().await {
match result {
Ok(message) => {
if ws_connection.send_message(message).await.is_err() {
error!("Failed to send message");
break;
}
}
Err(e) => {
error!("Received error from the stream: {}", e);
break;
}
}
}
}
Err(err) => error!("Error when performing handshake: {}", err),
}
}
#[tokio::main]
async fn main() {
env_logger::init();
let addr = "127.0.0.1:9002";
let listener = TcpListener::bind(&addr).await.expect("Can't listen");
info!("Listening on: {}", addr);
while let Ok((stream, peer)) = listener.accept().await {
info!("Peer address: {}", peer);
tokio::spawn(handle_connection(peer, stream));
}
}
For running this example, you can clone the repo and execute:
cargo run --color=always --package socket-flow --example echo_server
This example, creates a TcpListener, binding it to a port, accepting connections, handling each of these connections
inside a tokio task, for handling clients concurrently. The handle_connection function, make sure the handshake process
is performed, returning a WSConnection
, which implements Stream
trait, where you can consume incoming data for this client,
and perform write operations into the socket.
It includes error handling through Result
.
Simple client
Here is an example of how to run a client, that will perform some operations and disconnect gracefully:
use futures::StreamExt;
use log::*;
use rand::distr::Alphanumeric;
use rand::{thread_rng, Rng};
use socket_flow::handshake::connect_async;
use tokio::select;
use tokio::time::{interval, Duration};
async fn handle_connection(addr: &str) {
match connect_async(addr).await {
Ok(mut ws_connection) => {
let mut ticker = interval(Duration::from_secs(5));
// it will be used for closing the connection
let mut counter = 0;
loop {
select! {
Some(result) = ws_connection.next() => {
match result {
Ok(message) => {
info!("Received message: {}", message.as_text().unwrap());
counter = counter + 1;
// close the connection if 3 messages have already been sent and received
if counter >= 3 {
if ws_connection.close_connection().await.is_err() {
error!("Error occurred when closing connection");
}
break;
}
}
Err(err) => {
error!("Received error from the stream: {}", err);
break;
}
}
}
_ = ticker.tick() => {
let random_string = generate_random_string();
let binary_data = Vec::from(random_string);
if ws_connection.send(binary_data).await.is_err() {
eprintln!("Failed to send message");
break;
}
}
}
}
}
Err(err) => error!("Error when performing handshake: {}", err),
}
}
#[tokio::main]
async fn main() {
env_logger::init();
handle_connection("ws://localhost:9002").await;
}
fn generate_random_string() -> String {
thread_rng()
.sample_iter(&Alphanumeric)
.take(30)
.map(char::from)
.collect()
}
Since you need a server for testing the client, you can execute our echo-server example, and on another tab execute the client example:
cargo run --color=always --package socket-flow --example client
In this example, the client will try to connect to ws://127.0.0.1:9002
,
if the connection is established, it will start sending random strings every 5 seconds into the socket.
After sending three strings, it will close the connection gracefully and end its execution.
You can check more examples over Examples.
Testing
Socket-flow passes the Autobahn Test Suite for WebSockets. Also, it has some internal tests, for ensuring reliability.
TLS/SSL
By default, this library only accepts tokio-rustls, as an adapter library for adding TLS in your client/server implementation with socket-flow.
For checking how to set up TLS in server/client, and finding some examples, go to: TLS Examples.
References
Dependencies
~15–25MB
~457K SLoC