Connecting to a production cluster

A production cluster asks for TLS and authentication, and a service decides how long it waits for the cluster and how large a message it sends.

TLS and a token

src/main.rs

/// Where the cluster is and who the service is, as the deployment provides them.
struct Cluster {
    url: String,
    token: String,
}

impl Cluster {
    fn from_env() -> Self {
        let read = |name: &str| env::var(name).unwrap_or_default();
        Self {
            url: read("PULSAR_URL"),
            token: read("PULSAR_TOKEN"),
        }
    }
}

fn broker(cluster: Cluster) -> PulsarBroker {
    // Ten tries, three seconds apart; then the error.
    let retries = OperationRetries::attempts(nonzero!(10u32)).delay(Duration::from_secs(3));
    PulsarBroker::new(cluster.url)
        .token(cluster.token)
        .operation_retries(retries)
        // The cluster runs with `maxMessageSize=10485760`.
        .max_message_size(nonzero!(10_485_760usize))
}

The deployment provides the service URL and the token in environment variables. A Pulsar client dials one service URL. A cluster of several brokers is reached through one entry point in front of them: a DNS name, a load balancer or the Pulsar proxy.

A pulsar+ssl:// URL dials the cluster's TLS listener, port 6651 by convention. The client checks the server's certificate against the system's trusted roots, and its host name. token authenticates with a JWT, which the client sends when it connects and when it reconnects. StreamNative Cloud and a self-hosted cluster with token authentication both take this pair.

Retries and the message size

The client reconnects on its own after a broker restart, and asks again for any operation the cluster has not accepted yet. By default it asks for ever, five seconds apart, which waits out a restart of any length. OperationRetries bounds that: ten tries three seconds apart, and then the operation fails with the cluster's answer. The bound covers the lookups, the producers and the reconnect of every consumer. It also turns a second consumer of an Exclusive subscription into an error at startup instead of a wait.

max_message_size is the largest payload a publish may carry, five mebibytes by default as on the server. A deployment that changed the server's maxMessageSize sets the same value here. A larger publish fails in the service, before it reaches the cluster.

The service

Macros

src/main.rs

#[subscriber(PulsarSubscription::new("orders", "orders-svc"), reply)]
async fn confirm(order: &Order) -> Confirmation {
    tracing::info!(order.id, "order confirmed");
    Confirmation { id: order.id }
}

fn service(broker: PulsarBroker) -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(confirm);
    })
}

#[ruststream::app]
fn app() -> impl App {
    service(broker(Cluster::from_env()))
}

Manual

src/main.rs

struct Confirm;

impl Handle<Order, Confirmation> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
        tracing::info!(order.id, "order confirmed");
        ready(Ok(Confirmation { id: order.id }))
    }
}

fn service(broker: PulsarBroker) -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber(PulsarSubscription::new("orders", "orders-svc"), Confirm)
                .reply()
                .build(),
        );
    })
}

fn app() -> RustStream {
    service(broker(Cluster::from_env()))
}

fn main() -> ExitCode {
    run_main(app)
}

Every setting of the broker is a builder step that dials nothing, so #[ruststream::app] builds the app as it does locally. The connection opens when the service starts. The generated AsyncAPI document reports the host and the port of the URL, and never the token. A service that replays its topics also needs the namespace's retention policy on the cluster.

Testing the configuration

The tests start the production service with the production broker, given a test token, under TestApp::start. The in-process Pulsar checks the service URL as the client parses it, without dialling the cluster. A URL with another scheme fails the start, and the error names the schemes the client dials:

the service URL's scheme "https" is not one the client dials: pulsar or pulsar+ssl

TLS, the token and the retry bound take effect only on a real cluster.