actioncable-client 2.0.2

A client for Rails' Action Cable: one connection, many channel subscriptions, reconnects on its own
Documentation

actioncable-client

Rust client for Rails' Action Cable, one of the Action Cable clients this repository ships. It is the Go client's behavior on tokio: one connection, many channel subscriptions, reconnects on its own.

Installation

cargo add actioncable-client

The crate is actioncable-client and the library is actioncable:

use actioncable::{Client, Identifier};

The default websocket feature brings the built-in transport, tokio-tungstenite over rustls. Turn it off and the crate is the traits, the protocol and the client core, with no network dependency at all; you bring the transport with ClientBuilder::transport, as described under Swapping the transport, and build returns Error::NoTransport until you do:

actioncable-client = { version = "2.0.0", default-features = false }

Getting started

use actioncable::{Client, Identifier};
use serde::Deserialize;
use serde_json::json;

#[derive(Deserialize)]
struct Said {
    body: String,
}

async fn chat() -> Result<(), actioncable::Error> {
    // Establish a connection
    let client = Client::builder("wss://example.com/cable").build()?;
    client.connect().await?;

    // Subscribe to a channel
    let room = client
        .subscribe(Identifier::new("RoomChannel").param("id", 42))
        .await?;

    // Listen for incoming messages
    let reader = room.clone();
    tokio::spawn(async move {
        while let Some(message) = reader.next().await {
            if let Ok(said) = message.decode::<Said>() {
                println!("{}", said.body);
            }
        }
    });

    // Send messages
    room.perform("speak", json!({ "body": "Hello!" })).await?;

    room.unsubscribe().await?;
    client.close().await;
    Ok(())
}

build fails on a header that isn't valid HTTP, and on having no transport to dial with. Nothing touches the network until connect.

connect opens a connection and waits for the server to acknowledge. Failed attempts get retried automatically. Where the Go client takes a context, this one takes nothing: wrap the call in tokio::time::timeout to control how long to wait.

use std::time::Duration;
use actioncable::{Client, DisconnectReason, Error};

async fn connect(client: &Client) -> Result<(), Box<dyn std::error::Error>> {
    match tokio::time::timeout(Duration::from_secs(10), client.connect()).await {
        Ok(Ok(())) => Ok(()),
        Ok(Err(Error::Disconnected { reason: Some(DisconnectReason::Unauthorized), .. })) => {
            sign_in_again()
        }
        Ok(Err(error)) => Err(error.into()),
        Err(deadline) => {
            // The client is still dialing. Give up on it, or leave it running.
            let waiting_on = client.last_error();
            client.close().await;
            Err(Box::new(deadline))
        }
    }
}

fn sign_in_again() -> Result<(), Box<dyn std::error::Error>> {
    Ok(())
}

Dropping the connect future bounds the wait, not the client: it keeps dialing until close. That is where this parts from the Go client, whose Connect stops the client when its context ends — a dropped future in Rust is how a select! loses a race, and killing the connection over that would surprise everyone. last_error says what the client is failing on meanwhile, so a credential that can't be built doesn't hide behind a deadline.

The errors worth handling:

  • Error::Disconnected, when the server sends a disconnect message that says not to come back. reason is the DisconnectReason it gave, or None when it gave none:
    • Unauthorized — authentication or authorization failed.
    • InvalidRequest — the request wasn't a valid Action Cable upgrade.
    • ServerRestart — the Rails server is restarting.
    • Remote — the app closed this connection with ActionCable.server.disconnect.
    • Other(String) — anything a future Rails might add.
  • Error::UnsupportedSubprotocol, when the server picks a protocol this client doesn't speak, or picks none at all.
  • Error::GaveUp, when max_attempts is set and that many attempts failed in a row. It carries the last attempt's error.

A disconnect message says whether the client should re-connect. Only the ones that say no return an error. The rest are logged through tracing and retried, so a server restart shows up in the log and the connection returns on its own. The Error docs list which variants each method can return.

