Errors
MqttError is the one error type of the crate: the connect, a subscription, a delivery, a publish, a closed connection and an invalid setting.
Each variant names the step that failed.
The client's errors become text inside it, so the rumqttc types stay out of the crate's API.
| Variant | When |
|---|---|
Connect | the server refused the connection, the TLS handshake failed, or no CONNACK came within 30 seconds |
Subscribe | the server rejected a filter, or the connection ended while a subscription opened |
Receive | the connection ended for good: the server refused a reconnect for a reason no retry changes |
Publish | the topic is not one a server takes (empty, with a wildcard or a control character), a header has no MQTT form, or the connection task is gone |
NotConnected | a publisher was used, or a subscription opened, after the broker shut down |
Invalid | a setting or a descriptor cannot work, such as a keep-alive under five seconds or a share group with a / |
A service meets the error in two places.
A failed startup carries it as the source of the runtime's error, and the service exits before it reads anything.
A publish from a handler returns it inside the core's PublishError, and the handler decides how the delivery settles.
The enum is #[non_exhaustive], so a match on it ends with a wildcard arm.
Example
An alert goes to the topic of its device. A device id with a wildcard makes no topic, so the handler logs the refusal and drops the reading: a retry would fail the same way.
- Macros
- Manual
#[subscriber("broker-errors/telemetry")]
async fn check(telemetry: &Telemetry, Out(alerts): Out<impl Publisher>) -> HandlerOutcome {
if telemetry.temperature <= 30.0 {
return HandlerOutcome::ack();
}
let alert = Alert {
temperature: telemetry.temperature,
};
let topic = format!("alerts/{}", telemetry.device);
if let Err(err) = alerts.message(&alert).to(&topic).publish().await {
// A device id with `+` or `#` makes no topic, and no retry changes that.
tracing::warn!(
subscription = "broker-errors/telemetry",
device = telemetry.device,
error = %err,
"alert not published"
);
return HandlerOutcome::drop();
}
HandlerOutcome::ack()
}
struct Check;
impl<Entry> Handle<Telemetry, (), Outs<(Entry,)>> for Check
where
Entry: OutEntry<DefaultSlot, Wire: Publisher>,
{
async fn handle(
&self,
telemetry: &Telemetry,
outs: &Outs<(Entry,)>,
_ctx: &mut Context<'_>,
) -> Result<(), HandlerOutcome> {
if telemetry.temperature <= 30.0 {
return Ok(());
}
let alert = Alert {
temperature: telemetry.temperature,
};
let topic = format!("alerts/{}", telemetry.device);
if let Err(err) = outs
.get(DefaultSlot)
.message(&alert)
.to(&topic)
.publish()
.await
{
// A device id with `+` or `#` makes no topic, and no retry changes that.
tracing::warn!(
subscription = "broker-errors/telemetry",
device = telemetry.device,
error = %err,
"alert not published"
);
return Err(HandlerOutcome::drop());
}
Ok(())
}
}