"AimDB: Bring Your Own Connector"

A few posts ago we described connectors as the edges of the dataflow graph, the points where a typed record crosses out of one process and into the world. link_to streams a record out, link_from brings one in. Same builder, same buffers, opposite directions.

That post told one story: bring your own protocol bridge. MQTT, KNX, WebSocket — wrap a protocol library, declare a scheme, mirror records across the wire.

Since then a second story landed and it changes what "connector" means.

A connector no longer only bridges a record to a protocol. It can expose AimDB itself over a transport to any peer that dials in.

And the part that matters for you: adding a brand-new transport is a thin crate. You contribute three small methods. You inherit the wire protocol, reconnect, fan-out, security policy and remote introspection from the core.

This is what we mean by bring your own connector. Not "wire up another broker." Swap the entire transport under AimDB's remote-access layer without touching a single record or link.

Two kinds of connector, one registration spine

There are now two things the word connector covers. They feel different but they register the same way.

• Data-plane links — MQTT, KNX, WebSocket. A record opts in with link_to("mqtt://…") / link_from(…) and the connector mirrors that one record to or from an external topic. This is the publish path from the original post.
• Remote-access session connectors — UDS, serial, TCP. These don't mirror a single record. They stand AimDB up as a service over a transport, so a peer can introspect the whole database, subscribe to records and write to the ones you allow.

Both implement the same ConnectorBuilder trait and both register through one method:

// (1) data-plane link: mirror one record over MQTT
AimDbBuilder::new().runtime(rt)
    .with_connector(MqttConnector::new("mqtt://broker.local:1883"))
    .configure::<Temperature>("temp", |r| { r.link_from("mqtt://commands/temp"); })
    .build().await?;

// (2) remote-access SERVER: expose this whole db over a Unix socket — no links
AimDbBuilder::new().runtime(rt)
    .with_connector(UdsServer::new("/run/aimdb.sock").max_connections(32))
    .build().await?;

If you used an earlier version of AimDB, note what's gone: there is no with_remote_access(config) anymore. Standing up remote access is registering a connector. The transport became swappable, so the dedicated builder method dissolved into the connector spine.

The transport triple

Here's the move that makes a new transport cheap. Everything AimDB's remote-access layer needs from a transport is described by three traits, a framed, bidirectional pipe plus the two roles that produce one:

/// A framed, bidirectional pipe — role-neutral.
pub trait Connection: Send {
    /// Receive one logical frame. `Ok(None)` means the peer closed.
    fn recv(&mut self) -> BoxFut<'_, TransportResult<Option<Vec<u8>>>>;
    /// Send one logical frame.
    fn send<'a>(&'a mut self, frame: &'a [u8]) -> BoxFut<'a, TransportResult<()>>;
    /// Peer metadata (remote addr, headers, pre-resolved auth).
    fn peer(&self) -> &PeerInfo;
}

/// The accepting (server) side.
pub trait Listener: Send {
    fn accept(&mut self) -> BoxFut<'_, TransportResult<Box<dyn Connection>>>;
}

/// The initiating (client) side — the dual of `Listener`.
pub trait Dialer: Send {
    fn connect(&self) -> BoxFut<'_, TransportResult<Box<dyn Connection>>>;
}

That's the whole transport contract. recv hands back one logical frame. Framing is the transport's job, the layers above only ever see whole messages. Dialer dials out, Listener accepts, both yield the same role-neutral Connection.

How thin is a real implementation? Here is the entire Unix-domain-socket transport. Newline-delimited frames over a split UnixStream:

impl Connection for UdsConnection {
    fn recv(&mut self) -> BoxFut<'_, TransportResult<Option<Vec<u8>>>> {
        Box::pin(async move {
            let mut line = String::new();
            match self.reader.read_line(&mut line).await {
                Ok(0) => Ok(None),                  // EOF — peer closed
                Ok(_) => {
                    while matches!(line.as_bytes().last(), Some(b'\n' | b'\r')) {
                        line.pop();                 // strip the delimiter; codec owns the rest
                    }
                    Ok(Some(line.into_bytes()))
                }
                Err(_) => Err(TransportError::Io),
            }
        })
    }

    fn send<'a>(&'a mut self, frame: &'a [u8]) -> BoxFut<'a, TransportResult<()>> {
        Box::pin(async move {
            self.writer.write_all(frame).await.map_err(|_| TransportError::Closed)?;
            self.writer.write_all(b"\n").await.map_err(|_| TransportError::Closed)?;
            self.writer.flush().await.map_err(|_| TransportError::Closed)
        })
    }

    fn peer(&self) -> &PeerInfo { &self.peer }
}

Dialer connects the socket, Listener accepts on it — a few lines each, in the same vein. No protocol logic. No subscription handling. No reconnect loop. No serialization. Just "move bytes, find frame boundaries." That's the contract and that's all you write.

What the three methods buy you

The transport is the part you bring. Everything stacked on top is the part you inherit:

   your record / link code        ← never changes when the transport changes
 ──────────────────────────────
   Dispatch  (list/get/set/         ← reused: list, get, set, subscribe, drain,
              subscribe/drain)         producer-override safety, security policy
 ──────────────────────────────
   AimX codec (envelope framing)    ← reused: the wire format peers speak
 ──────────────────────────────
   Session engine (reconnect,       ← reused: accept loop, fan-out, keepalive,
                   fan-out, pumps)     per-connection sessions
 ──────────────────────────────
   Connection / Dialer / Listener   ← YOU bring this. ~40 lines.