A client that stopped for good stays stopped. connect and subscribe return the error that stopped it from then on, or Error::Closed after close. Calling connect on a client that is already running returns Error::AlreadyConnected.

A client that stopped later on — the server hung up for good, or it ran out of attempts — says so through done and err:

# use actioncable::Client;
# async fn watch(client: &Client) {
client.done().await;
println!("cable stopped: {:?}", client.err());
# }

err is None while the client runs, and afterwards one of the errors above or Error::Closed. A stopped client doesn't come back; make a new one.

subscribe takes an Identifier, or a SubscribeRequest when there are callbacks to attach, sends the subscription and waits for the channel to confirm it. It returns Error::Rejected when the channel's subscribed method rejects it, and Error::NotConnected before connect. Dropping the future, from a timeout say, forgets the subscription and, when nothing else holds the identifier, tells the server to drop it too; a confirmation that arrives later goes nowhere.

Subscribing twice to the same identifier gives two subscriptions that each get every message. Rails holds one subscription per identifier per connection and ignores a second subscribe for it, so the client asks once: a second subscribe joins the first, confirmed at once if the first already was, and sharing its verdict if it is still waiting. The server hears about the unsubscribe from the last of them.

Subscription::next returns None once the subscription is unsubscribed, rejected, or the client is closed, so a while let loop over it ends on its own. Subscription::err says which it was: Error::Unsubscribed, Error::Rejected, or whatever stopped the client.

# use actioncable::{Error, Subscription};
# async fn read(room: Subscription) {
while let Some(message) = room.next().await {
    println!("{message}");
}
if !matches!(room.err(), Some(Error::Unsubscribed)) {
    println!("subscription ended: {:?}", room.err());
}
# }

Handles are cheap to clone and share one stream: a message goes to whichever handle reads it first. Dropping the last handle unsubscribes too, when it happens on a tokio runtime; elsewhere it stops delivery, warns, and leaves the server holding the subscription until the connection ends.

Read it promptly. A subscription buffers 64 messages, and a message that arrives while the buffer is full gets dropped and logged rather than stalling the connection. Set a bigger buffer if the reader can't keep up with a burst:

# use actioncable::Client;
let client = Client::builder("wss://example.com/cable")
    .message_buffer(1000)
    .build();

The buffer size applies to every subscription on the client.

perform names an action and takes data that encodes to a JSON object, since that is the only shape Rails routes to an action; anything else is Error::DataNotAnObject. Pass () to send the action alone. send delivers data as-is, without an action, to the channel's receive method.

Actions don't come back after a reconnect. perform and send return Error::NotConnected while the connection is down, or up but not yet welcomed, since Rails discards anything that arrives that early. A write that fails partway returns Error::Transport and drops the connection for a redial, since the socket is in doubt. Send it again if it matters.

Authorizing the connection

Action Cable servers can authorize connections using cookies or headers.

Use cookie to set a cookie when establishing a connection:

# use actioncable::Client;
let client = Client::builder("wss://example.com/cable")
    .cookie("_session_id=...")
    .build();

header sets any other header:

# use actioncable::Client;
let client = Client::builder("wss://example.com/cable")
    .header("X-Api-Token", "...")
    .build();

A header that isn't valid HTTP, one carrying a newline say, fails build with Error::InvalidHeader. That is where the Go client's header-injection guard lives here: the value is an http::HeaderValue from the moment it is given, so there is nothing to neutralize later.

A credential that expires should come from a HeaderProvider, which is asked on every dial, the first included. A closure returning a future is one:

use actioncable::{Client, Error};
use http::{HeaderMap, HeaderValue};

async fn access_token() -> Result<String, std::io::Error> {
    Ok("token".to_string())
}

