RabbitMQ bounded source and target

RabbitMQ is a bounded messaging source in the current runtime. A RabbitMQ-backed <generate> requires an explicit count; it consumes at most that many messages and may return fewer when the queue remains idle for rabbitmq_consumer_idle_timeout_ms (default 250). Missing count raises I809 before broker access.

Selection and parallelism

Configuration	Worker policy	Ordering guarantee	Why to use it
omitted or distribution="ordered"	one consumer	broker FIFO delivery stream	deterministic ordered transformation
distribution="round_robin"	requested worker count	no global order	higher source throughput through competing consumers

The round-robin name describes RabbitMQ's distribution among consumers; DATAMIMIC does not reshuffle the delivered messages. The exact message set must have no duplicates or skips, but cross-worker output order is not defined.

Finite-pool selection does not map to destructive queue consumption. cyclic, unique, selector, iterationSelector, random, reservoir, weighted, stratified, and cumulated therefore raise I947. offset>0 raises I930. These options are never silently ignored.

Default execution streams one bounded load. If the platform selects forced paging, pageSize becomes the bounded consume size for each page. A count="1000" pageSize="100" run therefore performs ten sequential consume windows; page offsets are execution progress, not random access into the queue.

Topology and acknowledgement

Queue topology is broker-managed by default: allow_auto_declare_queue=false, queue_expires_ms is unset, and DATAMIMIC does not declare or bind an exchange. Temporary queues require explicit opt-in and their intended exclusive, auto_delete, and queue_expires_ms values. This avoids AMQP 406 errors from incompatible redeclarations of existing durable queues.

With auto_ack=false, the consumer acknowledges a delivery after the runtime accepts it into the bounded source page. ACK-after-export-flush is not part of the current bounded contract; it belongs to continuous streaming.

Publisher confirms

The target publishes on one long-lived channel, preserves that channel's order, and limits unconfirmed messages with rabbitmq_publisher_max_in_flight (default 100). flush() waits for ACK, NACK, and mandatory-return outcomes. rabbitmq_publisher_confirm_timeout_ms (default 60000) bounds publish backpressure, flush, and shutdown. Failures use the RabbitMQ exporter error catalog instead of exposing raw Pika errors.

The connection profile owns host, virtual host, and credentials. The deployment or test harness injects a unique rabbitmq_queue property and creates that broker-owned queue before execution.

<setup numProcess="1">
    <rabbitmq-exporter id="orders_out" system="rabbit_profile"
                       exchange="" routing_key="{rabbitmq_queue}"/>
    <rabbitmq-importer id="orders_in" system="rabbit_profile"
                       queue="{rabbitmq_queue}" auto_ack="false" prefetch_count="100"/>

    <generate name="created_orders" count="100" target="orders_out">
        <key name="id" generator="IncrementGenerator"/>
        <key name="status" constant="created"/>
    </generate>

    <generate name="received_orders" source="orders_in" count="100"
              distribution="ordered" target="LogExporter"/>
</setup>

Continuous consume status

Kafka also currently exposes only consume_iter_message(count). ADR-034 describes a future continuous path but remains proposed; consume_stream, commit-after-export-flush, rebalance, backpressure, and graceful termination are not implemented. Kafka continuous consumption must land first. A shared streaming SPI should be extracted only after both concrete broker contracts exist.