Struct PulsarBatchContext 

pub struct PulsarBatchContext { /* private fields */ }

The subscription-scoped batch context of a Pulsar subscription: its seeker, shared by every delivery of the batch.

The runtime builds one per dispatched batch from the batch’s first delivery (see BuildBatchContext), and a batch body reads it by key - SeekHandle - through ctx.context(..). A Position has no place here: a batch spans many deliveries, so where to seek rides the elements themselves (a &[Message<H, T>] batch reads it off each element’s header contract). Keeping this a separate type from PulsarContext is what rejects a batch body asking for per-delivery fields at compile time.

Pulsar’s client hands over one delivery at a time, so a batch here is assembled on the client from the size the mount site’s batch(n) names. The body sees the same context, seeker and settlement as on a broker that batches on the wire.

Examples

use ruststream_pulsar::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Job {
    id: u64,
    rewind: bool,
}

/// A batch that carries the rewind marker replays the retained backlog once it is settled:
/// the next batch opens at the beginning of the log.
#[subscriber(PulsarSubscription::new("jobs.bulk", "bulk-workers"))]
async fn work(jobs: &[Job], ctx: &mut Context<'_, PulsarBatchContext>) -> HandlerOutcome {
    let first = jobs.first().map(|job| job.id);
    tracing::info!(count = jobs.len(), ?first, "jobs received");
    if jobs.iter().any(|job| job.rewind)
        && ctx
            .context(SeekHandle)
            .seek(PulsarPosition::earliest())
            .await
            .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("jobs", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650"),
        |b| {
            b.include(work.batch(nonzero!(100)));
        },
    )
}

Trait Implementations

impl BuildBatchContext<PulsarMessage> for PulsarBatchContext

fn build(first: &PulsarMessage) -> Self

Builds the context value by reading subscription-scoped fields out of the batch’s first delivery, or out of the batch itself where it lends its values as one slice.

impl Debug for PulsarBatchContext

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Field<PulsarBatchContext> for SeekHandle

type Value<'a> = &'a PulsarSeeker

The value read through this key, borrowed from the source for 'a.

fn get(self, src: &PulsarBatchContext) -> &PulsarSeeker

Reads this key’s field out of src.