Rust ORM integration

For new Rust code, the native honker crate is the clean path. If you already have a sqlx app, load the SQLite extension and call the same honker_* functions inside your existing executors and transactions.

Rust: use the honker crate

If you’re writing a new Rust app, skip this section. The honker crate wraps rusqlite and already gives you typed Queue, Stream, and Database::transaction() primitives. There is nothing to integrate — it is the integration.

use honker::Database;

let db = Database::open("app.db")?;
let q = db.queue("emails", Default::default());
q.enqueue(&serde_json::json!({"to": "alice@example.com"}), Default::default())?;

See the Getting Started guide (Rust tab) for the full native API. Use sqlx below only if you’re retrofitting into an existing sqlx-based codebase.

sqlx

Wiring

use sqlx::sqlite::SqliteConnectOptions;
use sqlx::{ConnectOptions, Executor};
use std::str::FromStr;

// The Rust crate links honker-core directly, so it has no bundled .so
// to point at. Build the extension with
// `cargo build --release -p honker-extension`, or download it from a
// release: https://github.com/russellromney/honker/releases
let ext = std::env::var("HONKER_EXTENSION_PATH")?;

let opts = SqliteConnectOptions::from_str("sqlite://app.db")?
    .extension(ext);

let mut conn = opts.connect().await?;
conn.execute("SELECT honker_bootstrap()").await?;

sqlx enables SQLite’s C load-extension API only while the requested extensions load, then disables it — so SQL-injected load_extension() calls can’t fire later in the session.

Wrapper

src/honker_ext.rs

use serde::Serialize;
use sqlx::sqlite::SqliteExecutor;

pub struct Queue {
    pub name: String,
    pub max_attempts: i32,
}

impl Queue {
    pub async fn enqueue<'e, E, P>(
        &self,
        executor: E,
        payload: &P,
        delay: Option<i32>,
        priority: i32,
    ) -> sqlx::Result<i64>
    where
        E: SqliteExecutor<'e>,
        P: Serialize,
    {
        let payload_json = serde_json::to_string(payload)
            .map_err(|e| sqlx::Error::Encode(Box::new(e)))?;
        let (id,): (i64,) = sqlx::query_as(
            "SELECT honker_enqueue($1, $2, NULL, $3, $4, $5, NULL)"
        )
        .bind(&self.name)
        .bind(payload_json)
        .bind(delay)
        .bind(priority)
        .bind(self.max_attempts)
        .fetch_one(executor)
        .await?;
        Ok(id)
    }
}

pub async fn notify<'e, E, P>(executor: E, channel: &str, payload: Option<&P>) -> sqlx::Result<()>
where
    E: SqliteExecutor<'e>,
    P: Serialize,
{
    let body = payload
        .map(serde_json::to_string)
        .transpose()
        .map_err(|e| sqlx::Error::Encode(Box::new(e)))?;
    sqlx::query("SELECT notify($1, $2)")
        .bind(channel)
        .bind(body)
        .execute(executor)
        .await?;
    Ok(())
}

SqliteExecutor<'e> is sqlx’s “anything you can run a query against” — satisfied by &SqlitePool, &mut SqliteConnection, and &mut Transaction<'_, Sqlite>, so one helper serves transactional and non-transactional callers.

SeaORM

SeaORM sits on sqlx and exposes map_sqlx_sqlite_opts, so extension loading stays connection-local — every connection the pool opens gets Honker.

Wiring

use sea_orm::{ConnectOptions, Database};

let ext = honker::extension_path()?;      // or your own path
let mut opt = ConnectOptions::new("sqlite://app.db?mode=rwc");
opt.map_sqlx_sqlite_opts(move |opts| unsafe { opts.extension(ext.clone()) });

let db = Database::connect(opt).await?;
db.execute_unprepared("SELECT honker_bootstrap()").await?;

Business write plus enqueue, atomically

honker_enqueue is ordinary SQL, so it commits with the transaction SeaORM owns. Roll back and the job is gone with the row.

use sea_orm::{DbBackend, Statement, TransactionTrait};

let job_id = db.transaction::<_, i64, sea_orm::DbErr>(|txn| Box::pin(async move {
    txn.execute_raw(Statement::from_sql_and_values(
        DbBackend::Sqlite,
        "INSERT INTO orders (id, user_id) VALUES (?, ?)",
        [42i32.into(), 1i32.into()],
    )).await?;

    let row = txn.query_one_raw(Statement::from_sql_and_values(
        DbBackend::Sqlite,
        "SELECT honker_enqueue(?, ?, NULL, NULL, ?, ?, NULL) AS id",
        ["emails".into(), r#"{"order_id":42}"#.into(), 0i64.into(), 3i64.into()],
    )).await?.unwrap();

    row.try_get_by_index::<i64>(0)
})).await?;

Diesel

Diesel 2.3 does not expose its raw SQLite handle, so there is no per-connection hook to load an extension into. The route that works is SQLite’s sqlite3_auto_extension: register Honker, open the connection, then cancel the registration so it does not apply process-wide to unrelated connections.

Keep the loaded library alive until after the connection closes — dropping it first unloads the code the connection is still using.

Wiring

use diesel::sqlite::SqliteConnection;
use diesel::{Connection, RunQueryDsl, sql_query};
use libloading::Library;

let library = unsafe { Library::new(honker::extension_path()?)? };
let init = unsafe { *library.get::<ExtensionInit>(b"sqlite3_honkerext_init")? };

unsafe { libsqlite3_sys::sqlite3_auto_extension(Some(init)) };
let mut conn = SqliteConnection::establish("app.db")?;
unsafe { libsqlite3_sys::sqlite3_cancel_auto_extension(Some(init)) };

sql_query("SELECT honker_bootstrap()").execute(&mut conn)?;
// `library` must outlive `conn`

macOS

Apple’s system SQLite disables sqlite3_auto_extension, so this route does not work against it. Link both Diesel and Honker against the same upstream SQLite — Homebrew’s, for example.

Toasty

Toasty does not currently expose SQLite extension loading on the connection it owns, so Honker cannot join a Toasty transaction. Use sqlx, SeaORM, or Diesel for the transactional path until Toasty provides that hook.