boomnet 1.0.0-rc.2

Overview

BoomNet is a high-performance framework targeting development of low-latency network applications, particularly focusing on TCP stream-oriented clients that utilise various protocols.

Installation

Simply declare dependency on boomnet in your Cargo.toml and select desired features.

[dependencies]
boomnet = { version = "0.0.89", features = ["rustls-webpki", "ws", "mio"]}

Design Principles

The framework is structured into multiple layers, with each subsequent layer building upon its predecessor, enhancing functionality and abstraction.

Stream

The first layer defines stream as abstraction over TCP connection, adhering to the following characteristics.

• Must implement Read and Write traits for I/O operations.
• Operates in a non-blocking manner.
• Integrates with TLS using rustls or openssl.
• Supports recording and replay of network byte streams.
• Allows binding to specific network interface.
• Facilitates implementation of TCP oriented client protocols such as WebSocket, HTTP, and FIX.

Streams are designed to be fully generic, avoiding dynamic dispatch, and can be composed in flexible way.

let stream: RecordedStream<TlsStream<TcpStream>> = TcpStream::try_from((host, port))?
    .into_tls_stream()?
    .into_default_recorded_stream();

Different protocols can then be applied on top of a stream in order to create a client.

let ws: Websocket<RecordedStream<TlsStream<TcpStream>>> = stream.into_websocket("/ws");

Selector

Selector provides abstraction over OS specific mechanisms (like epoll) for efficiently monitoring socket readiness events. Though primarily utilised internally, selectors are crucial for the IOService functionality, currently offering both mio and direct (no-op) implementations.

let mut io_service = MioSelector::new()?.into_io_service();

Service

The last layer manages lifecycle of endpoints and provides auxiliary services (such as asynchronous DNS resolution and auto disconnect) through the IOService.

Endpoint serves as connection factory and is where application logic lives. IOService oversees the connection lifecycle within endpoints.

Protocols

The aim is to support a variety of protocols, including WebSocket, HTTP, and FIX.

Websocket

The websocket client protocol complies with the RFC 6455 specification, offering the following features.

• Compatibility with any stream.
• TCP batch-aware frame processing.
• Not blocking on partial frame(s).
• No memory allocations (except to initialise buffers)
• Designed for zero-copy read and write.
• Optional masking of outbound frames.
• Standalone usage or in conjunction with IOService.

Http

Provides http 1.1 client that is compatible with any non-blocking stream and does perform memory allocations.

Example Usage

The repository contains comprehensive list of examples.

The following example illustrates how to use multiple websocket connections with IOService in order to consume messages from the Binance cryptocurrency exchange. First, we define an Endpoint whose target is a WebSocket over TLS.

struct TradeEndpoint {
    connection_info: ConnectionInfo,
    ws_endpoint: String,
    instrument: &'static str,
}

impl TradeEndpoint {
    pub fn new(url: &'static str, instrument: &'static str) -> TradeEndpoint {
        let (connection_info, ws_endpoint, _) = boomnet::ws::util::parse_url(url).unwrap();
        Self { connection_info, ws_endpoint, instrument, }
    }
}

impl ConnectionInfoProvider for TradeEndpoint {
    fn connection_info(&self) -> &ConnectionInfo {
        &self.connection_info
    }
}

impl Endpoint for TradeEndpoint {
    type Target = Websocket<TlsStream<MioStream>>;

    // called by the IO service whenever a connection has to be established for this endpoint
    fn create_target(&mut self, addr: SocketAddr) -> io::Result<Option<Self::Target>> {

        let mut ws = TcpStream::try_from((&self.connection_info, addr))?
            .into_mio_stream()
            .into_tls_websocket(&self.ws_endpoint)?;

        // send subscription message
        ws.send_text(
            true,
            Some(format!(r#"{{"method":"SUBSCRIBE","params":["{}@trade"],"id":1}}"#, self.instrument).as_bytes()),
        )?;

        Ok(Some(ws))
    }
}

After defining the endpoint, it is registered with the IOService and polled within an event loop. The service handles connection lifecycle and exposes every active target through an ActiveEndpoint guard. I/O performed with try_with automatically starts the endpoint's reconnection lifecycle if it fails.

fn main() -> anyhow::Result<()> {
    let mut io_service = MioSelector::new()?.into_io_service();

    let endpoint_btc = TradeEndpoint::new("wss://stream1.binance.com:443/ws", "btcusdt");
    let endpoint_eth = TradeEndpoint::new("wss://stream2.binance.com:443/ws", "ethusdt");
    let endpoint_xrp = TradeEndpoint::new("wss://stream3.binance.com:443/ws", "xrpusdt");

    io_service.register(endpoint_btc)?;
    io_service.register(endpoint_eth)?;
    io_service.register(endpoint_xrp)?;

    loop {
        // will never block
        for event in io_service.poll()? {
            if let IOServiceEvent::Active(active) = event {
                let handle = active.handle();
                active.try_with(|ws| {
                    for frame in ws.read_batch()? {
                        if let WebsocketFrame::Text(fin, data) = frame? {
                            println!("[{handle:?}] ({fin}) {}", String::from_utf8_lossy(data));
                        }
                    }
                    Ok(())
                })?;
            }
        }
    }
}

It is often required to expose shared state to the Endpoint. This can be achieved with user defined Context.

struct FeedContext;

// use the marker trait
impl Context for FeedContext {}

When implementing our TradeEndpoint we can use EndpointWithContext instead.

impl EndpointWithContext<FeedContext> for TradeEndpoint {
    type Target = Websocket<TlsStream<MioStream>>;

    fn create_target(&mut self, addr: SocketAddr, ctx: &mut FeedContext) -> io::Result<Option<Self::Target>> {
        // we now have access to context
        // ...
    }
}

We will also need to create IOService that is Context aware.

let mut context = FeedContext::new();
let mut io_service = MioSelector::new()?.into_io_service_with_context();

The Context is passed to the service for lifecycle callbacks. The returned iterator does not borrow it, so application processing can use it too.

loop {
    for event in io_service.poll(&mut context)? {
        if let IOServiceEvent::Active(active) = event {
            active.try_with(|ws| {
                for frame in ws.read_batch()? {
                    let frame = frame?;
                    context.process(frame);
                }
                Ok(())
            })?;
        }
    }
}

Features

The framework feature set is modular, allowing for tailored functionality based on project needs.

• mio
• rustls-native
• rustls-webpki
• openssl
• ktls
• ws
• http

mio

Adds dependency on mio crate and enables MioSelector and MioStream.

rustls-native

Adds dependency on rustls crate with rustls-native-certs and enables TlsStream as well as more flexible TlsReadyStream.

rustls-webpki

Adds dependency on rustls crate with webpki-roots and enables TlsStream as well as more flexible TlsReadyStream.

openssl

Adds dependency on openssl crate and enables TlsStream as well as more flexible TlsReadyStream.

ktls

Activates openssl feature and enables KtlsStream that offloads TLS to the kernel (KTLS).

ws

Adds support for Websocket protocol.

http

Adds support for Http1.1 protocol.