"Connectors: Where AimDB Meets the Real World"

A reactive pipeline moves records from a source through transforms to a tap. It's a complete story inside the runtime. But records don't appear from nowhere and they rarely stop where they're produced. Something brought them in. Something will carry them out.

That something is a connector.

Connectors aren't an integration layer bolted onto AimDB. They aren't an SDK you call from a tap. They're the edges of the dataflow graph, the points where a typed record crosses out of one process and into the world.

Once you look at them that way, two things become obvious.

The first: connectors are symmetric. There is no separate "ingress API" and "egress API". The same builder method, the same record type, the same runtime machinery handles both directions. You declare whether a record flows in or out, the connector does the rest.

The second: you've already made the most important decision. When you picked a buffer type for your record, you encoded how that record behaves.

Connectors extend that behavior across a network boundary. They don't change it.

Connectors Are Edges, Not Integrations

Most systems treat ingress and egress as different problems. You wire up an MQTT subscriber on one side of your service, a publisher on the other and write the glue that connects them to your domain types. Two SDKs, two error models, two serializers. The actual record sits somewhere in the middle, as a struct that nobody else in the system knows about.

AimDB inverts this. The record is the unit. The connector is a property of the record's key.

// A sensor node: produce temperature readings, publish over MQTT
builder.configure::<Temperature>("sensor::temp", |reg| {
    reg.buffer(BufferCfg::SpmcRing { capacity: 256 })
        .source(|ctx, producer| async move {
            loop {
                let reading = read_sensor().await;
                producer.send(Temperature::set(reading, ctx.time().now())).await.ok();
                ctx.sleep(Duration::from_secs(1)).await;
            }
        })
        .link_to("mqtt://sensors/temperature")
        .with_serializer_raw(Temperature::to_bytes)
        .finish();
});

link_to is one half of the API. link_from is the other:

// An edge gateway: receive temperature readings from MQTT, observe locally
builder.configure::<Temperature>("gateway::temp", |reg| {
    reg.buffer(BufferCfg::SingleLatest)
        .link_from("mqtt://sensors/temperature")
        .with_deserializer_raw(Temperature::from_bytes)
        .tap(log_tap::<Temperature>("edge"))
        .finish();
});

Same builder. Same record type. Opposite directions. Try it yourself →

link_to and link_from aren't operations you perform on a record, they're declarations about where the record lives in the world.

The Buffer Tells You What Kind of Edge

When you registered your record, you chose a buffer type. That choice already encoded how the record moves inside the runtime. Connectors don't change those semantics, they extend them across a boundary.

• SPMC Ring + connector — a bounded stream that crosses the wire. The same backpressure model that governs the buffer governs the connector.
• SingleLatest + connector — shared state. Configuration, feature flags, mirrored UI state. Only the current value matters; older values were never the truth, just earlier guesses at it.
• Mailbox + connector — instructions that arrive from elsewhere. Device commands. Setpoints. Actuation requests. The latest instruction supersedes the previous one, because the previous one is no longer what the sender means.

Mailbox is how one part of the graph hands a command to another inside a process. Pair it with link_from and the sender can be anywhere on the network.

The dataflow graph isn't just consuming the world. It's participating in it.

A Round-Trip in One Builder Block

The clearest way to see the symmetry is to watch a record go out and a different record come back. A sensor publishes a reading. A controller sends a setpoint in response.

#[derive(Clone, Debug, Serialize, Deserialize)]
struct Temperature { celsius: f32, timestamp: u64 }

#[derive(Clone, Debug, Serialize, Deserialize)]
struct Setpoint { target_celsius: f32, issued_at: u64 }

#[derive(RecordKey, Clone, Copy, PartialEq, Eq, Debug)]
enum ClimateKey {
    #[key = "reading"]
    #[link_address = "mqtt://climate/zone-a/reading"]
    Reading,

    #[key = "setpoint"]
    #[link_address = "mqtt://climate/zone-a/setpoint"]
    Setpoint,
}

