Skip to main content

Connecting to production

A managed MQTT service, such as AWS IoT Core, HiveMQ Cloud or EMQX Cloud, asks for credentials and TLS, often with a client certificate. This guide configures the broker for one, from the deployment's environment.

Credentials and TLS​

src/main.rs
/// Where the server is and who the service is, as the deployment provides them.
struct Deployment {
    url: String,
    client_id: String,
    username: String,
    password: String,
    ca: Vec<u8>,
    cert: Vec<u8>,
    key: Vec<u8>,
}

impl Deployment {
    fn from_env() -> Self {
        let read = |name: &str| env::var(name).unwrap_or_default();
        let file = |name: &str| fs::read(read(name)).unwrap_or_default();
        Self {
            url: read("MQTT_URL"),
            client_id: read("MQTT_CLIENT_ID"),
            username: read("MQTT_USERNAME"),
            password: read("MQTT_PASSWORD"),
            ca: file("MQTT_CA"),
            cert: file("MQTT_CERT"),
            key: file("MQTT_KEY"),
        }
    }
}

const LARGEST_PACKET: NonZeroU32 = NonZeroU32::new(256 * 1024).unwrap();
const IN_FLIGHT: NonZeroU16 = NonZeroU16::new(200).unwrap();

fn broker(deployment: Deployment) -> MqttBroker {
    MqttBroker::new(deployment.url, deployment.client_id)
        .credentials(deployment.username, deployment.password)
        .tls_ca(deployment.ca)
        .tls_client_auth(deployment.cert, deployment.key)
        .keep_alive(Duration::from_secs(30))
        .max_packet_size(LARGEST_PACKET)
        .receive_maximum(IN_FLIGHT)
}

The deployment provides every setting in environment variables, and the certificates as files. credentials sets the user name and the password of the CONNECT packet. A login inside the URL, such as mqtt://user:p%40ss@host, works too and is percent-decoded; credentials wins over it. They stay out of the generated AsyncAPI document, which reports only the host and port the service connects to.

mqtts:// selects TLS, on port 8883 when the URL names none. tls_ca is the certificate authority, in PEM, that the client verifies the server against. The client trusts that authority and no other store, so a mqtts:// URL without tls_ca fails the start, naming tls_ca. tls_client_auth adds the client's certificate and key, for a service that authenticates clients by certificate.

One client id per process​

The client id is the second argument of MqttBroker::new, and the server keys the session by it. A second connection with the same id takes the session over and closes the first one. So two replicas of a service never share an id: here each takes its own from MQTT_CLIENT_ID, for example the pod's name. Replicas that split the work join a share group instead.

Limits​

  • keep_alive is how often the client proves it is alive on an idle connection; the server closes a connection silent for one and a half times as long. The rumqttc client's minimum is 5 seconds, and its default is 60.
  • max_packet_size is the largest packet the service accepts; the server does not send it a larger one. The broker's default is 1 MiB, above the client's own default of 10 KiB.
  • receive_maximum bounds the QoS 1 and 2 deliveries in flight (Delivery guarantees).

The service​

src/main.rs
#[subscriber(MqttFilter::new("devices/+/telemetry").qos(Qos::AtLeastOnce).shared("ingest"))]
async fn record(telemetry: &Telemetry) -> HandlerOutcome {
    tracing::info!(telemetry.device, telemetry.temperature, "reading recorded");
    HandlerOutcome::ack()
}

fn service(broker: MqttBroker) -> RustStream {
    RustStream::new(AppInfo::new("telemetry", "0.1.0")).with_broker(broker, |b| {
        b.include(record)
            .out_retry(Publish::default())
            .to("devices/retry/telemetry");
    })
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(broker(Deployment::from_env()))
}

The broker connects once the service starts. A lost connection reconnects with a backoff that starts at 100 ms and doubles up to 5 seconds, and the broker subscribes again when the server reports the session gone. A refusal the connection will not outgrow ends the connection task instead: bad credentials, a client id the server does not accept, or a TLS handshake the certificates cannot satisfy. Every subscription then returns that error.

Testing the configuration​

The tests start the production service with the production broker, given test settings, under TestApp::start. The in-process server reads the broker's configuration and refuses what the client would refuse, without dialling the server:

  • a keep-alive under 5 seconds fails the start with the client's minimum in the message;
  • a mqtts:// URL without tls_ca fails the start, naming tls_ca.

The credentials and the TLS handshake are checked by a real server only. The live tests of the Security page run both against Mosquitto listeners that demand them.