Struct remoc::Connect

pub struct Connect<'transport, TransportSinkError, TransportStreamError>(/* private fields */);

Available on crate feature rch only.

Methods for establishing a connection over a physical transport.

You must poll the returned Connect future or spawn it onto a task for the connection to work.

Physical transport

All functionality in Remoc requires that a connection over a physical transport is established. The underlying transport can either be of packet type (implementing Sink and Stream) or a socket-like object (implementing AsyncRead and AsyncWrite). In both cases it must be ordered and reliable. That means that all packets must arrive in the order they have been sent and no packets must be lost. The maximum packet size can be limited, see the configuration for that.

TCP is an example of an underlying transport that is suitable. But there are many more candidates, for example, UNIX domain sockets, pipes between processes, serial links, Bluetooth L2CAP streams, etc.

The connect functions are used to establish a base channel connection over a physical transport. Then, additional channels can be opened by sending either the sender or receiver half of them over the established base channel or another connected channel. See the examples in the remote channel module for details.

Convenience functions

Methods from the ConnectExt trait can be used on the return values of all connect methods. They streamline connection handling when a single value, such as a RTC client, should be exchanged over the connection and the flexibility of a base channel is not necessary.

Example

In the following example the server listens on TCP port 9875 and the client connects to it. Then both ends establish a Remoc connection using Connect::io over the TCP connection. The connection dispatchers are spawned onto new tasks and the client and server functions are called with the established base channel.

use std::net::Ipv4Addr;
use tokio::net::{TcpStream, TcpListener};
use remoc::prelude::*;

#[tokio::main]
async fn main() {
    // For demonstration we run both client and server in
    // the same process. In real life connect_client() and
    // connect_server() would run on different machines.
    futures::join!(connect_client(), connect_server());
}

// This would be run on the client.
async fn connect_client() {
    // Wait for server to be ready.
    tokio::time::sleep(std::time::Duration::from_secs(1)).await;

    // Establish TCP connection.
    let socket = TcpStream::connect((Ipv4Addr::LOCALHOST, 9875)).await.unwrap();
    let (socket_rx, socket_tx) = socket.into_split();

    // Establish Remoc connection over TCP.
    let (conn, tx, rx) =
        remoc::Connect::io(remoc::Cfg::default(), socket_rx, socket_tx).await.unwrap();
    tokio::spawn(conn);

    // Run client.
    client(tx, rx).await;
}

// This would be run on the server.
async fn connect_server() {
    // Listen for incoming TCP connection.
    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 9875)).await.unwrap();
    let (socket, _) = listener.accept().await.unwrap();
    let (socket_rx, socket_tx) = socket.into_split();

    // Establish Remoc connection over TCP.
    let (conn, tx, rx) =
        remoc::Connect::io(remoc::Cfg::default(), socket_rx, socket_tx).await.unwrap();
    tokio::spawn(conn);

    // Run server.
    server(tx, rx).await;
}

// This would be run on the client.
async fn client(mut tx: rch::base::Sender<u16>, mut rx: rch::base::Receiver<String>) {
    tx.send(1).await.unwrap();
    assert_eq!(rx.recv().await.unwrap(), Some("1".to_string()));
}

// This would be run on the server.
async fn server(mut tx: rch::base::Sender<String>, mut rx: rch::base::Receiver<u16>) {
    while let Some(number) = rx.recv().await.unwrap() {
        tx.send(number.to_string()).await.unwrap();
    }
}

Implementations

impl<'transport, TransportSinkError, TransportStreamError> Connect<'transport, TransportSinkError, TransportStreamError>

pub async fn framed<TransportSink, TransportStream, Tx, Rx, Codec>( cfg: Cfg, transport_sink: TransportSink, transport_stream: TransportStream ) -> Result<(Connect<'transport, TransportSinkError, TransportStreamError>, Sender<Tx, Codec>, Receiver<Rx, Codec>), ConnectError<TransportSinkError, TransportStreamError>>

where TransportSink: Sink<Bytes, Error = TransportSinkError> + Send + Sync + Unpin + 'transport, TransportSinkError: Error + Send + Sync + 'static, TransportStream: Stream<Item = Result<Bytes, TransportStreamError>> + Send + Sync + Unpin + 'transport, TransportStreamError: Error + Send + Sync + 'static, Tx: RemoteSend, Rx: RemoteSend, Codec: Codec,

Establishes a connection over a framed transport (a sink and a stream of binary data) and returns a remote sender and receiver.

This establishes a chmux connection over the transport and opens a remote channel.

You must poll the returned Connect future or spawn it for the connection to work.

Panics

Panics if the chmux configuration is invalid.

impl<'transport> Connect<'transport, Error, Error>

pub async fn io<Read, Write, Tx, Rx, Codec>( cfg: Cfg, input: Read, output: Write ) -> Result<(Connect<'transport, Error, Error>, Sender<Tx, Codec>, Receiver<Rx, Codec>), ConnectError<Error, Error>>

where Read: AsyncRead + Send + Sync + Unpin + 'transport, Write: AsyncWrite + Send + Sync + Unpin + 'transport, Tx: RemoteSend, Rx: RemoteSend, Codec: Codec,

Establishes a connection over an IO transport (an AsyncRead and AsyncWrite) and returns a remote sender and receiver.

A chmux connection is established over the transport and a remote channel is opened. This prepends a length header to each chmux packet for transportation over the unframed connection.

This method performs no buffering of read and writes and thus may exhibit suboptimal performance if the underlying reader and writer are unbuffered. In this case use io_buffered instead.

You must poll the returned Connect future or spawn it for the connection to work.

Panics

Panics if the chmux configuration is invalid.

pub async fn io_buffered<Read, Write, Tx, Rx, Codec>( cfg: Cfg, input: Read, output: Write, buffer: usize ) -> Result<(Connect<'transport, Error, Error>, Sender<Tx, Codec>, Receiver<Rx, Codec>), ConnectError<Error, Error>>

where Read: AsyncRead + Send + Sync + Unpin + 'transport, Write: AsyncWrite + Send + Sync + Unpin + 'transport, Tx: RemoteSend, Rx: RemoteSend, Codec: Codec,

Establishes a buffered connection over an IO transport (an AsyncRead and AsyncWrite) and returns a remote sender and receiver.

A chmux connection is established over the transport and a remote channel is opened. This prepends a length header to each chmux packet for transportation over the unframed connection.

This method performs internal buffering of reads and writes.

You must poll the returned Connect future or spawn it for the connection to work.

Panics

Panics if the chmux configuration is invalid.

Trait Implementations

impl<'transport, TransportSinkError, TransportStreamError> Future for Connect<'transport, TransportSinkError, TransportStreamError>

type Output = Result<(), ChMuxError<TransportSinkError, TransportStreamError>>

Result of connection after it has been terminated.

fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>

This future runs the dispatcher for this connection.