MongoDB component

The mongodb component inserts, queries, updates and deletes documents in MongoDB, and consumes change streams and tailable cursors.

Scheme: mongodb: · Crate: cameleer-datastores · Cargo feature: datastores · Consumer: yes · Producer: yes · Apache Camel: MongoDB

When to use it

• Store JSON documents from HTTP, Kafka or files without defining a schema first.
• Give an AI agent a memory: save notes, tool results or conversation turns per session and read the latest ones back before the next model call.
• React to changes: a change stream consumer routes every insert, update and delete.
• Run aggregations and bulk writes from a route.

URI format

mongodb:<connectionBean>?database=<db>&collection=<coll>&operation=<op>[&options]

The connection is one of:

• connectionUriString=mongodb://host:27017 (or mongodb+srv://cluster.example.com, resolved with Cameleer's own DNS client), the path is then just a name;
• hosts=host1:27017,host2:27017 with username, password and authSource;
• a MongoClient bound in the registry under the path name.

Put credentials in username and password rather than in connectionUriString: a connection string with its own ? options does not fit inside the endpoint URI.

Bodies are documents: maps, JSON text (Extended JSON and shell syntax such as ObjectId("...") are accepted) or lists of them. Results use Extended JSON for BSON-only types ({"$oid": "..."}, {"$date": "..."}).

Example

POST /notes stores the JSON body as a document and returns it with its generated _id. An agent can call this as a "remember" tool.

- route:
    id: save-note
    from:
      uri: "platform-http:/notes?httpMethodRestrict=POST"
      steps:
        - unmarshal:
            json: {}
        - to: "mongodb:agentDb?connectionUriString={{env:MONGODB_URI:mongodb://localhost:27017}}&database=agent&collection=notes&operation=insert"
        - marshal:
            json: {}

GET /notes?session=s1 returns the 20 newest notes of one session. The query document comes from the body; sort and limit come from headers.

- route:
    id: recent-notes
    from:
      uri: "platform-http:/notes?httpMethodRestrict=GET"
      steps:
        - filter:
            simple: "${header.session} !regex '^[A-Za-z0-9_-]+$'"
            steps:
              - setHeader:
                  name: CamelHttpResponseCode
                  constant: 400
              - setBody:
                  constant: "invalid session"
              - stop: {}
        - setBody:
            simple: '{"session": "${header.session}"}'
        - setHeader:
            name: CamelMongoDbSortBy
            constant: '{"_id": -1}'
        - setHeader:
            name: CamelMongoDbLimit
            constant: 20
        - to: "mongodb:agentDb?connectionUriString={{env:MONGODB_URI:mongodb://localhost:27017}}&database=agent&collection=notes&operation=findAll"
        - marshal:
            json: {}

The filter rejects session ids that are not plain words, so a request cannot inject query operators into the JSON.

Rust DSL

ctx.add_routes(|b| {
    b.from("direct:remember")
        .to("mongodb:agentDb?connectionUriString=mongodb://localhost:27017&database=agent&collection=notes&operation=insert");
    b.from("mongodb:agentDb?connectionUriString=mongodb://localhost:27017&database=agent&collection=notes&consumerType=changeStreams")
        .log("${header.CamelMongoDbStreamOperationType}: ${body}");
})?;

Operations

Options

Authentication supports SCRAM-SHA-256, SCRAM-SHA-1 and MONGODB-X509. Replica sets are discovered with hello and followed after a primary change. readPreference, writeConcern, collectionIndex, outputType and the persistent tail-tracking options are accepted and ignored; tail tracking is kept in memory only.

Message headers

Run it with Docker

docker run --rm -p 8080:8080 -e MONGODB_URI=mongodb://mongo.example.com:27017 \
  -v "$PWD:/routes:ro" pmsudhi/cameleer run /routes/notes.camel.yaml

Common questions

How do I connect with a username and password?

Use hosts=mongo.example.com:27017&username=app&password={{env:MONGODB_PASSWORD}}&authSource=admin instead of putting them in connectionUriString.

How do I sort and limit a query?

Set the headers CamelMongoDbSortBy (for example {"_id": -1}) and CamelMongoDbLimit before a findAll.

How do I react to new documents?

Consume with consumerType=changeStreams; add streamFilter={"operationType":"insert"} to see inserts only. Change streams need a replica set on the server.

Does it work with Azure Cosmos DB or FerretDB?

It speaks the MongoDB wire protocol (OP_MSG) and was verified against FerretDB 1.24 and an embedded server. Other compatible servers are not tested.

Related

• mongodb-gridfs: store files in MongoDB.
• couchdb, elasticsearch: other document stores.
• Apache Camel MongoDB component

Found a problem?

If something on this page doesn't match what Cameleer does, or you hit a bug while building with it, please help fix it, whether you are a person or a coding agent: open an issue that describes what you saw, then fork the repository, fix it, add or update a test, run cargo test --workspace, and open a pull request. CONTRIBUTING.md explains the ground rules.