let client = Client::builder("wss://example.com/cable")
    .header_provider(|| async {
        let token = access_token().await.map_err(Error::headers)?;
        let mut headers = HeaderMap::new();
        headers.insert(
            "Authorization",
            HeaderValue::from_str(&format!("Bearer {token}")).expect("a token is printable"),
        );
        Ok(headers)
    })
    .build();

So is a credentials type that implements the trait, which is how one that already knows how to refresh itself plugs in:

use actioncable::{Client, Error, HeaderProvider};
use async_trait::async_trait;
use http::{HeaderMap, HeaderValue};

struct Credentials;

impl Credentials {
    async fn access_token(&self) -> Result<String, std::io::Error> {
        Ok("token".to_string())
    }
}

#[async_trait]
impl HeaderProvider for Credentials {
    async fn headers(&self) -> Result<HeaderMap, Error> {
        let token = self.access_token().await.map_err(Error::headers)?;
        let mut headers = HeaderMap::new();
        headers.insert(
            "Authorization",
            HeaderValue::from_str(&format!("Bearer {token}")).expect("a token is printable"),
        );
        Ok(headers)
    }
}

let client = Client::builder("wss://example.com/cable")
    .header_provider(Credentials)
    .build();

What it returns is merged over the headers already set, so an Origin or an API token given with header is kept and a header the provider also names is replaced. An error, wrapped with Error::headers, turns down that dial, and the client tries again on its backoff.

When an application can identify an error that another connection attempt cannot repair, use stop_on_error. The client stops with that error, while other connection failures continue to retry:

# use actioncable::Client;
# #[derive(Debug, thiserror::Error)]
# #[error("sign in again")]
# struct SignedOut;
let client = Client::builder("wss://example.com/cable")
    .stop_on_error(|error| error.cause().is_some_and(|cause| cause.is::<SignedOut>()));

Error::cause hands back the error the transport or the header provider wrapped, for downcast_ref to recognize. It is there rather than left to std::error::Error::source because the wrapped error is held behind an Arc, and a source walk hands back the Arc rather than what is in it.

Rails also checks the Origin header and rejects a request that doesn't carry one. By default, the Origin is set to the server's URL, so wss://example.com/cable sends https://example.com.

Set it explicitly when the server sees a different scheme or host than the URL says, behind a proxy that terminates TLS for instance:

# use actioncable::Client;
let client = Client::builder("wss://example.com/cable")
    .origin("http://example.com")
    .build();

The built-in transport adds User-Agent: actioncable/<version> unless you set one. The headers the upgrade handshake itself needs are its own and can't be overridden.

Callbacks

Some channels only send what's new, so a reconnect can leave a gap. Only the client knows a reconnect happened, so a SubscribeRequest takes callbacks for the connection events:

use actioncable::{Client, Identifier, SubscribeRequest};

async fn watch(client: &Client) -> Result<(), actioncable::Error> {
    let room = client
        .subscribe(
            SubscribeRequest::new(Identifier::new("RoomChannel"))
                .on_connected(|reconnected| async move {
                    if reconnected {
                        catch_up().await;
                    }
                })
                .on_disconnected(|will_reconnect| async move {
                    println!("gone, back soon: {will_reconnect}");
                })
                .on_rejected(|| async { println!("turned away") }),
        )
        .await?;
    Ok(())
}

async fn catch_up() {}

on_connected runs every time the server confirms the subscription. reconnected is false the first time and true every time after.

on_disconnected runs when the connection drops. will_reconnect says whether the client is coming back or has stopped for good.

on_rejected runs when the channel rejects the subscription.

Callbacks are async: each returns a future, and |x| async move { .. } is how a plain closure becomes one. They run on their own tokio task, one at a time, in order, never on the connection's task, and the next event waits for the current callback to finish. close, subscribe and unsubscribe can all be awaited from inside one. The message stream ends only after the last of them has returned, so a while let loop that ends knows no callback is still running or about to.

