# camel-component-sql
> SQL component for rust-camel integration framework
## Overview
The SQL component provides comprehensive database integration for rust-camel, supporting both producer (query execution) and consumer (polling) patterns. It enables seamless interaction with SQL databases using parameterized queries, batch operations, and streaming results.
## Features
- **Producer Mode**: Execute SQL queries (SELECT, INSERT, UPDATE, DELETE)
- **Consumer Mode**: Poll database tables for new rows
- **Parameter Binding**: Named (`:#name`), positional (`#`), IN clause (`:#in:ids`), and expressions (`:#${body.field}`)
- **Batch Operations**: Execute multiple inserts/updates in a single transaction
- **Streaming Results**: Stream large result sets as NDJSON without loading all rows into memory
- **Post-Processing Hooks**: `onConsume`, `onConsumeFailed`, `onConsumeBatchComplete` for consumer workflows
- **Connection Pooling**: Configurable pool with min/max connections, idle timeout, and max lifetime
- **Async Native**: Built on `tokio` and `sqlx`
- **Health Check**: Async connection validation probe for SQL databases
## Installation
Add to your `Cargo.toml`:
```toml
[dependencies]
camel-component-sql = "*"
```
## URI Format
```
sql:<query>?datasource=<name>&options
sql:<query>?db_url=<database-url>&options
sql:file:<path-to-sql-file>?datasource=<name>&options
sql:file:<path-to-sql-file>?db_url=<database-url>&options
```
- `<query>`: SQL query with optional parameter placeholders
- `<path-to-sql-file>`: Path to a file containing the SQL query
- `<datasource>`: Named datasource from Camel.toml (alternative to `db_url`)
- `<database-url>`: Database connection URL
Either `datasource` or `db_url` is required (not both).
## URI Options
### Connection Options
| `datasource` | - | Named datasource from Camel.toml (alternative to `db_url`) |
| `db_url` | (required if no `datasource`) | Database connection URL |
| `maxConnections` | `5` | Maximum connections in pool |
| `minConnections` | `1` | Minimum connections in pool |
| `idleTimeoutSecs` | `300` | Idle connection timeout in seconds |
| `maxLifetimeSecs` | `1800` | Maximum connection lifetime in seconds |
### Named Datasources
Instead of embedding `db_url` in every URI, define named datasources in `Camel.toml`:
```toml
[default.datasources.orders]
db_url = "postgres://user:pass@db:5432/orders"
max_connections = 20
min_connections = 2
[default.datasources.analytics]
db_url = "postgres://user:pass@db:5432/analytics"
max_connections = 5
```
Then reference by name in URIs:
```
sql:SELECT * FROM orders?datasource=orders
sql:INSERT INTO events VALUES (#, #)?datasource=analytics
```
**Benefits:**
- Shared connection pools across endpoints referencing the same datasource
- Credentials not exposed in route URIs
- Pool configuration centralized in one place
- Health checks per datasource via `PoolFactory::check()`
**Restrictions:** When using `datasource`, the following pool-affecting URI parameters are rejected (set them in `Camel.toml` instead):
`maxConnections`, `minConnections`, `idleTimeoutSecs`, `maxLifetimeSecs`, `sslMode`, `sslRootCert`, `sslCert`, `sslKey`
Non-pool parameters like `outputType`, `onConsume`, `delay` etc. work normally with named datasources.
**Backward compatible:** Inline `db_url` continues to work as before.
### Query Options
| `outputType` | `SelectList` | Output format: `SelectList`, `SelectOne`, `StreamList` |
| `placeholder` | `#` | Character used for parameter placeholders |
| `inSeparator` | `", "` | Separator for IN clause expansion (e.g. `", "` → `1, 2, 3`) |
| `noop` | `false` | If true, preserve original body after DML operations |
### Consumer Options
| `delay` | `500` | Polling delay in milliseconds |
| `initialDelay` | `1000` | Initial delay before first poll (ms) |
| `maxMessagesPerPoll` | - | Maximum rows to process per poll |
| `onConsume` | - | SQL to execute after successful row processing |
| `onConsumeFailed` | - | SQL to execute after failed row processing |
| `onConsumeBatchComplete` | - | SQL to execute after batch completes |
| `routeEmptyResultSet` | `false` | Process empty result sets |
| `useIterator` | `true` | Process rows individually (true) or as batch (false) |
| `expectedUpdateCount` | - | Expected rows affected (error if mismatch) |
| `breakBatchOnConsumeFail` | `false` | Stop batch processing on failure |
### Producer Options
| `batch` | `false` | Enable batch mode (body must be array of arrays) |
| `useMessageBodyForSql` | `false` | Use message body as SQL query |
### SSL/TLS Options
| `sslMode` | - | SSL mode (e.g. `require`, `verify-ca`, `verify-full`) |
| `sslRootCert` | - | Path to CA certificate for server verification |
| `sslCert` | - | Path to client certificate for mTLS |
| `sslKey` | - | Path to client private key for mTLS |
SSL parameters are appended as query string parameters to the database URL, with database-specific key names:
| `sslMode` | `sslmode` | `ssl-mode` |
| `sslRootCert` | `sslrootcert` | `ssl-ca` |
| `sslCert` | `sslcert` | `ssl-cert` |
| `sslKey` | `sslkey` | `ssl-key` |
## Parameter Binding
The SQL component supports multiple parameter placeholder styles:
### Positional Parameters (`#`)
```sql
INSERT INTO users (name, age) VALUES (#, #)
```
Body must be a JSON array: `["Alice", 30]`
### Named Parameters (`:#name`)
```sql
SELECT * FROM users WHERE id = :#id AND status = :#status
```
Values resolved from body (if JSON object), headers, or properties.
### IN Clause (`:#in:name`)
```sql
SELECT * FROM users WHERE id IN (:#in:ids)
```
Value must be a JSON array: `[1, 2, 3]` → expands to `IN ($1, $2, $3)`.
Customize the separator with `inSeparator` (default `", "`):
```
sql:SELECT * FROM users WHERE id IN (:#in:ids)?...&inSeparator=|
```
Produces: `IN ($1|$2|$3)`
### Expression Parameters (`:#${expr}`)
```sql
SELECT * FROM users WHERE id = :#${body.user_id} AND name = :#${header.userName}
```
Supported expressions: `body.field`, `header.name`, `property.key`
## Usage
### SELECT Query
```rust
use camel_builder::RouteBuilder;
use camel_component_sql::SqlComponent;
let mut ctx = CamelContext::new();
ctx.register_component(SqlComponent::new());
let route = RouteBuilder::from("direct:get-users")
.to("sql:SELECT * FROM users?db_url=postgres://localhost/mydb")
.build()?;
ctx.add_route_definition(route).await?;
```
### Parameterized Query
```rust
// Named parameters from body
let route = RouteBuilder::from("direct:get-user")
.set_body(Body::Json(serde_json::json!({"id": 42})))
.to("sql:SELECT * FROM users WHERE id = :#id?db_url=postgres://localhost/mydb")
.build()?;
// Positional parameters from array body
let route = RouteBuilder::from("direct:insert")
.set_body(Body::Json(serde_json::json!(["Alice", 30])))
.to("sql:INSERT INTO users (name, age) VALUES (#, #)?db_url=postgres://localhost/mydb")
.build()?;
```
### Streaming Large Results
```rust
// Stream results as NDJSON (memory efficient for large datasets)
let route = RouteBuilder::from("direct:export")
.to("sql:SELECT * FROM large_table?db_url=postgres://localhost/mydb&outputType=StreamList")
.to("file:/tmp/export.ndjson")
.build()?;
```
### Batch Insert
```rust
// Batch insert with transaction
let route = RouteBuilder::from("direct:batch-insert")
.set_body(Body::Json(serde_json::json!([
["Alice", 30],
["Bob", 25],
["Charlie", 35]
])))
.to("sql:INSERT INTO users (name, age) VALUES (#, #)?db_url=postgres://localhost/mydb&batch=true")
.build()?;
```
### Dynamic Query from Body
```rust
// Use message body as SQL
let route = RouteBuilder::from("direct:dynamic")
.set_body(Body::Text("SELECT COUNT(*) FROM users".to_string()))
.to("sql:placeholder?db_url=postgres://localhost/mydb&useMessageBodyForSql=true")
.build()?;
```
### Polling Consumer
```rust
// Poll for pending orders and mark as processed
let route = RouteBuilder::from(
"sql:SELECT * FROM orders WHERE status = 'pending'?\
db_url=postgres://localhost/mydb\
&delay=5000\
&onConsume=UPDATE orders SET status = 'processed' WHERE id = :#id\
&onConsumeFailed=UPDATE orders SET status = 'failed' WHERE id = :#id"
)
.process(|ex| async move {
// Process each order
Ok(ex)
})
.build()?;
```
### Load Query from File
```rust
// Load SQL from external file
let route = RouteBuilder::from("direct:query")
.to("sql:file:/etc/queries/get-users.sql?db_url=postgres://localhost/mydb")
.build()?;
```
## Exchange Headers
### Input Headers
| `CamelSql.Query` | Override the SQL query from URI |
| `CamelSql.Parameters` | Override parameters as JSON array |
### Output Headers
| `CamelSql.RowCount` | out | Number of rows returned by SELECT |
| `CamelSql.UpdateCount` | out | Number of rows affected by INSERT/UPDATE/DELETE |
### Consumer Row Headers
When `useIterator=true`, each row's columns are also set as headers:
| `CamelSql.<column>` | Column value from current row (e.g., `CamelSql.id`, `CamelSql.name`) |
## Output Types
| `SelectList` | All rows as array | `[{...}, {...}]` |
| `SelectOne` | First row only | `{...}` or empty |
| `StreamList` | Stream rows on demand | NDJSON stream (`{...}\n{...}\n`) |
## Example: Order Processing Pipeline
```rust
use camel_builder::RouteBuilder;
use camel_component_sql::SqlComponent;
use camel_core::CamelContext;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let mut ctx = CamelContext::new();
ctx.register_component(SqlComponent::new());
// Producer: Insert new orders
let producer = RouteBuilder::from("direct:create-order")
.process(|ex| async move {
// Transform input to order record
Ok(ex)
})
.to("sql:INSERT INTO orders (customer_id, total) VALUES (#, #)?db_url=postgres://localhost/mydb&expectedUpdateCount=1")
.build()?;
// Consumer: Process pending orders
let consumer = RouteBuilder::from(
"sql:SELECT * FROM orders WHERE status = 'pending' ORDER BY created_at?\
db_url=postgres://localhost/mydb\
&delay=2000\
&maxMessagesPerPoll=10\
&onConsume=UPDATE orders SET status = 'completed', processed_at = NOW() WHERE id = :#id"
)
.process(|ex| async move {
let order_id = ex.input.header("CamelSql.id").and_then(|v| v.as_i64());
println!("Processing order: {:?}", order_id);
Ok(ex)
})
.build()?;
ctx.add_route_definition(producer).await?;
ctx.add_route_definition(consumer).await?;
ctx.start().await?;
Ok(())
}
```
## Global Configuration
Configure default connection pool settings in `Camel.toml` that apply to all SQL endpoints:
```toml
[default.components.sql]
max_connections = 5 # Maximum pool connections (default: 5)
min_connections = 1 # Minimum pool connections (default: 1)
idle_timeout_secs = 300 # Idle connection timeout (default: 300)
max_lifetime_secs = 1800 # Max connection lifetime (default: 1800)
ssl_mode = "require" # SSL mode (optional)
ssl_root_cert = "/etc/ssl/ca.pem" # CA certificate path (optional)
ssl_cert = "/etc/ssl/client.pem" # Client certificate path (optional)
ssl_key = "/etc/ssl/client.key" # Client key path (optional)
```
URI parameters always override global defaults:
```rust
// Uses global pool settings
.to("sql:SELECT * FROM users?db_url=postgres://localhost/mydb")
// Overrides maxConnections from global config
.to("sql:SELECT * FROM users?db_url=postgres://localhost/mydb&maxConnections=10")
```
### Profile-Specific Configuration
```toml
[default.components.sql]
max_connections = 5
min_connections = 1
[production.components.sql]
max_connections = 20
min_connections = 5
idle_timeout_secs = 600
```
## Health Check
The `camel-component-sql` component registers an async health check via `AsyncHealthCheck`.
- **Probe**: Executes a `SELECT 1` validation query against the connection pool
- **Healthy**: Connection pool returns a valid connection and query succeeds
- **Degraded**: Connection fails or validation query times out
Health checks are exposed via the health server:
```toml
[observability.health]
enabled = true
port = 8080
```
## Documentation
- [API Documentation](https://docs.rs/camel-component-sql)
- [Repository](https://github.com/kennycallado/rust-camel)
## License
Apache-2.0
## Contributing
Contributions are welcome! Please see the [main repository](https://github.com/kennycallado/rust-camel) for details.