AsyncAPI

With the asyncapi feature the document names the database server and, on each subscription's channel, the table it reads and its group; rows get their schema.

The core builds the AsyncAPI document from the app, before anything connects. This crate adds what it knows about the parts the core describes. A server reports the database the pool reaches: its host and port and the protocol, never a credential. A channel of an InboxQueue subscription reports the table it reads under the extension x-ruststream-sqlx, with the group its name selects. Everything comes from the service's own declarations: the pool's options, the table's description and the subscription's name.

A message a handler decodes is documented by its type's JsonSchema, as on every broker. A handler in row mode takes the row itself, so the row's JsonSchema documents the message: the derive names schemars through the core, #[schemars(crate = "ruststream::schemars")]. A broker made with from_url builds the document with no database in reach and no runtime.

Example

The dependency line:

ruststream-sqlx = { version = "0.7", features = ["inbox", "postgres", "asyncapi"] }

The service registers its broker under the label postgres, the key of its server in the document. The tests build the document with build_spec and read it as JSON.

Macros
Manual

examples/sqlx_asyncapi.rs
fn service(url: &str) -> RustStream {
    RustStream::new(AppInfo::new("office", "0.1.0")).with_broker_labeled(
        "postgres",
        broker(url),
        |b| {
            b.include(send_email);
            b.include(render);
            b.include(send_mail);
        },
    )
}

#[ruststream::app]
fn app() -> impl App {
    let url = env::var("DATABASE_URL").unwrap_or_else(|_| "postgres://localhost/office".to_owned());
    service(&url)
}

examples/manual/sqlx_asyncapi.rs
fn service(url: &str) -> RustStream {
    RustStream::new(AppInfo::new("office", "0.1.0")).with_broker_labeled(
        "postgres",
        broker(url),
        |b| {
            let email =
                subscriber("emails", Mailer).map_source(|_| InboxQueue::<SendEmail>::new("emails"));
            b.include(email.build());
            let report = subscriber("reports", Renderer)
                .map_source(|_| InboxQueue::<RenderReport>::new("reports"));
            b.include(report.build());
            let mail = subscriber("mail", Postman).map_source(|_| InboxQueue::<Mail>::new("mail"));
            b.include(mail.build());
        },
    )
}

fn main() -> ExitCode {
    let url = env::var("DATABASE_URL").unwrap_or_else(|_| "postgres://localhost/office".to_owned());
    run_main(|| service(&url))
}

The mail handler takes the row of mail_jobs itself, and the message it reads is documented by the row's columns.

Macros
Manual

examples/sqlx_asyncapi.rs
/// A mail the handler takes as the row itself: its `JsonSchema` documents the message.
#[derive(Debug, Clone, Inbox, sqlx::FromRow, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[inbox(table = "mail_jobs")]
struct Mail {
    #[field(id, generated)]
    id: i64,
    #[field(group)]
    name: String,
    recipient: String,
    subject: Option<String>,
}

#[subscriber(InboxQueue::<Mail>::new("mail"))]
async fn send_mail(mail: &Mail) -> HandlerOutcome {
    tracing::info!(to = %mail.recipient, subject = ?mail.subject, "mailing");
    HandlerOutcome::ack()
}

examples/manual/sqlx_asyncapi.rs
use ruststream::runtime::SoloCarried;

/// A mail the handler takes as the row itself: its `JsonSchema` documents the message.
#[derive(Debug, Clone, sqlx::FromRow, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Mail {
    id: i64,
    recipient: String,
    subject: Option<String>,
}

impl InboxTable for Mail {
    type Id = i64;
    type Table = InboxSpec<()>;
    const TABLE: Self::Table = InboxSpec::new("mail_jobs", Column::new("id").generated())
        .group(Column::new("name"))
        .data(&[Column::new("recipient"), Column::new("subject")]);

    fn id(&self) -> &i64 {
        &self.id
    }
}

// Row mode: a handler takes the row itself.
impl Input for Mail {
    type Axis = SoloCarried<Self>;
}

struct Postman;

impl Handle<Mail> for Postman {
    fn handle(
        &self,
        mail: &Mail,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(to = %mail.recipient, subject = ?mail.subject, "mailing");
        ready(Ok(()))
    }
}

Example​

Example