// Outbound: stream readings as they happen.
builder.configure::<Temperature>(ClimateKey::Reading, |reg| {
    reg.buffer(BufferCfg::SpmcRing { capacity: 256 })
        .source(|ctx, producer| async move {
            loop {
                let reading = read_sensor().await;
                producer.send(Temperature { celsius: reading, timestamp: ctx.time().now() })
                    .await.ok();
                ctx.sleep(Duration::from_secs(1)).await;
            }
        })
        .link_to("mqtt://climate/zone-a/reading")
        .with_serializer_raw(Temperature::to_bytes)
        .finish();
});

// Inbound: receive setpoints. Latest instruction wins.
builder.configure::<Setpoint>(ClimateKey::Setpoint, |reg| {
    reg.buffer(BufferCfg::Mailbox)
        .link_from("mqtt://climate/zone-a/setpoint")
        .with_deserializer_raw(Setpoint::from_bytes)
        .tap(apply_setpoint)
        .finish();
});

There is no daemon, no controller process, no message-handling service. There are two records on one node. One has a connector pointing out, one has a connector pointing in. The buffer choice on each side reflects what kind of edge it is, a stream of readings or a queue of one instruction.

The same shape compiles for the controller, just with the directions reversed: it consumes the reading stream from link_from and publishes setpoints with link_to. Two nodes. One contract per record. Zero translation.

Connectors Are User Extensions

Today AimDB ships connectors for MQTT, WebSocket and KNX. Enough to span browser dashboards, telemetry meshes and physical building buses.

Whatever protocol you need next isn't blocked on us shipping it. Connectors are user extensions by design. Two traits handle the publish path:

// 1. The runtime side: how a record gets out.
pub trait Connector: Send + Sync {
    fn publish(
        &self,
        destination: &str,
        config: &ConnectorConfig,
        payload: &[u8],
    ) -> Pin<Box<dyn Future<Output = Result<(), PublishError>> + Send + '_>>;
}

// 2. The builder side: how the runtime constructs your connector
//    and gives you the records that linked to your scheme.
pub trait ConnectorBuilder<R>: Send + Sync {
    fn build<'a>(&'a self, db: &'a AimDb<R>)
        -> Pin<Box<dyn Future<Output = DbResult<Arc<dyn Connector>>> + Send + 'a>>;

    fn scheme(&self) -> &str;
}

publish for outbound, scheme to declare which URL prefix you handle ("mqtt", "knx", your-protocol-here), and build to assemble your connector after the database has collected the records that link to it. Three methods. The hard part of writing a connector is the protocol library you wrap, not AimDB's surface.

If your installed base speaks Modbus, Kafka or anything else, the door is open. A contributor doesn't need to understand the runtime internals to add one. The trait is the contract. Browse the existing connectors →

What Changes Once the Graph Has Edges

When connectors enter the picture, two things stop being someone else's problem.

The first is that the graph doesn't end at the process. A record produced on a microcontroller and consumed in a browser is the same record. The same SchemaType, the same version and the same fields. The connector is the seam, not the boundary. Which means a question that didn't exist while everything ran in one runtime now exists everywhere: when the same record lives on two sides of a connector, which side is right?

The second is that some connectors don't just carry data, they carry consequence. A link_from on a Mailbox buffer isn't observing the world. It's reaching into the world. The pipeline that follows that buffer doesn't have the luxury of being a passive consumer, it has to act. Which raises the second question: when a connector drives an output, what does the rest of the graph owe it? Latency. Ordering. A guarantee that the executed instruction wasn't already obsolete when it arrived.

We won't answer those questions in this post. We'll answer them in the next two.

For now, the move is small but real. A reactive pipeline was the engine of a data-first graph. A connector is what happens when that engine stops being bound to one process. Same contracts. Same buffers. New surface area.

The graph just got an outside.

Want to see a connector mesh in practice? The weather-mesh-demo in the repo wires three sensors, an MQTT broker and a hub together with nothing but contracts and link_to / link_from. Star it, clone it, run it. · Live Demo

Get Involved

AimDB is open source, Apache 2.0 and growing. If you've ever maintained a translation layer between firmware, an edge gateway and a cloud service for what was conceptually one stream of data, you already know why we're building this.

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

---

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