Implementation:ArroyoSystems Arroyo Rabbitmq Connector

Knowledge Sources	ArroyoSystems_Arroyo
Domains	Streaming, Connectors
Last Updated	2026-02-08 08:00 GMT

Overview

RabbitmqConnector implements the Arroyo Connector trait for RabbitMQ Streams, providing a source connector with TLS support, configurable offsets, and load balancer mode.

Description

The RabbitMQ connector uses the rabbitmq-stream-client crate to consume from RabbitMQ Streams (not classic queues). Connection configuration supports optional host, username (default "guest"), password (default "guest"), virtual_host (default "/"), port (default 5552), TLS settings (enable, trust_certificates, root/client certificate paths, client keys path), and load_balancer_mode. The connector currently only supports source table types; the sink case uses todo!(). Source offsets are configured via SourceOffset with First, Last, and Next options, mapped to RabbitMQ's OffsetSpecification. The test method validates connectivity by attempting to build a consumer on the configured stream. The make_operator method constructs a RabbitmqStreamSourceFunc with the connection config, stream name, offset mode, format, framing, and bad_data settings.

Usage

Use RabbitmqConnector when building Arroyo pipelines that need to consume messages from RabbitMQ Streams with durable offset tracking.

Code Reference

Source Location

• Repository: ArroyoSystems_Arroyo
• File: crates/arroyo-connectors/src/rabbitmq/mod.rs

Signature

pub struct RabbitmqConnector {}

impl Connector for RabbitmqConnector {
    type ProfileT = RabbitmqStreamConfig;
    type TableT = RabbitmqStreamTable;

    fn name(&self) -> &'static str; // returns "rabbitmq"
    fn table_type(&self, _: RabbitmqStreamConfig,
        _: RabbitmqStreamTable) -> ConnectionType; // always Source
    fn from_config(&self, id: Option<i64>, name: &str, config: RabbitmqStreamConfig,
        table: RabbitmqStreamTable, schema: Option<&ConnectionSchema>) -> anyhow::Result<Connection>;
    fn make_operator(&self, profile: RabbitmqStreamConfig, table: RabbitmqStreamTable,
        config: OperatorConfig) -> anyhow::Result<ConstructedOperator>;
}

impl RabbitmqStreamConfig {
    async fn get_environment(&self) -> anyhow::Result<Environment>;
}

impl SourceOffset {
    pub fn offset(&self) -> OffsetSpecification; // First, Last, Next
}

impl From<TlsConfig> for TlsConfiguration { ... }

Import

use arroyo_connectors::rabbitmq::RabbitmqConnector;

I/O Contract

Inputs

Name	Type	Required	Description
stream	String	Yes	RabbitMQ Stream name to consume from
type	TableType	Yes	Source with offset configuration
host	Option<String>	No	RabbitMQ host (default: "localhost")
port	Option<u16>	No	RabbitMQ port (default: 5552)
username	Option<String>	No	RabbitMQ username (default: "guest")
password	Option<String>	No	RabbitMQ password (default: "guest")
source.offset	SourceOffset	No	First, Last (default), or Next
tls_config	Option<TlsConfig>	No	TLS configuration with certificate paths
load_balancer_mode	Option<bool>	No	Enable load balancer mode

Outputs

Name	Type	Description
Connection	Connection	Configured RabbitMQ Stream connection
ConstructedOperator	ConstructedOperator	RabbitmqStreamSourceFunc operator

Usage Examples

CREATE TABLE rabbitmq_source (
    value TEXT
) WITH (
    connector = 'rabbitmq',
    host = 'localhost',
    port = '5552',
    stream = 'my-stream',
    type = 'source',
    'source.offset' = 'first',
    format = 'json'
);

Related Pages

• Environment:ArroyoSystems_Arroyo_Rust_Runtime