SubscriptionType

SubscriptionType says how the consumers of one subscription share it: Exclusive, Shared, Failover or KeyShared.

PulsarSubscription::subscription_type sets it, and Shared is the default. Each handler mounted on a subscription is one consumer of it, and so is each instance of the service. A message reaches every subscription on its topic once, and the type decides which consumer of the subscription takes it:

Type	Who takes a message
`Exclusive`	the one consumer of the subscription; a second consumer waits until the first leaves
`Shared`	any consumer, in turn
`Failover`	the active consumer; a standby takes over when it leaves
`KeyShared`	the consumer that holds the message's key, so one key is read in order

The server refuses a second Exclusive consumer as busy, and the client asks again until its operation retries run out; with the default retries it asks for ever, so the second instance does not finish starting. An Exclusive subscription belongs to a service that runs one instance. Under the test harness that second consumer is refused at startup instead, so the test fails rather than hangs. A message's key is its partition key. A retry cap needs Shared or KeyShared, because only there does the server count a message's redeliveries; a cap on the other two stops the service at startup, as Dead letters explains.

Subscription types and ordering reads a shipping service under each of the four types.

Example

Two workers on the Shared subscription workers: each task runs once, on one of them.

Macros
Manual

examples/pulsar_subscriptions_type.rs
/// Two workers attach to the subscription `workers`: each task goes to one of them.
#[subscriber(
    PulsarSubscription::new("tasks", "workers").subscription_type(SubscriptionType::Shared)
)]
async fn first_worker(task: &Task) -> HandlerOutcome {
    tracing::info!(task.id, "first worker ran the task");
    HandlerOutcome::ack()
}

#[subscriber(
    PulsarSubscription::new("tasks", "workers").subscription_type(SubscriptionType::Shared)
)]
async fn second_worker(task: &Task) -> HandlerOutcome {
    tracing::info!(task.id, "second worker ran the task");
    HandlerOutcome::ack()
}

fn app() -> impl App {
    let broker = PulsarBroker::new("pulsar://localhost:6650");
    RustStream::new(AppInfo::new("workers", "0.1.0")).with_broker(broker, |b| {
        b.include(first_worker);
        b.include(second_worker);
    })
}

examples/manual/pulsar_subscriptions_type.rs
/// One worker; the service runs two of them on one subscription.
struct Worker(&'static str);

impl Handle<Task> for Worker {
    fn handle(
        &self,
        task: &Task,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(task.id, worker = self.0, "task ran");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    let broker = PulsarBroker::new("pulsar://localhost:6650");
    RustStream::new(AppInfo::new("workers", "0.1.0")).with_broker(broker, |b| {
        // Both workers attach to the subscription `workers`: each task goes to one of them.
        for name in ["first", "second"] {
            b.include(
                subscriber(
                    PulsarSubscription::new("tasks", "workers")
                        .subscription_type(SubscriptionType::Shared),
                    Worker(name),
                )
                .build(),
            );
        }
    })
}

Example​

Example