Names

A subscription by name, #[subscriber("emails")], reads the table its route leads to; it takes the broker's settings and no retry cap.

A route, SqlxBroker::route::<Row>("emails"), leads a name into the table of a row type in both directions. A publish to the name writes a row through the row's Publish, and a subscription by the name reads that table. On a table with a group column the name selects its group, as an InboxQueue of the same name would. A name without a route stops the service at startup. A route leads only into a table in payload mode, so a subscription by name reads payload tables. It takes the broker's poll interval and lease, and refuses max_attempts(..) and dead_letter(..) at startup: an InboxQueue takes those. Where the route's row leaves every event to the crate and holds column types the crate reads itself, the subscription reads the rows by those columns, with no box and no dynamic call per message, in every form. Any other row runs its own code at one boxed delivery and one boxed settlement per message (NamedSubscriber). A subscription by name needs a dialect that implements ByName for its database, as every built-in dialect does.

Example

Both names lead into one table: a signup read by name is answered by a welcome, which becomes a row of the welcome group that a second subscription by name reads.

Macros
Manual

examples/sqlx_subscriptions_names.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;

/// One row of `jobs`: the group column holds the name a row was published to.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "jobs")]
struct Job {
    #[field(id, generated)]
    id: i64,
    #[field(group)]
    name: String,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[subscriber("signups", reply)]
async fn greet(signup: &Signup) -> Welcome {
    Welcome {
        email: signup.email.clone(),
    }
}

#[subscriber("welcome")]
async fn send_welcome(welcome: &Welcome) -> HandlerOutcome {
    tracing::info!(to = %welcome.email, "sending the welcome email");
    HandlerOutcome::ack()
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    // Both names lead into `jobs`, in both directions: a publish writes a row of the name's
    // group, and a subscription by the name reads that group.
    let broker = broker.route::<Job>("signups").route::<Job>("welcome");
    RustStream::new(AppInfo::new("signup", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker, |b| {
            b.include(greet);
            b.include(send_welcome);
        })
}

examples/manual/sqlx_subscriptions_names.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;

/// One row of `jobs`: the group column holds the name a row was published to.
#[derive(sqlx::FromRow)]
struct Job {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for Job {
    type Id = i64;
    type Table = InboxSpec<(Lease<DateTime<Utc>>, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("jobs", Column::new("id").generated())
        .group(Column::new("name"))
        .lease(Column::new("locked_until"))
        .payload(Column::new("payload"));

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

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

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

struct Greet;

impl Handle<Signup, Welcome> for Greet {
    fn handle(
        &self,
        signup: &Signup,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Welcome, HandlerOutcome>> {
        ready(Ok(Welcome {
            email: signup.email.clone(),
        }))
    }
}

struct SendWelcome;

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

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    // Both names lead into `jobs`, in both directions: a publish writes a row of the name's
    // group, and a subscription by the name reads that group.
    let broker = broker.route::<Job>("signups").route::<Job>("welcome");
    RustStream::new(AppInfo::new("signup", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker, |b| {
            b.include(subscriber("signups", Greet).reply().build());
            b.include(subscriber("welcome", SendWelcome).build());
        })
}

Example​

Example