Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

MongoDB

Schemes: mongodb://, mongodb+srv://

Query parameters recognised as config fields for this connector. The object-typed tls is set with a JSON literal, e.g. ?tls={...}. Any other ?key=value pair is passed through unchanged as a driver option on the connection URL.

NameTypeRequiredDefaultDescription
capped_size_bytesintegerno—(Publisher only) If set, creates a capped collection with this size in bytes.
change_streambooleannofalse(Consumer only) Deprecated — use consume: capture_new. Kept for compatibility.
checkpoint_storestringno—(Consumer only) Where to persist the resume cursor in capture_new/capture_all mode. A URL selects the backend; a bare name (or /name) reuses the source database with that name: - absent → source database, collection mqb_cursors_<source_collection> (auto-unique) - /my_cursors → source database, collection my_cursors - file:///var/lib/mqb/cursors.json → local JSON file (read-only / write-restricted sources) - mongodb://host/db/collection → external MongoDB collection (collection optional) - postgres://user@host/db/table or mysql://host/db/table → external SQL table (table optional) - s3://bucket/prefix (also gs://, az://, abfs://) → cloud object store; creds via env When no collection/table is named, it defaults to mqb_cursors_<source_collection>. May embed connection credentials, so it is treated as a secret.
collectionstringno—The MongoDB collection name.
consumedelete | snapshot | capture_new | capture_allno—(Consumer only) How to consume the collection: capture_all (default, read existing documents first, then watch for changes — non-destructive, for bulk reads and ETL), capture_new (watch an existing collection for changes only), snapshot (one-shot non-destructive read that ends on drain, the option without a replica set), or delete (competing-consumers work queue — destructive and intended for jobs; also consumer). The bridge selects the underlying mechanism automatically. If unset, the deprecated change_stream boolean is honored for backward compatibility.
cursor_idstringno—The ID used for the cursor in sequenced mode. If not provided, consumption starts from the current sequence (ephemeral).
databasestringyes—The database name.
findstringno—(Publisher only) Extended-JSON filter template for lookup, e.g. {"_id": "${payload:id}"}. Answers with the first match. {"_id": {"$in": ["${payload:id}"]}} answers a batch at once.
formatnormal | json | text | rawnonormalFormat for storing messages. Defaults to Normal.
id_fieldstringno—(Publisher only) Top-level payload field whose value becomes the document _id, or a replay-stable ${...} template such as ${metadata:mqb.id}. Enables idempotent inserts through MongoDB’s unique _id index. Sink collections only.
meta_collectionstringno—Deprecated — unused since the publisher no longer writes seq. Kept for compatibility.
passwordstringno—Optional password. Takes precedence over any credentials embedded in the url. Use embedded URL credentials for simple one-off connections but prefer explicit username/password fields (or environment-sourced secrets) for clarity and secret management in production.
polling_interval_msintegerno—(Consumer only) Polling interval in milliseconds for the consumer (when not using Change Streams). Defaults to 100ms.
receive_querystringno—(Consumer only) Optional custom MongoDB query to filter messages. Provided as a JSON string (e.g., ‘{“type”: “notification”}’).
reply_polling_msintegerno—(Publisher only) Polling interval in milliseconds for the publisher when waiting for a reply. Defaults to 50ms.
report_outcomebooleannofalse(Publisher only) Return the message with metadata mongodb.outcome = inserted/existed (dup-key) so a request+switch can branch. Sink collections only; pair with id_field.
request_replybooleannofalse(Publisher only) If true, the publisher will wait for a response in a dedicated collection. Defaults to false.
request_timeout_msintegerno—(Publisher only) Timeout for request-reply operations in milliseconds. Defaults to 30000ms.
sharedbooleannotrueShare one MongoDB client per connection (default: true); false forces a dedicated client.
source_metadatabooleannofalse(Consumer only, capture_new/capture_all) Include authoritative mqb.src.mongodb_* source positions. Defaults to false.
tlsobjectnosee belowTLS configuration.
ttl_secondsintegerno—(Publisher only) TTL in seconds for documents created by the publisher. If set, a TTL index is created.
urlstringyes—MongoDB connection string URI. Can contain a comma-separated list of hosts for a replica set. If it contains userinfo, it will be treated as a secret. Credentials provided via the separate username and password fields take precedence over any credentials embedded in the URL.
usernamestringno—Optional username. Takes precedence over any credentials embedded in the url. Use embedded URL credentials for simple one-off connections but prefer explicit username/password fields (or environment-sourced secrets) for clarity and secret management in production.

Struct-typed fields

tls

TLS configuration for secure connections.

Configures Transport Layer Security (TLS/SSL) for encrypted communication. Supports both client certificate (mutual TLS) and server certificate validation.

Examples

use mq_bridge::models::TlsConfig;

let tls = TlsConfig {
    required: true,
    ca_file: Some("/path/to/ca.pem".to_string()),
    cert_file: Some("/path/to/cert.pem".to_string()),
    key_file: Some("/path/to/key.pem".to_string()),
    ..Default::default()
};
NameTypeRequiredDefaultDescription
accept_invalid_certsbooleannofalseIf true, disable server certificate verification (insecure).
ca_filestringno—Path to the CA certificate file.
cert_filestringno—Path to the client certificate file (PEM).
cert_passwordstringno—Password for the private key (if encrypted).
key_filestringno—Path to the client private key file (PEM).
requiredbooleannofalseIf true, enable TLS/SSL.