Channels

Each InboxQueue channel carries x-ruststream-sqlx: the table it reads, qualified with its schema, and the group its name selects where the table has groups.

The extension holds two fields. table is the table's name, with its schema in front where the description names one (app.email_jobs). group is the subscription's name, present only when the table has a group column: in a table without groups the name addresses the whole table. A channel the service publishes to and a subscription by name carry no extension: a name leads into a table through a route, which the channel does not describe.

Example

The channel emails reports { "table": "app.email_jobs", "group": "emails" }, and the channel reports reports { "table": "reports" }.

Macros

examples/sqlx_asyncapi.rs

/// One row of `app.email_jobs`: the channel of each subscription names the table and its group.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs", schema = "app")]
struct SendEmail {
    #[field(id, generated)]
    id: i64,
    #[field(group)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
}

/// One row of `reports`, a table without groups: its channel names the table alone.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "reports")]
struct RenderReport {
    #[field(id, generated)]
    id: i64,
    #[field(payload)]
    payload: Vec<u8>,
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send_email(email: &Email) -> HandlerOutcome {
    tracing::info!(to = %email.to, "sending");
    HandlerOutcome::ack()
}

#[subscriber(InboxQueue::<RenderReport>::new("reports"))]
async fn render(report: &Report) -> HandlerOutcome {
    tracing::info!(month = report.month, "rendering");
    HandlerOutcome::ack()
}

Manual

examples/manual/sqlx_asyncapi.rs

use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::spec::Payload;
use ruststream_sqlx::{InboxSpec, InboxTable};

/// One row of `app.email_jobs`: the channel of each subscription names the table and its group.
#[derive(sqlx::FromRow)]
struct SendEmail {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for SendEmail {
    type Id = i64;
    type Table = InboxSpec<(Payload,)>;
    const TABLE: Self::Table = InboxSpec::new("email_jobs", Column::new("id").generated())
        .within("app")
        .group(Column::new("name"))
        .payload(Column::new("payload"));

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

impl PayloadRow for SendEmail {
    type Column = Vec<u8>;

    fn payload(&self) -> &[u8] {
        &self.payload
    }
}

/// One row of `reports`, a table without groups: its channel names the table alone.
#[derive(sqlx::FromRow)]
struct RenderReport {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for RenderReport {
    type Id = i64;
    type Table = InboxSpec<(Payload,)>;
    const TABLE: Self::Table =
        InboxSpec::new("reports", Column::new("id").generated()).payload(Column::new("payload"));

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

impl PayloadRow for RenderReport {
    type Column = Vec<u8>;

    fn payload(&self) -> &[u8] {
        &self.payload
    }
}

struct Mailer;

impl Handle<Email> for Mailer {
    fn handle(
        &self,
        email: &Email,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(to = %email.to, "sending");
        ready(Ok(()))
    }
}

struct Renderer;

impl Handle<Report> for Renderer {
    fn handle(
        &self,
        report: &Report,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(month = report.month, "rendering");
        ready(Ok(()))
    }
}