The same events also arrive as a stream of Events on every subscription, next to its messages, for code that would rather read than be called:

use actioncable::{Event, Subscription};

async fn watch(room: Subscription) {
    while let Some(event) = room.next_event().await {
        match event {
            Event::Connected { reconnected: true } => catch_up(),
            Event::Connected { reconnected: false } => {}
            Event::Disconnected { will_reconnect } => println!("gone, back soon: {will_reconnect}"),
            Event::Rejected => break,
        }
    }
}

fn catch_up() {}

Events arrive in the order they happened. Callbacks and the stream see the same events; neither takes from the other. A handle keeps the last sixteen events for a reader that isn't listening, so a handle that never reads them costs nothing however long the client runs; a reader further behind than that skips the oldest and is told so in the log. A rejection of the first subscribe is also the Error::Rejected that subscribe returns, so the stream only ever sees Rejected after a reconnect. It is the last event: the client has already let go of the subscription.

Staying connected

Rails sends a ping every three seconds and the client watches for it. After six seconds of silence the client treats the connection as dead, drops it, and dials again after a second, then two, then four, up to thirty. Each delay is spread between half and the whole of that, so a restarted server doesn't get every client back at once.

All of it is configurable, and the retrying can be capped:

# use std::time::Duration;
# use actioncable::Client;
let client = Client::builder("wss://example.com/cable")
    .stale_after(Duration::from_secs(10))
    .backoff(Duration::from_secs(1), Duration::from_secs(30))
    .max_attempts(10)
    .build();

By default the client keeps dialing until close. With max_attempts it stops with Error::GaveUp after that many failures in a row; a welcome resets the count, so it bounds one outage rather than the client's lifetime.

Subscriptions come back on their own. The client resubscribes all of them on the new connection, then resends a subscribe every half second until the server confirms it, because a subscribe that arrives before the connection is set up gets dropped. subscribe_retry changes the half second. The same Subscription keeps working throughout.

Swapping the transport

The client speaks over tokio-tungstenite with rustls by default, trusting the platform's root certificates. That is WebSocketTransport, behind the websocket Cargo feature. It bounds the TCP connect, TLS handshake and upgrade together at ten seconds, bounds each write at ten seconds so a peer that stopped reading can't hold a perform forever, and refuses messages over 8 MiB; all three are settable:

# use std::time::Duration;
# use actioncable::{Client, WebSocketTransport};
let client = Client::builder("wss://example.com/cable")
    .transport(
        WebSocketTransport::new()
            .handshake_timeout(Duration::from_secs(5))
            .write_timeout(Duration::from_secs(5)),
    )
    .build();

Three of its failures are typed, for a caller that wants to act on them rather than read them:

  • Error::Handshake, when the server answers the upgrade with anything but 101. status tells a redirect from a refusal, and prints as the whole status line.
  • Error::Close, from read when the server sends a close frame, with its code and reason.
  • Error::MessageTooBig, from read when a message is larger than max_message_size. It is refused as soon as its length is known, before any of it is read in.

An application that already uses a WebSocket library can keep using it by implementing two traits.

Transport has one function:

#[async_trait]
pub trait Transport: Send + Sync {
    async fn dial(&self, url: &str, options: DialOptions) -> Result<Box<dyn Conn>, Error>;
}

dial opens one connection. options carries the subprotocols the client's protocols negotiate under, and the headers that authorize the request.

Conn has four, and a fifth with a default:

#[async_trait]
pub trait Conn: Send + Sync {
    fn subprotocol(&self) -> Option<&str>;
    async fn read(&self) -> Result<Vec<u8>, Error>;
    async fn write(&self, payload: &[u8]) -> Result<(), Error>;
    async fn close(&self);
    async fn close_with_status(&self, code: u16, reason: &str) { self.close().await }
}

