laminar_connectors/rabbitmq/config/
mod.rs

1//! RabbitMQ source and sink configuration.
2//!
3//! Connection settings use the AMQP URI natively (`heartbeat`, `frame_max`, `channel_max`,
4//! `connection_timeout`, and `auth_mechanism` query parameters are interpreted by the client).
5//! Every other option is either a native AMQP property/argument or a local resource bound.
6
7use std::path::PathBuf;
8use std::str::FromStr;
9use std::time::Duration;
10
11use lapin::types::{AMQPValue, FieldTable, LongString, ShortString};
12use lapin::uri::{AMQPScheme, AMQPUri};
13
14use super::metadata::MetadataColumns;
15use crate::config::ConnectorConfig;
16use crate::error::ConnectorError;
17use crate::serde::Format;
18
19const RUNTIME_SOURCE_KEYS: &[&str] = &[
20    "_arrow_schema",
21    "_primary_key_columns",
22    "format",
23    "laminar.source.name",
24];
25const RUNTIME_SINK_KEYS: &[&str] = &[
26    "_arrow_schema",
27    "delivery.guarantee",
28    "format",
29    "sink.write.timeout.ms",
30];
31const CONNECTION_KEYS: &[&str] = &[
32    "url",
33    "username",
34    "password",
35    "connection.name",
36    "connect.timeout.ms",
37    "tls.ca.location",
38    "tls.cert.location",
39    "tls.key.location",
40];
41const SOURCE_KEYS: &[&str] = &[
42    "queue",
43    "queue.declare",
44    "queue.durable",
45    "queue.type",
46    "queue.arguments",
47    "queue.bind.exchange",
48    "queue.bind.routing.keys",
49    "queue.bind.arguments",
50    "prefetch.count",
51    "consumer.tag",
52    "consumer.exclusive",
53    "consumer.arguments",
54    "max.message.bytes",
55    "max.batch.bytes",
56    "on.decode.error",
57    "metadata.columns",
58];
59const SINK_KEYS: &[&str] = &[
60    "exchange",
61    "exchange.declare",
62    "exchange.type",
63    "exchange.durable",
64    "routing.key",
65    "routing.key.column",
66    "mandatory",
67    "delivery.mode",
68    "content.type",
69    "content.encoding",
70    "message.id.column",
71    "correlation.id.column",
72    "priority",
73    "priority.column",
74    "expiration.ms",
75    "reply.to",
76    "type",
77    "app.id",
78    "header.columns",
79    "max.pending",
80    "max.pending.bytes",
81    "confirm.timeout.ms",
82];
83
84/// RabbitMQ's default `max_message_size` (16 MiB); larger messages are rejected by the broker.
85pub(super) const DEFAULT_MAX_MESSAGE_BYTES: usize = 16 * 1024 * 1024;
86const DEFAULT_MAX_BATCH_BYTES: usize = 32 * 1024 * 1024;
87const DEFAULT_PREFETCH: u16 = 1_000;
88const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
89const DEFAULT_CONFIRM_TIMEOUT: Duration = Duration::from_secs(30);
90const DEFAULT_MAX_PENDING: usize = 4_096;
91const DEFAULT_MAX_PENDING_BYTES: usize = 64 * 1024 * 1024;
92const MAX_IO_TIMEOUT: Duration = Duration::from_secs(300);
93
94/// Validated broker connection settings shared by the source and sink.
95#[derive(Clone)]
96pub struct ConnectionConfig {
97    /// Ordered endpoints; each start tries them once, in order.
98    pub endpoints: Vec<AMQPUri>,
99    /// Client-provided connection name shown by the broker.
100    pub connection_name: String,
101    /// Bound for TCP, TLS, and AMQP connection negotiation per endpoint.
102    pub connect_timeout: Duration,
103    /// PEM CA bundle trusted in addition to the Web PKI roots for `amqps` peer verification.
104    pub tls_ca: Option<PathBuf>,
105    /// PEM client certificate and PKCS#8 key for mutual TLS.
106    pub tls_client: Option<(PathBuf, PathBuf)>,
107}
108
109impl std::fmt::Debug for ConnectionConfig {
110    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
111        let endpoints: Vec<String> = self.endpoints.iter().map(redacted_endpoint).collect();
112        f.debug_struct("ConnectionConfig")
113            .field("endpoints", &endpoints)
114            .field("connection_name", &self.connection_name)
115            .field("connect_timeout", &self.connect_timeout)
116            .field("tls_ca", &self.tls_ca)
117            .field(
118                "tls_client",
119                &self.tls_client.as_ref().map(|(cert, _)| cert),
120            )
121            .finish()
122    }
123}
124
125/// An endpoint rendered without credentials, for logs and errors.
126pub(super) fn redacted_endpoint(uri: &AMQPUri) -> String {
127    let scheme = match uri.scheme {
128        AMQPScheme::AMQP => "amqp",
129        AMQPScheme::AMQPS => "amqps",
130    };
131    format!(
132        "{scheme}://{}:{}/{}",
133        uri.authority.host, uri.authority.port, uri.vhost
134    )
135}
136
137/// What the source does with a message its codec cannot decode or that exceeds the size bound.
138#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
139pub enum DecodeErrorPolicy {
140    /// Stop the source; the message stays unacknowledged and is redelivered after recovery.
141    #[default]
142    Fail,
143    /// `basic.reject(requeue=false)`: the broker dead-letters the message if the queue has a
144    /// dead-letter exchange (by policy or argument) and otherwise discards it.
145    Reject,
146}
147
148str_enum!(fromstr DecodeErrorPolicy, lowercase, ConnectorError, "invalid on.decode.error",
149    Fail => "fail";
150    Reject => "reject"
151);
152
153/// Opt-in, idempotent queue declaration. Without it the queue must already exist.
154#[derive(Debug, Clone, PartialEq)]
155pub struct QueueDeclaration {
156    /// `durable` declaration flag.
157    pub durable: bool,
158    /// Declaration arguments, including `x-queue-type` when `queue.type` is set.
159    pub arguments: FieldTable,
160}
161
162/// Opt-in, idempotent `queue.bind` calls applied at startup.
163#[derive(Debug, Clone, PartialEq)]
164pub struct QueueBinding {
165    /// Source exchange.
166    pub exchange: String,
167    /// One `queue.bind` per routing key.
168    pub routing_keys: Vec<String>,
169    /// Binding arguments (for example a headers-exchange match).
170    pub arguments: FieldTable,
171}
172
173/// Validated RabbitMQ source settings.
174#[derive(Debug, Clone)]
175pub struct RabbitMqSourceConfig {
176    /// Broker connection.
177    pub connection: ConnectionConfig,
178    /// Queue consumed by this source.
179    pub queue: String,
180    /// Declaration applied before consuming, if requested.
181    pub declare: Option<QueueDeclaration>,
182    /// Bindings applied before consuming, if requested.
183    pub binding: Option<QueueBinding>,
184    /// `basic.qos` prefetch count: the most unacknowledged deliveries the broker sends.
185    pub prefetch: u16,
186    /// Consumer tag; the broker generates one when absent.
187    pub consumer_tag: Option<String>,
188    /// `basic.consume` exclusive flag.
189    pub exclusive: bool,
190    /// `basic.consume` arguments (for example `x-priority`).
191    pub consumer_arguments: FieldTable,
192    /// Largest accepted message body.
193    pub max_message_bytes: usize,
194    /// Payload bytes decoded into one batch.
195    pub max_batch_bytes: usize,
196    /// Undecodable or oversized message handling.
197    pub on_decode_error: DecodeErrorPolicy,
198    /// Payload codec.
199    pub format: Format,
200    /// Requested message metadata columns.
201    pub metadata: MetadataColumns,
202}
203
204/// AMQP `delivery-mode` written by the sink.
205#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
206pub enum DeliveryMode {
207    /// Delivery mode 2: written to disk by durable queues before the publisher confirm.
208    #[default]
209    Persistent,
210    /// Delivery mode 1: may be lost on broker restart.
211    Transient,
212}
213
214str_enum!(fromstr DeliveryMode, lowercase, ConnectorError, "invalid delivery.mode",
215    Persistent => "persistent", "2";
216    Transient => "transient", "1"
217);
218
219impl DeliveryMode {
220    /// The AMQP octet.
221    #[must_use]
222    pub const fn amqp(self) -> u8 {
223        match self {
224            Self::Persistent => 2,
225            Self::Transient => 1,
226        }
227    }
228}
229
230/// Opt-in, idempotent exchange declaration.
231#[derive(Debug, Clone, PartialEq, Eq)]
232pub struct ExchangeDeclaration {
233    /// Exchange type (`direct`, `fanout`, `topic`, `headers`, or a plugin type).
234    pub kind: String,
235    /// `durable` declaration flag.
236    pub durable: bool,
237}
238
239/// Where a per-message value comes from.
240#[derive(Debug, Clone, PartialEq, Eq)]
241pub enum Valued<T> {
242    /// Same value for every message.
243    Literal(T),
244    /// Read from a query column.
245    Column(String),
246}
247
248/// Validated RabbitMQ sink settings.
249#[derive(Debug, Clone)]
250pub struct RabbitMqSinkConfig {
251    /// Broker connection.
252    pub connection: ConnectionConfig,
253    /// Target exchange; empty is the default exchange.
254    pub exchange: String,
255    /// Declaration applied at open, if requested.
256    pub exchange_declare: Option<ExchangeDeclaration>,
257    /// Routing key.
258    pub routing_key: Valued<String>,
259    /// `mandatory` publish flag: unroutable messages are returned and fail the write.
260    pub mandatory: bool,
261    /// `delivery-mode` property.
262    pub delivery_mode: DeliveryMode,
263    /// `content-type` property.
264    pub content_type: Option<String>,
265    /// `content-encoding` property.
266    pub content_encoding: Option<String>,
267    /// Column providing `message-id`.
268    pub message_id_column: Option<String>,
269    /// Column providing `correlation-id`.
270    pub correlation_id_column: Option<String>,
271    /// `priority` property.
272    pub priority: Option<Valued<u8>>,
273    /// `expiration` property: per-message TTL in milliseconds, enforced by the broker.
274    pub expiration_ms: Option<u32>,
275    /// `reply-to` property.
276    pub reply_to: Option<String>,
277    /// `type` property.
278    pub message_type: Option<String>,
279    /// `app-id` property.
280    pub app_id: Option<String>,
281    /// Columns written as typed message headers.
282    pub header_columns: Vec<String>,
283    /// Most publishes awaiting a broker confirm.
284    pub max_pending: usize,
285    /// Most payload bytes awaiting a broker confirm.
286    pub max_pending_bytes: usize,
287    /// Bound for one confirm drain.
288    pub confirm_timeout: Duration,
289    /// Payload codec.
290    pub format: Format,
291}
292
293impl RabbitMqSinkConfig {
294    /// Columns consumed as routing data or message properties rather than payload.
295    #[must_use]
296    pub fn property_columns(&self) -> Vec<&str> {
297        let mut columns = Vec::new();
298        if let Valued::Column(name) = &self.routing_key {
299            columns.push(name.as_str());
300        }
301        if let Some(Valued::Column(name)) = &self.priority {
302            columns.push(name.as_str());
303        }
304        columns.extend(self.message_id_column.as_deref());
305        columns.extend(self.correlation_id_column.as_deref());
306        columns.extend(self.header_columns.iter().map(String::as_str));
307        columns
308    }
309}
310
311impl RabbitMqSourceConfig {
312    /// Parse and validate a source configuration without I/O.
313    ///
314    /// # Errors
315    /// Returns a configuration error for unknown, missing, conflicting, or invalid options.
316    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
317        let format = parse_source_format(config)?;
318        reject_unknown(
319            config,
320            &[
321                RUNTIME_SOURCE_KEYS,
322                CONNECTION_KEYS,
323                SOURCE_KEYS,
324                format.source_option_keys(),
325            ],
326            "source",
327        )?;
328        let connection = ConnectionConfig::from_config(config, "laminardb-rabbitmq-source")?;
329        let queue = config
330            .get_non_empty("queue")?
331            .ok_or_else(|| ConnectorError::missing_config("queue"))?;
332        let declare = config.get_parsed::<bool>("queue.declare")?.unwrap_or(false);
333        let declaration_keys = ["queue.durable", "queue.type", "queue.arguments"];
334        if !declare {
335            if let Some(key) = declaration_keys
336                .iter()
337                .find(|key| config.get(key).is_some())
338            {
339                return Err(conflict(&format!(
340                    "'{key}' applies only with 'queue.declare' = 'true'; an existing queue keeps \
341                     its declaration and operator policies"
342                )));
343            }
344        }
345        let declare = declare
346            .then(|| -> Result<QueueDeclaration, ConnectorError> {
347                let mut arguments = parse_table(config, "queue.arguments")?;
348                if let Some(kind) = config.get_non_empty("queue.type")? {
349                    if arguments
350                        .inner()
351                        .contains_key(&ShortString::from("x-queue-type"))
352                    {
353                        return Err(conflict(
354                            "'queue.type' and 'queue.arguments' x-queue-type are both set",
355                        ));
356                    }
357                    arguments.insert("x-queue-type".into(), AMQPValue::LongString(kind.into()));
358                }
359                reject_stream_queue(&arguments)?;
360                Ok(QueueDeclaration {
361                    durable: config.get_parsed::<bool>("queue.durable")?.unwrap_or(true),
362                    arguments,
363                })
364            })
365            .transpose()?;
366        let binding = parse_binding(config)?;
367        let prefetch = config
368            .get_parsed::<u16>("prefetch.count")?
369            .unwrap_or(DEFAULT_PREFETCH);
370        if prefetch == 0 {
371            return Err(ConnectorError::ConfigurationError(
372                "'prefetch.count' must be positive; 0 means unlimited in AMQP and leaves client \
373                 memory unbounded"
374                    .into(),
375            ));
376        }
377        let max_message_bytes =
378            config.get_positive("max.message.bytes", DEFAULT_MAX_MESSAGE_BYTES)?;
379        let max_batch_bytes = config.get_positive("max.batch.bytes", DEFAULT_MAX_BATCH_BYTES)?;
380        if max_batch_bytes < max_message_bytes {
381            return Err(conflict(
382                "'max.batch.bytes' must be at least 'max.message.bytes' so every accepted message \
383                 fits one batch",
384            ));
385        }
386        let metadata = MetadataColumns::parse(config.get("metadata.columns"))?;
387        Ok(Self {
388            connection,
389            queue,
390            declare,
391            binding,
392            prefetch,
393            consumer_tag: config.get_non_empty("consumer.tag")?,
394            exclusive: config
395                .get_parsed::<bool>("consumer.exclusive")?
396                .unwrap_or(false),
397            consumer_arguments: parse_table(config, "consumer.arguments")?,
398            max_message_bytes,
399            max_batch_bytes,
400            on_decode_error: config
401                .get("on.decode.error")
402                .map(DecodeErrorPolicy::from_str)
403                .transpose()?
404                .unwrap_or_default(),
405            format,
406            metadata,
407        })
408    }
409
410    /// Semantic identity: the broker object consumed, without endpoints or credentials.
411    #[must_use]
412    pub fn identity(&self) -> std::collections::BTreeMap<String, String> {
413        let vhost = self
414            .connection
415            .endpoints
416            .first()
417            .map_or_else(String::new, |uri| uri.vhost.clone());
418        std::collections::BTreeMap::from([
419            ("vhost".to_string(), vhost),
420            ("queue".to_string(), self.queue.clone()),
421            ("format".to_string(), self.format.to_string()),
422            ("metadata.columns".to_string(), self.metadata.to_string()),
423        ])
424    }
425}
426
427impl RabbitMqSinkConfig {
428    /// Parse and validate a sink configuration without I/O.
429    ///
430    /// # Errors
431    /// Returns a configuration error for unknown, missing, conflicting, or invalid options.
432    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
433        reject_unknown(
434            config,
435            &[RUNTIME_SINK_KEYS, CONNECTION_KEYS, SINK_KEYS],
436            "sink",
437        )?;
438        let connection = ConnectionConfig::from_config(config, "laminardb-rabbitmq-sink")?;
439        let exchange = config.get("exchange").unwrap_or_default().to_string();
440        let declare = config
441            .get_parsed::<bool>("exchange.declare")?
442            .unwrap_or(false);
443        if !declare
444            && (config.get("exchange.type").is_some() || config.get("exchange.durable").is_some())
445        {
446            return Err(conflict(
447                "'exchange.type' and 'exchange.durable' apply only with 'exchange.declare' = 'true'",
448            ));
449        }
450        if declare && exchange.is_empty() {
451            return Err(conflict("the default exchange cannot be declared"));
452        }
453        let exchange_declare = declare
454            .then(|| -> Result<ExchangeDeclaration, ConnectorError> {
455                Ok(ExchangeDeclaration {
456                    kind: config
457                        .get_non_empty("exchange.type")?
458                        .ok_or_else(|| ConnectorError::missing_config("exchange.type"))?,
459                    durable: config
460                        .get_parsed::<bool>("exchange.durable")?
461                        .unwrap_or(true),
462                })
463            })
464            .transpose()?;
465        let routing_key = match (
466            config.get("routing.key"),
467            config.get_non_empty("routing.key.column")?,
468        ) {
469            (Some(_), Some(_)) => {
470                return Err(conflict(
471                    "set 'routing.key' or 'routing.key.column', not both",
472                ))
473            }
474            (Some(key), None) => Valued::Literal(key.to_string()),
475            (None, Some(column)) => Valued::Column(column),
476            (None, None) => {
477                return Err(ConnectorError::ConfigurationError(
478                    "set 'routing.key' (may be empty) or 'routing.key.column'".into(),
479                ))
480            }
481        };
482        let priority = match (
483            config.get("priority"),
484            config.get_non_empty("priority.column")?,
485        ) {
486            (Some(_), Some(_)) => {
487                return Err(conflict("set 'priority' or 'priority.column', not both"))
488            }
489            (Some(_), None) => Some(Valued::Literal(config.require_parsed::<u8>("priority")?)),
490            (None, Some(column)) => Some(Valued::Column(column)),
491            (None, None) => None,
492        };
493        let header_columns = config.get_list("header.columns");
494        let confirm_timeout = config.get_millis(
495            "confirm.timeout.ms",
496            DEFAULT_CONFIRM_TIMEOUT,
497            MAX_IO_TIMEOUT,
498        )?;
499        let max_pending = config.get_positive("max.pending", DEFAULT_MAX_PENDING)?;
500        let max_pending_bytes =
501            config.get_positive("max.pending.bytes", DEFAULT_MAX_PENDING_BYTES)?;
502        let sink = Self {
503            connection,
504            exchange,
505            exchange_declare,
506            routing_key,
507            mandatory: config.get_parsed::<bool>("mandatory")?.unwrap_or(true),
508            delivery_mode: config
509                .get("delivery.mode")
510                .map(DeliveryMode::from_str)
511                .transpose()?
512                .unwrap_or_default(),
513            content_type: config.get_non_empty("content.type")?,
514            content_encoding: config.get_non_empty("content.encoding")?,
515            message_id_column: config.get_non_empty("message.id.column")?,
516            correlation_id_column: config.get_non_empty("correlation.id.column")?,
517            priority,
518            expiration_ms: config.get_parsed::<u32>("expiration.ms")?,
519            reply_to: config.get_non_empty("reply.to")?,
520            message_type: config.get_non_empty("type")?,
521            app_id: config.get_non_empty("app.id")?,
522            header_columns,
523            max_pending,
524            max_pending_bytes,
525            confirm_timeout,
526            format: parse_sink_format(config)?,
527        };
528        let mut seen = std::collections::BTreeSet::new();
529        if let Some(duplicate) = sink
530            .property_columns()
531            .into_iter()
532            .find(|name| !seen.insert(*name))
533        {
534            return Err(conflict(&format!(
535                "column '{duplicate}' is mapped to more than one message property"
536            )));
537        }
538        Ok(sink)
539    }
540}
541
542impl ConnectionConfig {
543    fn from_config(config: &ConnectorConfig, default_name: &str) -> Result<Self, ConnectorError> {
544        let urls = config.get_list("url");
545        if urls.is_empty() {
546            return Err(ConnectorError::missing_config("url"));
547        }
548        let username = config.get("username");
549        let password = config.get("password");
550        if password.is_some() && username.is_none() {
551            return Err(conflict("'password' requires 'username'"));
552        }
553        let endpoints = urls
554            .iter()
555            .map(|url| {
556                let mut uri = AMQPUri::from_str(url).map_err(|error| {
557                    // The parser error never echoes the URI, so credentials stay out of logs.
558                    ConnectorError::ConfigurationError(format!("invalid AMQP 'url': {error}"))
559                })?;
560                if let Some(username) = username {
561                    if url.contains('@') {
562                        return Err(conflict(
563                            "credentials are set both in 'url' and in 'username'/'password'",
564                        ));
565                    }
566                    uri.authority.userinfo.username = username.to_string();
567                    uri.authority.userinfo.password = password.unwrap_or_default().to_string();
568                }
569                Ok(uri)
570            })
571            .collect::<Result<Vec<_>, ConnectorError>>()?;
572        let tls_ca = config.get_non_empty("tls.ca.location")?.map(PathBuf::from);
573        let tls_client = match (
574            config.get_non_empty("tls.cert.location")?,
575            config.get_non_empty("tls.key.location")?,
576        ) {
577            (Some(cert), Some(key)) => Some((PathBuf::from(cert), PathBuf::from(key))),
578            (None, None) => None,
579            _ => {
580                return Err(conflict(
581                    "'tls.cert.location' and 'tls.key.location' must be set together",
582                ))
583            }
584        };
585        if (tls_ca.is_some() || tls_client.is_some())
586            && endpoints.iter().any(|uri| uri.scheme != AMQPScheme::AMQPS)
587        {
588            return Err(conflict(
589                "TLS options require every 'url' to use the amqps scheme",
590            ));
591        }
592        let uri_timeouts: Vec<u64> = endpoints
593            .iter()
594            .filter_map(|uri| uri.query.connection_timeout)
595            .collect();
596        if !uri_timeouts.is_empty() && config.get("connect.timeout.ms").is_some() {
597            return Err(conflict(
598                "the connection timeout is set both by 'connect.timeout.ms' and by the 'url' \
599                 connection_timeout query parameter",
600            ));
601        }
602        let connect_timeout = match uri_timeouts.iter().max() {
603            Some(millis) => Duration::from_millis(*millis),
604            None => config.get_millis(
605                "connect.timeout.ms",
606                DEFAULT_CONNECT_TIMEOUT,
607                MAX_IO_TIMEOUT,
608            )?,
609        };
610        let vhosts: std::collections::BTreeSet<&str> =
611            endpoints.iter().map(|uri| uri.vhost.as_str()).collect();
612        if vhosts.len() > 1 {
613            return Err(conflict(
614                "every 'url' must name the same virtual host; endpoints are failover addresses \
615                 for one broker or cluster",
616            ));
617        }
618        Ok(Self {
619            endpoints,
620            connection_name: config
621                .get_non_empty("connection.name")?
622                .unwrap_or_else(|| default_name.to_string()),
623            connect_timeout,
624            tls_ca,
625            tls_client,
626        })
627    }
628}
629
630fn reject_unknown(
631    config: &ConnectorConfig,
632    groups: &[&[&str]],
633    direction: &str,
634) -> Result<(), ConnectorError> {
635    let allowed: Vec<&str> = groups
636        .iter()
637        .flat_map(|group| group.iter().copied())
638        .collect();
639    config.reject_unknown_properties(&allowed, &format!("RabbitMQ {direction}"))
640}
641
642fn parse_binding(config: &ConnectorConfig) -> Result<Option<QueueBinding>, ConnectorError> {
643    let Some(exchange) = config.get_non_empty("queue.bind.exchange")? else {
644        if config.get("queue.bind.routing.keys").is_some()
645            || config.get("queue.bind.arguments").is_some()
646        {
647            return Err(conflict(
648                "'queue.bind.routing.keys' and 'queue.bind.arguments' require 'queue.bind.exchange'",
649            ));
650        }
651        return Ok(None);
652    };
653    let mut routing_keys = config.get_list("queue.bind.routing.keys");
654    if routing_keys.is_empty() {
655        routing_keys.push(String::new());
656    }
657    Ok(Some(QueueBinding {
658        exchange,
659        routing_keys,
660        arguments: parse_table(config, "queue.bind.arguments")?,
661    }))
662}
663
664/// Stream queues do not redeliver unacknowledged messages after a consumer closes; a consumer
665/// restarts from its `x-stream-offset`, so this connector's recovery contract does not hold.
666fn reject_stream_queue(arguments: &FieldTable) -> Result<(), ConnectorError> {
667    let is_stream = arguments
668        .inner()
669        .get(&ShortString::from("x-queue-type"))
670        .is_some_and(
671            |value| matches!(value, AMQPValue::LongString(kind) if kind.as_bytes() == b"stream"),
672        );
673    if is_stream {
674        return Err(ConnectorError::FeatureUnsupported(
675            "stream queues do not redeliver unacknowledged messages and are not supported by the \
676             AMQP 0-9-1 source; use a classic or quorum queue"
677                .into(),
678        ));
679    }
680    Ok(())
681}
682
683fn format_option(config: &ConnectorConfig) -> Result<Format, ConnectorError> {
684    Ok(config
685        .get("format")
686        .map(Format::parse)
687        .transpose()
688        .map_err(|error| ConnectorError::ConfigurationError(error.to_string()))?
689        .unwrap_or(Format::Json))
690}
691
692/// Sources also read FORMAT FPML: one complete document per message.
693fn parse_source_format(config: &ConnectorConfig) -> Result<Format, ConnectorError> {
694    match format_option(config)? {
695        format @ (Format::Json | Format::Csv | Format::Raw) => Ok(format),
696        #[cfg(feature = "fpml")]
697        Format::Fpml => Ok(Format::Fpml),
698        #[cfg(not(feature = "fpml"))]
699        Format::Fpml => Err(ConnectorError::ConfigurationError(
700            "format 'fpml' requires a build with the fpml feature".into(),
701        )),
702        other => Err(ConnectorError::FeatureUnsupported(format!(
703            "RabbitMQ sources support FORMAT JSON, CSV, RAW, and FPML; {other} is not qualified \
704             for them"
705        ))),
706    }
707}
708
709fn parse_sink_format(config: &ConnectorConfig) -> Result<Format, ConnectorError> {
710    match format_option(config)? {
711        format @ (Format::Json | Format::Csv | Format::Raw) => Ok(format),
712        other => Err(ConnectorError::FeatureUnsupported(format!(
713            "RabbitMQ sinks support FORMAT JSON, CSV, and RAW; {other} is not qualified for them"
714        ))),
715    }
716}
717
718/// A JSON object of scalar AMQP arguments. JSON numbers become signed 64-bit integers (or
719/// doubles when fractional), strings become long strings, and booleans stay booleans.
720fn parse_table(config: &ConnectorConfig, key: &str) -> Result<FieldTable, ConnectorError> {
721    let Some(text) = config.get(key) else {
722        return Ok(FieldTable::default());
723    };
724    let value: serde_json::Value = serde_json::from_str(text).map_err(|error| {
725        ConnectorError::ConfigurationError(format!("'{key}' must be a JSON object: {error}"))
726    })?;
727    let serde_json::Value::Object(entries) = value else {
728        return Err(ConnectorError::ConfigurationError(format!(
729            "'{key}' must be a JSON object"
730        )));
731    };
732    let mut table = FieldTable::default();
733    for (name, value) in entries {
734        let amqp = match value {
735            serde_json::Value::Bool(flag) => AMQPValue::Boolean(flag),
736            serde_json::Value::String(text) => AMQPValue::LongString(LongString::from(text)),
737            serde_json::Value::Number(number) => match (number.as_i64(), number.as_f64()) {
738                (Some(integer), _) => AMQPValue::LongLongInt(integer),
739                (None, Some(float)) => AMQPValue::Double(float),
740                (None, None) => {
741                    return Err(ConnectorError::ConfigurationError(format!(
742                        "'{key}' entry '{name}' is out of the signed 64-bit range"
743                    )))
744                }
745            },
746            _ => {
747                return Err(ConnectorError::ConfigurationError(format!(
748                    "'{key}' entry '{name}' must be a string, number, or boolean"
749                )))
750            }
751        };
752        if name.len() > 255 {
753            return Err(ConnectorError::ConfigurationError(format!(
754                "'{key}' entry name exceeds the 255-byte AMQP short string"
755            )));
756        }
757        table.insert(name.into(), amqp);
758    }
759    Ok(table)
760}
761
762fn conflict(message: &str) -> ConnectorError {
763    ConnectorError::ConfigurationError(message.to_string())
764}
765
766#[cfg(test)]
767mod tests;
laminar_connectors/rabbitmq/config/mod.rs

laminar_connectors/rabbitmq/config/
mod.rs
