Prelude

One glob, ruststream_pulsar::prelude, brings the core prelude, the broker, its descriptor, positions, context keys and Publish.

use ruststream_pulsar::prelude::*; is the one import a Pulsar routes file writes. It carries the core's own prelude, so #[subscriber], RustStream, HandlerOutcome and Out come with it. It adds PulsarBroker and OperationRetries, the descriptor PulsarSubscription with SubscriptionType and PulsarTopic, the positions PulsarPosition and PulsarSeeker, and the context types with their keys Position and SeekHandle. The core traits Positioned and Seeker come with them.

The publish policy comes under the name every broker crate uses: Publish is PulsarPublish. So a mount site reads the same on every broker. Two broker preludes in one file collide on this name; the prefixed type at the crate root tells them apart.

A handler body bounds its slots with capability traits, such as Out<impl Publisher>, and names no broker type. The one exception is a body that keys a publish: partition_key needs PulsarPublishSteps in scope and the slot bounded Out<impl Publisher<Options = PulsarPublishOptions>, Marker>, both from this prelude.

Example

The descriptor, its subscription type, the broker's retries and the reply policy all come from the one glob.

Macros
Manual

examples/pulsar_broker_prelude.rs
use ruststream_pulsar::prelude::*;

#[subscriber(
    PulsarSubscription::new("orders", "confirmations-svc")
        .subscription_type(SubscriptionType::KeyShared),
    reply
)]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation { id: order.id }
}

#[ruststream::app]
fn app() -> impl App {
    let broker = PulsarBroker::new("pulsar://localhost:6650")
        .operation_retries(OperationRetries::attempts(nonzero!(5u32)));
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(confirm).out_reply(Publish);
    })
}

examples/manual/pulsar_broker_prelude.rs
use ruststream_pulsar::prelude::*;

struct Confirm;

impl Handle<Order, Confirmation> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
        ready(Ok(Confirmation { id: order.id }))
    }
}

fn app() -> impl App {
    let broker = PulsarBroker::new("pulsar://localhost:6650")
        .operation_retries(OperationRetries::attempts(nonzero!(5u32)));
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber(
                PulsarSubscription::new("orders", "confirmations-svc")
                    .subscription_type(SubscriptionType::KeyShared),
                Confirm,
            )
            .reply()
            .build(),
        )
        .out_reply(Publish);
    })
}

Example​

Example