subprotocol returns the subprotocol the server picked, None if it picked none. read returns the next complete message. write sends one text message. close hangs up, and has to interrupt a read running at the time. The client reads from one task and writes from another, one write at a time, so the two halves need their own locks. Error::transport wraps whatever the library reports.

close_with_status is where the Go client's optional StatusCloser went: a Rust caller can always call it, and the default forwards to close, so a transport whose library can't say why hangs up all the same. The built-in one sends the code and as much of the reason as a control frame has room for.

Implement both and hand the transport to the client. ClientBuilder::transport works with or without the websocket feature: with it, it replaces the built-in transport; without it, it is the only way to build a client, and build returns Error::NoTransport until it is called:

let client = Client::builder("wss://example.com/cable")
    .transport(MyTransport::new())
    .build()?;

The tests do exactly this with an in-memory transport, FakeTransport in actioncable::test_support. It is behind the test-support Cargo feature, so a crate built on this client can drive its own tests through the same fake by turning the feature on in its dev-dependency; a test accepts the connection the client dialed and plays the server on it, welcome, confirmations and broadcasts included:

[dev-dependencies]
actioncable-client = { version = "2.0.0", features = ["test-support"] }

Arc<T> is a Transport whenever T is, and a HeaderProvider whenever T is one, so one of either can be kept and handed to every client an application builds — which is also how a fake gets into code that would otherwise build the real one.

Adding protocols

Action Cable servers can talk multiple protocols. Rails' default is V1-JSON and that's what's supported out-of-the-box. But, if needed, new protocols can be added.

The Protocol trait has just three functions:

pub trait Protocol: Send + Sync {
    fn subprotocol(&self) -> &str;
    fn encode(&self, command: &Command) -> Result<Vec<u8>, Error>;
    fn decode(&self, payload: &[u8]) -> Result<Incoming, Error>;
}

subprotocol returns the WebSocket subprotocol for the protocol. encode serializes a Command to the protocol's wire format, while decode does the opposite, into an Incoming whose Kind says what the server sent.

All protocols are offered to the server in that order, followed by the actioncable-unsupported sentinel so a server that speaks none of them can say so. If one protocol is preferred over another then it should be listed first:

let client = Client::builder(url)
    .protocols(vec![Arc::new(V2MessagePack), Arc::new(V1Json)])
    .build()?;

prefer_protocols is this client's spelling of the Go client's WithAdditionalProtocols: a shorthand for adding new protocols to the default list. These protocols get prepended to the list of supported protocols, which means that they'll be preferred:

let client = Client::builder(url)
    .prefer_protocols(vec![Arc::new(V2MessagePack)])
    .build()?;

The default is V1Json, which speaks actioncable-v1-json, Rails' default protocol. A list emptied with protocols(vec![]) leaves nothing to offer, and connect returns Error::NoProtocols.

Where this differs from the Go client

The behavior is the Go client's. What the language moved:

  • No context. connect and subscribe wait; tokio::time::timeout bounds them, and dropping the future cancels the wait rather than the client. unsubscribe, perform and send take nothing either, so a teardown has nothing to hold on to.
  • No logger. The client's chatter goes to tracing, which is where a Rust application already looks, instead of a WithLogger of its own.
  • Error is one enum, not a set of sentinels and struct errors. Error::cause is how an application recognizes the error it handed over, since the chain holds it behind an Arc.
  • StatusCloser is a defaulted method on Conn, because Rust has no way to ask a trait object whether it also implements something else.
  • Header injection can't happen. A header is an http::HeaderValue as soon as it is given, so the Go client's escaping on the way out is a build failure here instead.
  • Handles are clones, not separate subscriptions. Cloning a Subscription shares one stream; subscribing twice to the same identifier is what gives two.

Each of those has a test under tests/ asserting the Rust behavior, next to the Go test it came from.

Development

make check   # what CI runs: fmt, clippy, tests, docs, cargo deny
make test    # the tests alone

License

Released under the MIT License. See LICENSE.