vld-lapin

vld-lapin wraps a lapin::Channel. publish validates a value and sends it as JSON; the decode helpers parse and validate deliveries and can settle invalid ("poison") messages for you.

Install

[dependencies]
vld = "0.5"
vld-lapin = "0.5"  # lapin 2 by default
lapin = "2"

Your lapin dependency must be the same major as the vld-lapin feature.

Features

Exactly one must be enabled:

vld-lapin = { version = "0.5", default-features = false, features = ["lapin-3"] }

MSRV: 1.85, 1.88 with lapin-4 (see MSRV).

Example

use vld_lapin::prelude::*;

vld::schema! {
    #[derive(Debug, serde::Serialize)]
    pub struct EventSchema {
        pub event: String => vld::string().min(1),
        pub retries: i64 => vld::number().int().min(0).max(10),
    }
}

async fn run() -> Result<(), Box<dyn std::error::Error>> {
    let conn = lapin::Connection::connect(
        "amqp://guest:guest@127.0.0.1:5672/%2f",
        lapin::ConnectionProperties::default(),
    )
    .await?;
    let channel = LapinChannel::new(conn.create_channel().await?);

    channel
        .queue_declare("events", Default::default(), Default::default())
        .await?;

    let event = EventSchema { event: "user.created".into(), retries: 0 };
    channel
        .publish("", "events", Default::default(), Default::default(), &event)
        .await?;

    if let Some(message) = channel.basic_get("events", Default::default()).await? {
        // Ack when valid, nack when invalid: every message is settled.
        let parsed: EventSchema = channel
            .ack_or_nack_decode(&message, Default::default(), Default::default())
            .await?;
        println!("{}", parsed.event);
    }
    Ok(())
}

API

• LapinChannel::new(channel), into_inner(); Deref / DerefMut to lapin::Channel for everything else.
• publish(exchange, routing_key, options, properties, &value) — validate, encode as JSON, publish and await the confirmation.
• queue_declare(name, options, args), basic_get(queue, options) — &str convenience wrappers.
• decode_bytes, decode_delivery, decode_get — parse and validate without settling.
• ack_or_nack_decode(delivery, ack, nack) — ack when valid, nack when invalid.
• ack_decode(delivery, ack) — ack when valid; an invalid message stays unsettled.
• nack_decode(delivery, nack) / reject_decode(delivery, reject) — settle only when invalid; a valid message stays unsettled for you to ack after processing.
• ack_decode_get, nack_decode_get, reject_decode_get — the same for basic_get messages.
• VldLapinError — Validation, Serialization, Deserialization, Lapin(lapin::Error).

Gotchas

• The value type is the schema: published values must be Serialize + VldParse (a schema! struct with #[derive(serde::Serialize)]).
• Pick the settle helper deliberately. With ack_decode, a poison message stays unacked: it holds a basic_qos prefetch slot and is redelivered (to fail again) when the channel closes. Use ack_or_nack_decode, nack_decode or reject_decode so invalid messages are dropped or dead-lettered (requeue: false).
• When settling an invalid message fails, the settle error is returned instead of the validation error.