max_message_size

max_message_size is the largest payload a publish may carry, five mebibytes as on the server; a larger one fails before it is sent.

The default matches the server's own maxMessageSize. A deployment that changed that setting sets the same number here, in bytes.

A publish over the limit returns PulsarError::Publish, naming the topic, and nothing leaves the service. The check matters because a server answers an oversized frame by dropping the connection, and every producer and consumer on that connection fails with it. A handler that receives the error decides how its delivery settles. Retrying cannot make a payload smaller, so a handler usually logs it and drops the delivery or settles it some other final way.

A TestApp test refuses the same publishes in process.

Connecting to a production cluster sets the message size a cluster accepts.

Example

The cluster runs with a one-mebibyte limit. A scan that encodes to more is logged and dropped; a smaller one is archived.

Macros
Manual

examples/pulsar_broker_limits.rs
#[subscriber(PulsarSubscription::new("uploads", "archiver"))]
async fn archive(upload: &Upload, Out(scans): Out<impl Publisher, Archive>) -> HandlerOutcome {
    let scan = Scan {
        upload_id: upload.id,
        pages: (1..=upload.pages)
            .map(|page| format!("page-{page:06}"))
            .collect(),
    };
    if let Err(err) = scans.message(&scan).publish().await {
        // Retrying cannot shrink the scan, so it is logged and dropped.
        tracing::warn!(subscription = "archiver", upload.id, error = %err, "scan not archived");
        return HandlerOutcome::drop();
    }
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    // The cluster runs with `maxMessageSize=1048576`.
    let broker =
        PulsarBroker::new("pulsar://localhost:6650").max_message_size(nonzero!(1_048_576usize));
    RustStream::new(AppInfo::new("archive", "0.1.0")).with_broker(broker, |b| {
        b.include(archive).out(Archive, Publish).build();
    })
}

examples/manual/pulsar_broker_limits.rs
struct ArchiveUpload;

impl<Entry> Handle<Upload, (), Outs<(Entry,)>> for ArchiveUpload
where
    Entry: OutEntry<Archive, Wire: Publisher>,
{
    async fn handle(
        &self,
        upload: &Upload,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let scan = Scan {
            upload_id: upload.id,
            pages: (1..=upload.pages)
                .map(|page| format!("page-{page:06}"))
                .collect(),
        };
        if let Err(err) = outs.get(Archive).message(&scan).publish().await {
            // Retrying cannot shrink the scan, so it is logged and dropped.
            tracing::warn!(subscription = "archiver", upload.id, error = %err, "scan not archived");
            return Err(HandlerOutcome::drop());
        }
        Ok(())
    }
}

fn app() -> impl App {
    // The cluster runs with `maxMessageSize=1048576`.
    let broker =
        PulsarBroker::new("pulsar://localhost:6650").max_message_size(nonzero!(1_048_576usize));
    RustStream::new(AppInfo::new("archive", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber(
                PulsarSubscription::new("uploads", "archiver"),
                ArchiveUpload,
            )
            .build(),
        )
        .out(Archive, Publish)
        .build();
    })
}

Example​

Example