ractogateway.kafka.stream

Batched consume-process-publish utilities for real-time Kafka streams.

class ractogateway.kafka.stream.KafkaStreamProcessor(*, consumer, producer=None, config=None)

Bases: object

Real-time stream loop helper for high-throughput event processing.

Parameters:

• consumer (KafkaConsumerClient) – Source consumer used to read input events.

• producer (KafkaProducerClient | None) – Optional producer used to publish handler outputs.

• config (KafkaStreamConfig | None) – Batch and commit controls. Defaults are applied when omitted.

collect_batch()

Collect up to batch_size messages bounded by max_wait_ms.

Return type:

list[KafkaMessage]

process_once(handler, *, output_topic=None)

Run one batch cycle and return the number of consumed messages.

Return type:

int

Handler return options

• None: consume-only, no output publish.

• Sequence[KafkaPublishRequest]: explicit per-message publish plans.

• Sequence[Any]: publish raw values to output_topic.

run(handler, *, output_topic=None, iterations=None)

Run repeated batch cycles. Returns total consumed message count.

Return type:

int