Module stream_buffer 

Correlated in-process stream response buffer endpoint.

stream_buffer is a small endpoint for workflows where one publisher send produces many response messages. The main use case is HttpConfig::stream_response_to: an HTTP publisher sends a request as usual, reads a streaming HTTP response, and publishes each response item into this buffer with a correlation_id.

The buffer is partitioned by (topic, correlation_id). Publishers write to a topic and require each message to carry metadata["correlation_id"]. Consumers must be configured with both the same topic and the exact correlation id they want to read. This makes parallel streams safe by default: a consumer for request A cannot accidentally drain response items for request B.

Commit semantics are intentionally mq-bridge-like:

• received messages stay uncommitted until the returned batch commit function is called;
• Ack finalizes the read;
• Nack or dropping the batch without committing requeues messages to the same correlation partition;
• acking an end marker with metadata["http_stream_end"] == "true" removes that correlation partition.

HTTP streaming response example

use mq_bridge::models::{Endpoint, EndpointType, HttpConfig, StreamBufferConfig};
use mq_bridge::{CanonicalMessage, Payload};

let buffer_topic = "llm-response-streams";
let correlation_id = "request-42";

// Configure the HTTP publisher to capture streamed response items into the
// shared stream_buffer topic. This endpoint is publisher-only, so it does
// not set a correlation_id.
let http = Endpoint::new(EndpointType::Http(HttpConfig {
    url: "http://127.0.0.1:8000/v1/generate".to_string(),
    stream_response_to: Some(Box::new(Endpoint::new(EndpointType::StreamBuffer(
        StreamBufferConfig::new(buffer_topic).with_capacity(100),
    )))),
    ..Default::default()
}));

// Send the request with an explicit correlation id. If you omit this
// metadata, the HTTP publisher uses format!("{:032x}", message.message_id).
let mut request = CanonicalMessage::new(Payload::Text("hello".into()));
request.metadata.insert("correlation_id".into(), correlation_id.into());
// create_publisher(&http).await?.send(request).await?;

// Read only this request's streamed responses using the same topic and
// correlation id. Other parallel HTTP responses remain isolated.
let responses = Endpoint::new(EndpointType::StreamBuffer(
    StreamBufferConfig::new(buffer_topic)
        .with_correlation_id(correlation_id)
        .with_capacity(100),
));
// let batch = create_consumer(&responses).await?.receive_batch(10).await?;
// (batch.commit)(MessageDisposition::Ack).await?;

Structs

StreamBufferConsumer

Consumer side of the stream_buffer endpoint.

StreamBufferPublisher

Publisher side of the stream_buffer endpoint.