laminar_connectors/ibmmq/config/
mod.rs

1//! IBM MQ source and sink configuration.
2//!
3//! Connection settings map to MQI structures: `queue.manager` to the `MQCONNX` queue manager
4//! name, `channel` and `connection.name` to an `MQCD` client channel, `ccdt.url` to the
5//! `MQCNO` CCDT URL, `username`/`password` to `MQCSP`, TLS options to `MQCD`/`MQSCO`, and
6//! `reconnect` to the `MQCNO` reconnect option. With neither `channel` nor `ccdt.url`, the
7//! client's own environment (`MQSERVER`, `MQCCDTURL`, `mqclient.ini`) selects the channel.
8
9use std::path::PathBuf;
10use std::str::FromStr;
11use std::time::Duration;
12
13use super::metadata::MetadataColumns;
14use crate::config::ConnectorConfig;
15use crate::error::ConnectorError;
16use crate::serde::Format;
17
18const RUNTIME_SOURCE_KEYS: &[&str] = &[
19    "_arrow_schema",
20    "_primary_key_columns",
21    "format",
22    "laminar.source.name",
23];
24const RUNTIME_SINK_KEYS: &[&str] = &[
25    "_arrow_schema",
26    "delivery.guarantee",
27    "format",
28    "sink.write.timeout.ms",
29];
30const CONNECTION_KEYS: &[&str] = &[
31    "queue.manager",
32    "channel",
33    "connection.name",
34    "ccdt.url",
35    "username",
36    "password",
37    "application.name",
38    "tls.cipher.spec",
39    "tls.key.repository",
40    "tls.key.repository.password",
41    "tls.certificate.label",
42    "tls.peer.name",
43    "reconnect",
44    "client.library",
45    "operation.timeout.ms",
46];
47const SOURCE_KEYS: &[&str] = &[
48    "queue",
49    "wait.interval.ms",
50    "max.message.bytes",
51    "max.batch.bytes",
52    "max.uow.messages",
53    "max.uow.bytes",
54    "on.decode.error",
55    "convert",
56    "metadata.columns",
57];
58const SINK_KEYS: &[&str] = &[
59    "queue",
60    "target.queue.manager",
61    "persistence",
62    "priority",
63    "priority.column",
64    "expiry",
65    "message.id.column",
66    "correlation.id.column",
67    "reply.to.queue",
68    "reply.to.queue.manager",
69    "message.format",
70    "ccsid",
71    "property.columns",
72    "max.uow.messages",
73];
74
75/// MQ's default `MAXMSGL` for queues and channels (4 MiB).
76const DEFAULT_MAX_MESSAGE_BYTES: usize = 4 * 1024 * 1024;
77/// MQ's largest supported message (100 MiB).
78pub(super) const MQ_MAX_MESSAGE_BYTES: usize = 100 * 1024 * 1024;
79const DEFAULT_MAX_BATCH_BYTES: usize = 16 * 1024 * 1024;
80/// MQ's default queue manager `MAXUMSGS`.
81const DEFAULT_MAX_UOW_MESSAGES: usize = 10_000;
82const DEFAULT_MAX_UOW_BYTES: usize = 64 * 1024 * 1024;
83const DEFAULT_SINK_UOW_MESSAGES: usize = 1_000;
84const DEFAULT_WAIT: Duration = Duration::from_millis(100);
85const MAX_WAIT: Duration = Duration::from_secs(5);
86const DEFAULT_OPERATION_TIMEOUT: Duration = Duration::from_secs(30);
87const MAX_OPERATION_TIMEOUT: Duration = Duration::from_secs(300);
88
89/// `MQCNO` reconnect option.
90#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
91pub enum Reconnect {
92    /// `MQCNO_RECONNECT_DISABLED`: a broken connection ends the connector generation.
93    #[default]
94    Disabled,
95    /// `MQCNO_RECONNECT_Q_MGR`: reconnect only to the same queue manager.
96    QueueManager,
97    /// `MQCNO_RECONNECT`: reconnect to any queue manager the channel definitions allow.
98    Any,
99}
100
101str_enum!(fromstr Reconnect, lowercase, ConnectorError, "invalid reconnect",
102    Disabled => "disabled";
103    QueueManager => "qmgr", "queue_manager";
104    Any => "any"
105);
106
107/// How the client finds its server-connection channel.
108#[derive(Debug, Clone, PartialEq, Eq)]
109pub enum Endpoint {
110    /// An `MQCD` built from `channel` and `connection.name` (`host(port)[,host(port)]`).
111    Channel {
112        /// Server-connection channel name.
113        channel: String,
114        /// `CONNAME` list; the client tries entries in order.
115        connection_name: String,
116    },
117    /// A client channel definition table (`file://`, `http(s)://`, or `ftp://` URL).
118    Ccdt(String),
119    /// The IBM MQ client's environment and `mqclient.ini`.
120    ClientEnvironment,
121}
122
123/// TLS settings applied to the client channel.
124#[derive(Clone, PartialEq, Eq)]
125pub struct TlsConfig {
126    /// `SSLCIPH` cipher specification, for example `ANY_TLS13_OR_HIGHER`.
127    pub cipher_spec: String,
128    /// Key repository stem (`.kdb`) or PKCS#12 file the client trusts and presents.
129    pub key_repository: Option<String>,
130    /// Key repository password.
131    pub key_repository_password: Option<String>,
132    /// Client certificate label in the key repository.
133    pub certificate_label: Option<String>,
134    /// `SSLPEER` distinguished-name filter the server certificate must match.
135    pub peer_name: Option<String>,
136}
137
138impl std::fmt::Debug for TlsConfig {
139    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
140        f.debug_struct("TlsConfig")
141            .field("cipher_spec", &self.cipher_spec)
142            .field("key_repository", &self.key_repository)
143            .field("certificate_label", &self.certificate_label)
144            .field("peer_name", &self.peer_name)
145            .finish_non_exhaustive()
146    }
147}
148
149/// Validated queue manager connection settings.
150#[derive(Clone)]
151pub struct ConnectionConfig {
152    /// Queue manager name; `None` lets the channel definition decide.
153    pub queue_manager: Option<String>,
154    /// Channel selection.
155    pub endpoint: Endpoint,
156    /// `MQCSP` user.
157    pub username: Option<String>,
158    /// `MQCSP` password.
159    pub password: Option<String>,
160    /// `MQCNO` application name (at most 28 characters).
161    pub application_name: String,
162    /// TLS settings.
163    pub tls: Option<TlsConfig>,
164    /// Reconnect option.
165    pub reconnect: Reconnect,
166    /// Explicit IBM MQ client library (`libmqm_r`) path.
167    pub client_library: Option<PathBuf>,
168    /// Bound the async side waits for one MQI operation; it does not cancel the native call.
169    pub operation_timeout: Duration,
170}
171
172impl std::fmt::Debug for ConnectionConfig {
173    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
174        f.debug_struct("ConnectionConfig")
175            .field("queue_manager", &self.queue_manager)
176            .field("endpoint", &self.endpoint)
177            .field("username", &self.username)
178            .field("application_name", &self.application_name)
179            .field("tls", &self.tls)
180            .field("reconnect", &self.reconnect)
181            .field("client_library", &self.client_library)
182            .finish_non_exhaustive()
183    }
184}
185
186/// What the source does with a message it cannot decode or that exceeds `max.message.bytes`.
187#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
188pub enum DecodeErrorPolicy {
189    /// Stop the source; MQBACK returns the message and increments its `BackoutCount`.
190    #[default]
191    Fail,
192    /// Move the message to the queue's `BOQNAME` in the same unit of work.
193    BackoutQueue,
194}
195
196str_enum!(fromstr DecodeErrorPolicy, lowercase, ConnectorError, "invalid on.decode.error",
197    Fail => "fail";
198    BackoutQueue => "backout_queue", "backout-queue"
199);
200
201/// Validated IBM MQ source settings.
202#[derive(Debug, Clone)]
203pub struct IbmMqSourceConfig {
204    /// Queue manager connection.
205    pub connection: ConnectionConfig,
206    /// Local, alias, or cluster queue consumed by this source.
207    pub queue: String,
208    /// `MQGMO` wait interval for the first message of a poll.
209    pub wait_interval: Duration,
210    /// Receive buffer size; a longer message is never partially delivered.
211    pub max_message_bytes: usize,
212    /// Payload bytes per poll.
213    pub max_batch_bytes: usize,
214    /// Messages one unit of work may hold before intake waits for a checkpoint.
215    pub max_uow_messages: usize,
216    /// Payload bytes one unit of work may hold before intake waits for a checkpoint.
217    pub max_uow_bytes: usize,
218    /// Undecodable or oversized message handling.
219    pub on_decode_error: DecodeErrorPolicy,
220    /// `MQGMO_CONVERT` to UTF-8 (CCSID 1208) for messages in a convertible format.
221    pub convert: bool,
222    /// Payload codec.
223    pub format: Format,
224    /// Requested `MQMD` and property metadata columns.
225    pub metadata: MetadataColumns,
226}
227
228/// `MQMD` `Persistence` written by the sink.
229#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
230pub enum Persistence {
231    /// `MQPER_PERSISTENT`: logged by the queue manager before `MQCMIT` returns.
232    #[default]
233    Persistent,
234    /// `MQPER_NOT_PERSISTENT`: lost on queue manager restart.
235    NotPersistent,
236    /// `MQPER_PERSISTENCE_AS_Q_DEF`: the queue's `DEFPSIST`.
237    AsQueueDefinition,
238}
239
240str_enum!(fromstr Persistence, lowercase, ConnectorError, "invalid persistence",
241    Persistent => "persistent";
242    NotPersistent => "not_persistent", "not-persistent";
243    AsQueueDefinition => "as_queue_definition", "as-queue-definition"
244);
245
246/// Where a per-message value comes from.
247#[derive(Debug, Clone, PartialEq, Eq)]
248pub enum Valued<T> {
249    /// Same value for every message.
250    Literal(T),
251    /// Read from a query column.
252    Column(String),
253}
254
255/// Validated IBM MQ sink settings.
256#[derive(Debug, Clone)]
257pub struct IbmMqSinkConfig {
258    /// Queue manager connection.
259    pub connection: ConnectionConfig,
260    /// Target queue (local, alias, remote definition, or cluster queue).
261    pub queue: String,
262    /// `MQOD` `ObjectQMgrName` for a queue on another queue manager.
263    pub target_queue_manager: Option<String>,
264    /// `MQMD` `Persistence`.
265    pub persistence: Persistence,
266    /// `MQMD` `Priority` (0-9).
267    pub priority: Option<Valued<i32>>,
268    /// `MQMD` `Expiry` in tenths of a second.
269    pub expiry: Option<i32>,
270    /// Column providing `MsgId` (Binary or Utf8, at most 24 bytes); otherwise a new id.
271    pub message_id_column: Option<String>,
272    /// Column providing `CorrelId` (Binary or Utf8, at most 24 bytes).
273    pub correlation_id_column: Option<String>,
274    /// `MQMD` `ReplyToQ`.
275    pub reply_to_queue: Option<String>,
276    /// `MQMD` `ReplyToQMgr`.
277    pub reply_to_queue_manager: Option<String>,
278    /// `MQMD` `Format` override (8 characters); defaults by codec.
279    pub message_format: Option<String>,
280    /// `MQMD` `CodedCharSetId` of text payloads.
281    pub ccsid: i32,
282    /// Columns written as typed message properties.
283    pub property_columns: Vec<String>,
284    /// Messages per unit of work; a larger batch commits in several units.
285    pub max_uow_messages: usize,
286    /// Payload codec.
287    pub format: Format,
288}
289
290impl IbmMqSinkConfig {
291    /// Columns consumed as message descriptor fields or properties rather than payload.
292    #[must_use]
293    pub fn property_columns(&self) -> Vec<&str> {
294        let mut columns = Vec::new();
295        if let Some(Valued::Column(name)) = &self.priority {
296            columns.push(name.as_str());
297        }
298        columns.extend(self.message_id_column.as_deref());
299        columns.extend(self.correlation_id_column.as_deref());
300        columns.extend(self.property_columns.iter().map(String::as_str));
301        columns
302    }
303}
304
305impl IbmMqSourceConfig {
306    /// Parse and validate a source configuration without I/O.
307    ///
308    /// # Errors
309    /// Returns a configuration error for unknown, missing, conflicting, or invalid options.
310    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
311        let format = parse_source_format(config)?;
312        reject_unknown(
313            config,
314            &[
315                RUNTIME_SOURCE_KEYS,
316                CONNECTION_KEYS,
317                SOURCE_KEYS,
318                format.source_option_keys(),
319            ],
320            "source",
321        )?;
322        let connection = ConnectionConfig::from_config(config)?;
323        if connection.reconnect != Reconnect::Disabled {
324            return Err(ConnectorError::ConfigurationError(
325                "the IBM MQ source requires 'reconnect' = 'disabled': a client reconnect backs \
326                 out the unit of work holding uncheckpointed messages, and the engine restart is \
327                 the qualified recovery path"
328                    .into(),
329            ));
330        }
331        let max_message_bytes =
332            config.get_positive("max.message.bytes", DEFAULT_MAX_MESSAGE_BYTES)?;
333        if max_message_bytes > MQ_MAX_MESSAGE_BYTES {
334            return Err(ConnectorError::ConfigurationError(format!(
335                "'max.message.bytes' exceeds IBM MQ's {MQ_MAX_MESSAGE_BYTES}-byte message limit"
336            )));
337        }
338        let max_batch_bytes = config.get_positive("max.batch.bytes", DEFAULT_MAX_BATCH_BYTES)?;
339        let max_uow_bytes = config.get_positive("max.uow.bytes", DEFAULT_MAX_UOW_BYTES)?;
340        if max_batch_bytes < max_message_bytes || max_uow_bytes < max_batch_bytes {
341            return Err(ConnectorError::ConfigurationError(
342                "limits must satisfy max.message.bytes <= max.batch.bytes <= max.uow.bytes".into(),
343            ));
344        }
345        let wait_interval = config.get_millis("wait.interval.ms", DEFAULT_WAIT, MAX_WAIT)?;
346        Ok(Self {
347            connection,
348            queue: config
349                .get_non_empty("queue")?
350                .ok_or_else(|| ConnectorError::missing_config("queue"))?,
351            wait_interval,
352            max_message_bytes,
353            max_batch_bytes,
354            max_uow_messages: config.get_positive("max.uow.messages", DEFAULT_MAX_UOW_MESSAGES)?,
355            max_uow_bytes,
356            on_decode_error: config.get_parsed("on.decode.error")?.unwrap_or_default(),
357            convert: config.get_parsed("convert")?.unwrap_or(false),
358            format,
359            metadata: MetadataColumns::parse(config.get("metadata.columns"))?,
360        })
361    }
362
363    /// Semantic identity: the queue consumed, without channel endpoints or credentials.
364    #[must_use]
365    pub fn identity(&self) -> std::collections::BTreeMap<String, String> {
366        std::collections::BTreeMap::from([
367            (
368                "queue.manager".to_string(),
369                self.connection.queue_manager.clone().unwrap_or_default(),
370            ),
371            ("queue".to_string(), self.queue.clone()),
372            ("format".to_string(), self.format.to_string()),
373            ("convert".to_string(), self.convert.to_string()),
374            ("metadata.columns".to_string(), self.metadata.to_string()),
375        ])
376    }
377}
378
379impl IbmMqSinkConfig {
380    /// Parse and validate a sink configuration without I/O.
381    ///
382    /// # Errors
383    /// Returns a configuration error for unknown, missing, conflicting, or invalid options.
384    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
385        reject_unknown(
386            config,
387            &[RUNTIME_SINK_KEYS, CONNECTION_KEYS, SINK_KEYS],
388            "sink",
389        )?;
390        let priority = match (
391            config.get_parsed::<i32>("priority")?,
392            config.get_non_empty("priority.column")?,
393        ) {
394            (Some(_), Some(_)) => {
395                return Err(conflict("set 'priority' or 'priority.column', not both"))
396            }
397            (Some(priority), None) => Some(Valued::Literal(check_priority(priority)?)),
398            (None, Some(column)) => Some(Valued::Column(column)),
399            (None, None) => None,
400        };
401        let expiry = config.get_parsed::<i32>("expiry")?;
402        if expiry.is_some_and(|tenths| tenths <= 0) {
403            return Err(conflict(
404                "'expiry' is in tenths of a second and must be positive; omit it for unlimited",
405            ));
406        }
407        let message_format = config.get_non_empty("message.format")?;
408        if message_format
409            .as_ref()
410            .is_some_and(|format| format.len() > 8)
411        {
412            return Err(conflict(
413                "'message.format' is an MQMD Format of at most 8 characters",
414            ));
415        }
416        let sink = Self {
417            connection: ConnectionConfig::from_config(config)?,
418            queue: config
419                .get_non_empty("queue")?
420                .ok_or_else(|| ConnectorError::missing_config("queue"))?,
421            target_queue_manager: config.get_non_empty("target.queue.manager")?,
422            persistence: config.get_parsed("persistence")?.unwrap_or_default(),
423            priority,
424            expiry,
425            message_id_column: config.get_non_empty("message.id.column")?,
426            correlation_id_column: config.get_non_empty("correlation.id.column")?,
427            reply_to_queue: config.get_non_empty("reply.to.queue")?,
428            reply_to_queue_manager: config.get_non_empty("reply.to.queue.manager")?,
429            message_format,
430            ccsid: config.get_parsed("ccsid")?.unwrap_or(1208),
431            property_columns: config.get_list("property.columns"),
432            max_uow_messages: config.get_positive("max.uow.messages", DEFAULT_SINK_UOW_MESSAGES)?,
433            format: parse_sink_format(config)?,
434        };
435        let mut seen = std::collections::BTreeSet::new();
436        if let Some(duplicate) = sink
437            .property_columns()
438            .into_iter()
439            .find(|name| !seen.insert(*name))
440        {
441            return Err(conflict(&format!(
442                "column '{duplicate}' is mapped to more than one message field"
443            )));
444        }
445        Ok(sink)
446    }
447}
448
449impl ConnectionConfig {
450    fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
451        let endpoint = match (
452            config.get_non_empty("channel")?,
453            config.get_non_empty("connection.name")?,
454            config.get_non_empty("ccdt.url")?,
455        ) {
456            (Some(channel), Some(connection_name), None) => {
457                bounded("channel", &channel, 20)?;
458                bounded("connection.name", &connection_name, 264)?;
459                Endpoint::Channel {
460                    channel,
461                    connection_name,
462                }
463            }
464            (None, None, Some(url)) => Endpoint::Ccdt(url),
465            (None, None, None) => Endpoint::ClientEnvironment,
466            (_, _, Some(_)) => {
467                return Err(conflict(
468                    "set 'channel'/'connection.name' or 'ccdt.url', not both",
469                ))
470            }
471            _ => {
472                return Err(conflict(
473                    "'channel' and 'connection.name' must be set together",
474                ))
475            }
476        };
477        let username = config.get_non_empty("username")?;
478        let password = config.get("password").map(str::to_string);
479        if password.is_some() && username.is_none() {
480            return Err(conflict("'password' requires 'username'"));
481        }
482        let application_name = config
483            .get_non_empty("application.name")?
484            .unwrap_or_else(|| "laminardb".to_string());
485        bounded("application.name", &application_name, 28)?;
486        let queue_manager = config.get_non_empty("queue.manager")?;
487        if let Some(name) = &queue_manager {
488            bounded("queue.manager", name, 48)?;
489        }
490        Ok(Self {
491            queue_manager,
492            endpoint,
493            username,
494            password,
495            application_name,
496            tls: tls(config)?,
497            reconnect: config.get_parsed("reconnect")?.unwrap_or_default(),
498            client_library: config.get_non_empty("client.library")?.map(PathBuf::from),
499            operation_timeout: config.get_millis(
500                "operation.timeout.ms",
501                DEFAULT_OPERATION_TIMEOUT,
502                MAX_OPERATION_TIMEOUT,
503            )?,
504        })
505    }
506}
507
508fn tls(config: &ConnectorConfig) -> Result<Option<TlsConfig>, ConnectorError> {
509    let cipher_spec = config.get_non_empty("tls.cipher.spec")?;
510    let others = [
511        "tls.key.repository",
512        "tls.key.repository.password",
513        "tls.certificate.label",
514        "tls.peer.name",
515    ];
516    let Some(cipher_spec) = cipher_spec else {
517        if let Some(key) = others.iter().find(|key| config.get(key).is_some()) {
518            return Err(conflict(&format!("'{key}' requires 'tls.cipher.spec'")));
519        }
520        return Ok(None);
521    };
522    if config.get_non_empty("ccdt.url")?.is_some() {
523        return Err(conflict(
524            "with 'ccdt.url' the channel definition owns its CipherSpec and peer name",
525        ));
526    }
527    bounded("tls.cipher.spec", &cipher_spec, 32)?;
528    Ok(Some(TlsConfig {
529        cipher_spec,
530        key_repository: config.get_non_empty("tls.key.repository")?,
531        key_repository_password: config
532            .get("tls.key.repository.password")
533            .map(str::to_string),
534        certificate_label: config.get_non_empty("tls.certificate.label")?,
535        peer_name: config.get_non_empty("tls.peer.name")?,
536    }))
537}
538
539fn check_priority(priority: i32) -> Result<i32, ConnectorError> {
540    if (0..=9).contains(&priority) {
541        Ok(priority)
542    } else {
543        Err(conflict("'priority' must be between 0 and 9"))
544    }
545}
546
547fn reject_unknown(
548    config: &ConnectorConfig,
549    groups: &[&[&str]],
550    direction: &str,
551) -> Result<(), ConnectorError> {
552    let allowed: Vec<&str> = groups
553        .iter()
554        .flat_map(|group| group.iter().copied())
555        .collect();
556    config.reject_unknown_properties(&allowed, &format!("IBM MQ {direction}"))
557}
558
559fn format_option(config: &ConnectorConfig) -> Result<Format, ConnectorError> {
560    Ok(config
561        .get("format")
562        .map(Format::from_str)
563        .transpose()
564        .map_err(|error| ConnectorError::ConfigurationError(error.to_string()))?
565        .unwrap_or(Format::Json))
566}
567
568/// Sources also read FORMAT FPML: one complete document per message.
569fn parse_source_format(config: &ConnectorConfig) -> Result<Format, ConnectorError> {
570    match format_option(config)? {
571        format @ (Format::Json | Format::Csv | Format::Raw) => Ok(format),
572        #[cfg(feature = "fpml")]
573        Format::Fpml => Ok(Format::Fpml),
574        #[cfg(not(feature = "fpml"))]
575        Format::Fpml => Err(ConnectorError::ConfigurationError(
576            "format 'fpml' requires a build with the fpml feature".into(),
577        )),
578        other => Err(ConnectorError::FeatureUnsupported(format!(
579            "IBM MQ sources support FORMAT JSON, CSV, RAW, and FPML; {other} is not qualified \
580             for them"
581        ))),
582    }
583}
584
585fn parse_sink_format(config: &ConnectorConfig) -> Result<Format, ConnectorError> {
586    match format_option(config)? {
587        format @ (Format::Json | Format::Csv | Format::Raw) => Ok(format),
588        other => Err(ConnectorError::FeatureUnsupported(format!(
589            "IBM MQ sinks support FORMAT JSON, CSV, and RAW; {other} is not qualified for them"
590        ))),
591    }
592}
593
594fn bounded(key: &str, value: &str, max: usize) -> Result<(), ConnectorError> {
595    if value.len() > max {
596        return Err(conflict(&format!(
597            "'{key}' exceeds the {max}-character MQI field"
598        )));
599    }
600    Ok(())
601}
602
603fn conflict(message: &str) -> ConnectorError {
604    ConnectorError::ConfigurationError(message.to_string())
605}
606
607#[cfg(test)]
608mod tests;
laminar_connectors/ibmmq/config/mod.rs

laminar_connectors/ibmmq/config/
mod.rs
