Source

A subscriber reads a subscription named by a string or by the broker's descriptor, fixed in the attribute or left to the mount site.

The attribute fixes the kind of subscription and may leave its value to the mount site:

Attribute	Kind	Value
#[subscriber("orders")]	by name	fixed here
#[subscriber]	by name	.name(..) at the mount site
#[subscriber(MemorySource::new("orders"))]	the broker's descriptor	fixed here
#[subscriber(MemorySource)]	the broker's descriptor	.name(..) at the mount site

A name is a Name source and works on any broker that implements Subscribe, which opens its default subscription kind for that name. A descriptor is a type of the broker crate that implements SubscriptionSource and carries the broker's own settings, such as a consumer group or a durable name; the in-memory broker's is MemorySource. The macro reads the descriptor's type off the constructor call, so the expression ends in Type::new(..) and every step after it returns Self. A descriptor kind that the mount site names implements FromName. A source left open must be named: a mount without .name(..) does not compile. .name(..) is one of the mount-site settings and takes any string, so the name may come from configuration read before the app is built. A descriptor kind that needs more than a name, such as a Pulsar topic with its subscription, is written out in full in the attribute. In the manual form the first argument of subscriber is the source: a string, a descriptor, or Unnamed::<Kind>::new() for a source the mount site names, an Unnamed value.

Example

The bare attribute leaves the name to the mount site.

Macros

examples/subscriber_source.rs

/// The attribute leaves the name open: the mount site gives it.
#[subscriber]
async fn audit(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "audited");
    HandlerOutcome::ack()
}

fn app(shard: u32) -> impl App {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(audit.name(format!("audit-{shard}")));
    })
}

Manual

examples/manual/subscriber_source.rs

struct Audit;

impl Handle<Order> for Audit {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "audited");
        ready(Ok(()))
    }
}

fn app(shard: u32) -> impl App {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        // `Unnamed<Name>` is no source yet: the mount does not compile until `.name(..)`.
        b.include(
            subscriber(Unnamed::<Name>::new(), Audit)
                .name(format!("audit-{shard}"))
                .build(),
        );
    })
}

The broker's descriptor

A descriptor is fixed in the attribute, or only its kind is, and the name comes at the mount site.

Macros

examples/subscriber_source.rs

/// The broker's descriptor, fixed in the attribute.
#[subscriber(MemorySource::new("orders"))]
async fn accept(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    HandlerOutcome::ack()
}

/// The descriptor's kind in the attribute, its name at the mount site.
#[subscriber(MemorySource)]
async fn archive(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "archived");
    HandlerOutcome::ack()
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(accept);
        b.include(archive.name("archive"));
    })
}

Manual

examples/manual/subscriber_source.rs

struct Accept;

impl Handle<Order> for Accept {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "accepted");
        ready(Ok(()))
    }
}

struct Archive;

impl Handle<Order> for Archive {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "archived");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        // The broker's descriptor, built here.
        b.include(subscriber(MemorySource::new("orders"), Accept).build());
        // The descriptor's kind, named at the mount site.
        b.include(
            subscriber(Unnamed::<MemorySource>::new(), Archive)
                .name("archive")
                .build(),
        );
    })
}