The generic spine that joins them is two structs in core: SessionClientConnector<D, C> (the dialing half) and SessionServerConnector<C, LF, DF> (the accepting half). You inject a transport D/L and a codec C; they implement ConnectorBuilder and drive the engine. A transport crate wraps each in a one-line sugar constructor so the call site stays clean — UdsServer, UdsClient.

Standing up a server, with selective write access, looks like this:

// Read everything; allow writes only to AppSettings.
let mut policy = SecurityPolicy::read_write();
policy.allow_write_key("server::AppSettings");

let config = AimxConfig::uds_default()
    .socket_path("/run/aimdb.sock")
    .security_policy(policy)
    .max_connections(10)
    .max_subs_per_connection(32);

let mut builder = AimDbBuilder::new()
    .runtime(rt)
    .with_connector(UdsServer::from_config(config));

// A record is only reachable remotely if it opts in.
builder.configure::<Temperature>("server::Temperature", |reg| {
    reg.buffer(BufferCfg::SpmcRing { capacity: 100 })
        .with_remote_access()        // ← the gate: visible to peers
        .source(temperature_simulator)
        .tap(temperature_logger);
});

builder.configure::<AppSettings>("server::AppSettings", |reg| {
    reg.buffer(BufferCfg::SingleLatest).with_remote_access();
});

with_remote_access() is the per-record gate. No flag, no exposure — remote access is opt-in per record and writes are gated again by the security policy. The buffer choice carries across the wire exactly as it does in-process: a SpmcRing peer can drain history, a SingleLatest peer reads the canonical latest.

A peer then speaks the AimX surface with nothing but a connection handle:

let conn = AimxConnection::connect("/run/aimdb.sock").await?;

let records = conn.list_records().await?;             // discover the schema
let cfg     = conn.get_record("server::Config").await?; // point-in-time read
conn.set_record("server::AppSettings", json!({ ... })).await?; // gated write

let mut sub = conn.subscribe("server::Temperature")?;           // live stream
while let Some(reading) = sub.next().await { /* … */ }

Or, if the peer is itself an AimDB node, it skips the raw client and mirrors records with the same link_to / link_from from the original post — pointed at the uds:// scheme instead of mqtt://:

AimDbBuilder::new().runtime(rt)
    .with_connector(UdsClient::new("/run/aimdb.sock"))
    .configure::<Temperature>("temp", |r| {
        r.with_remote_access().link_from("uds://server::Temperature");
    })
    .build().await?;

Same builder. Same buffers. The seam moved from a broker to a socket and nothing else changed.

Swap the wire

Now the punchline. Everything above — the dispatch, the codec, the engine, the security policy, every record and link — is transport-agnostic. So to run AimDB's remote access over a serial line instead of a socket, you bring a different transport triple and change nothing else.

That's exactly what aimdb-serial-connector is. The same Dialer/Listener/Connection triple, framed with COBS (Consistent Overhead Byte Stuffing) and a 0x00 delimiter instead of a newline — self-synchronizing on a lossy, unframed serial medium. The codec, dispatch and session engine are reused verbatim from core. And because none of that needs std, the serial connector ships an Embassy half too: AimDB's full remote-access surface, over a UART, on a microcontroller.

Two transports. One wire protocol. One dispatch. The MCU on the end of a UART answers record.list and streams subscriptions with the same code path as the cloud node behind a socket. Bring your own connector means: pick the medium, write the three methods, inherit the rest.

The publish path evolved too

Two corrections for data-plane connectors — the MQTT/KNX/WebSocket kind — since the original post. Registration is a single argument now: there's no with_connector("mqtt", Arc::new(conn)), you pass MqttConnectorBuilder::new("mqtt://…") and its scheme() declares the prefix. And ConnectorBuilder::build no longer returns an Arc<dyn Connector> — it hands back the futures that drive your connector (the transport loop plus one publisher per route) and the runtime owns them. The outbound Connector::publish(destination, config, payload) itself is unchanged; only the assembly moved — which is the same shape the session connectors use, and why both kinds register through one method.

What ships today

The bottom row is the point. A data-plane connector is a protocol library you wrap behind publish + scheme + build. A remote-access connector is a Dialer/Listener/Connection triple you hand to the session spine. Either way, the runtime internals stay behind a trait and the hard part is the medium you're speaking — never AimDB's surface.

Want to see remote access end to end? The remote-access-demo in the repo runs a server that exposes five records over UDS — with a security policy, a writable record, drainable history and a live subscription — and a client that walks the whole AimX surface. Star it, clone it, run it. · Live Demo

Get Involved

AimDB is open source, Apache 2.0 and growing. If you've ever exposed an internal data model over a socket, a serial line and a network protocol — and written three different servers to do it — you already know why the transport should be the only part that changes.

Star AimDB on GitHub · Join the discussion · Read the docs

---

Spell checks and minor copy edits in this post were assisted by an LLM.