SQL component

The sql component runs SQL queries and updates against PostgreSQL or the embedded in-memory database, and polls tables for new rows.

Scheme: sql: · Crate: cameleer-sql · Cargo feature: sql · Consumer: yes · Producer: yes · Apache Camel: SQL

When to use it

• Insert or update rows from a route: save orders received over HTTP, keep an audit table of the decisions an AI agent made, store an LLM's classification next to the ticket.
• Poll a table as a work queue: pick up rows where processed = false, route each one, and mark it done with onConsume.
• Look up reference data in the middle of a route (outputType=SelectOne).
• Prototype without a database server: jdbc:rustsql:mem:<name> is an embedded in-memory SQL engine with the PostgreSQL dialect, so the same routes later run against PostgreSQL.

URI format

sql:<statement>[?options]
sql:file:<path-to-sql-file>[?options]

The statement uses Camel's parameter syntax:

Choosing the database

dataSource names the database. It is either a JDBC-style URL or a registry bean:

Without dataSource, the endpoint uses the only data source in the registry. Cameleer binds one named dataSource from the camel.datasource.url property (with camel.datasource.username and camel.datasource.password), which the container also reads from the CAMEL_DATASOURCE_URL environment variable. camel.datasource.<name>.url binds more data sources by name. Other databases (MySQL, Oracle, SQL Server) are not supported: there are no JDBC drivers, only these two engines.

A PostgreSQL data source is a connection pool, sized in its URL like HikariCP: ?poolSize=16 opens at most 16 connections (default 8) and reuses them. When all of them are busy, a caller waits for one to come back, up to poolTimeout seconds (default 30), then fails with Connection is not available. Size the pool for the database, not for the number of concurrent requests: PostgreSQL handles a few busy connections better than many.

Example

This route polls a table as a work queue: every new order is logged once and marked as processed. A one-shot timer route creates the table first, so the file runs as is against the embedded database or against PostgreSQL.

- route:
    id: create-schema
    from:
      uri: "timer:schema?repeatCount=1&delay=0"
      steps:
        - to: "sql:create table if not exists orders (id serial primary key, item text, qty int, processed boolean default false)?dataSource={{env:DB_URL:jdbc:rustsql:mem:shop}}"
- route:
    id: new-orders
    from:
      uri: "sql:select * from orders where processed = false order by id?dataSource={{env:DB_URL:jdbc:rustsql:mem:shop}}&onConsume=update orders set processed = true where id = :#id&delay=2000"
      steps:
        - log: "new order ${body[id]}: ${body[qty]} x ${body[item]}"

Each row arrives as a map (${body[item]}), one exchange per row. onConsume runs once per row after the route finished it, with the row's columns as :# parameters.

This route saves JSON orders posted over HTTP. The fields of the JSON object fill the :#item and :#qty parameters, and the response reports the number of inserted rows.

- route:
    id: save-order
    from:
      uri: "platform-http:/orders?httpMethodRestrict=POST"
      steps:
        - unmarshal:
            json: {}
        - to: "sql:insert into orders (item, qty) values (:#item, :#qty)?dataSource={{env:DB_URL:jdbc:rustsql:mem:shop}}"
        - setBody:
            simple: "inserted ${header.CamelSqlUpdateCount}"

Rust DSL

ctx.add_routes(|b| {
    b.from("direct:save-order")
        .to("sql:insert into orders (item, qty) values (:#item, :#qty)?dataSource=jdbc:rustsql:mem:shop");
    b.from("direct:find-order")
        .to("sql:select item, qty from orders where id = :#${header.id}?dataSource=jdbc:rustsql:mem:shop&outputType=SelectOne");
})?;

Options

outputClass, processingStrategy, prepareStatementStrategy, alwaysPopulateStatement and rowMapperFactory are accepted for compatibility and ignored: rows are always maps.

Message headers

The consumer sets the exchange properties CamelBatchIndex, CamelBatchSize and CamelBatchComplete.

Run it with Docker

Against the embedded database (data is lost when the container stops):

docker run --rm -p 8080:8080 -v "$PWD:/routes:ro" pmsudhi/cameleer run /routes/orders.camel.yaml

Against PostgreSQL:

docker run --rm -p 8080:8080 \
  -e DB_URL='jdbc:postgresql://camel:secret@db.example.com:5432/shop?sslmode=require' \
  -v "$PWD:/routes:ro" pmsudhi/cameleer run /routes/orders.camel.yaml

Common questions

How do I connect to PostgreSQL?

Set dataSource=jdbc:postgresql://user:password@host:5432/db, or set camel.datasource.url (environment variable CAMEL_DATASOURCE_URL) and leave dataSource out. URL-encode special characters in the password.

How do I get the id of an inserted row?

Set the header CamelSqlRetrieveGeneratedKeys to true. The keys come back in CamelSqlGeneratedKeysRows, for example [{id=7}]. On PostgreSQL you can also write insert ... returning id and read the result as a query.

How do I process each row only once?

Use onConsume to mark the row (update ... set processed = true where id = :#id) and filter on that column in the polling query. Add transacted=true to roll the marks back when a row fails.

Does it support MySQL or Oracle?

No. Cameleer has no JDBC drivers: it speaks the PostgreSQL wire protocol (tested against PostgreSQL 16) and embeds its own engine.

Related

• sql-stored: call stored procedures and functions.
• jdbc: run SQL taken from the message body.
• Apache Camel SQL 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.