emergent-client

Rust SDK for building event-driven workflows on the Emergent engine. Connect to a running engine over Unix IPC and publish or consume messages through three async primitives: Source, Handler, and Sink.

use emergent_client::prelude::*;

#[tokio::main]
async fn main() -> emergent_client::Result<()> {
    let sink = EmergentSink::connect("my_sink").await?;
    let mut stream = sink.subscribe(["timer.tick"]).await?;

    while let Some(msg) = stream.next().await {
        println!("{:?}", msg.payload());
    }
    Ok(())
}

Install

Add the crate to your project:

cargo add emergent-client

Then import the prelude for the most common types:

use emergent_client::prelude::*;

Or import individual types as needed:

use emergent_client::{EmergentSource, EmergentHandler, EmergentSink, EmergentMessage};

Three Primitives

Every Emergent workflow is composed of Sources, Handlers, and Sinks. Each primitive has a single, well-defined role:

Quick Start

Sink -- consume messages

A Sink subscribes to message types and processes each one as it arrives. EmergentSink::messages is a convenience method that connects, subscribes, and yields messages in a single call:

use emergent_client::prelude::*;

let mut stream = EmergentSink::messages("my_sink", ["timer.tick"]).await?;

while let Some(msg) = stream.next().await {
    let data: serde_json::Value = msg.payload_as()?;
    println!("Tick: {data}");
}

For explicit lifecycle control, connect and subscribe separately:

let sink = EmergentSink::connect("my_sink").await?;
let mut stream = sink.subscribe(["timer.tick", "timer.filtered"]).await?;

while let Some(msg) = stream.next().await {
    println!("{} {:?}", msg.message_type(), msg.payload());
}

Source -- publish messages

A Source publishes messages into the engine. It cannot subscribe:

use emergent_client::{EmergentSource, EmergentMessage};
use serde_json::json;

let source = EmergentSource::connect("my_source").await?;
source.publish(
    EmergentMessage::new("sensor.reading")
        .with_payload(json!({"value": 42.5, "unit": "celsius"}))
).await?;

Handler -- subscribe and publish

A Handler subscribes to incoming messages and publishes new ones. Use with_causation_from_message to link output messages to the input that triggered them:

use emergent_client::prelude::*;
use serde_json::json;

let handler = EmergentHandler::connect("order_processor").await?;
let mut stream = handler.subscribe(["order.created"]).await?;

while let Some(msg) = stream.next().await {
    handler.publish(
        EmergentMessage::new("order.processed")
            .with_causation_from_message(msg.id())
            .with_payload(json!({"status": "ok"}))
    ).await?;
}

Publishing Messages

Every primitive that can publish supports two calling styles. Both produce the same result:

use emergent_client::{EmergentMessage, create_message};
use serde_json::json;

// Builder pattern with fluent API
source.publish(
    EmergentMessage::new("timer.tick")
        .with_payload(json!({"count": 1}))
).await?;

// Factory function (matches Python and TypeScript SDKs)
source.publish(
    create_message("timer.tick")
        .with_payload(json!({"count": 1}))
).await?;

Streaming Publish

Publish a collection or async stream of messages. Each message is sent individually so subscribers begin consuming immediately. Both methods return the count of successfully published messages and stop on the first error.

// From a Vec or any IntoIterator
let messages: Vec<EmergentMessage> = records.iter().map(|r| {
    EmergentMessage::new("record.imported").with_payload(json!(r))
}).collect();

let count = source.publish_all(messages).await?;

// From an async stream (e.g., tokio channel)
use tokio_stream::wrappers::ReceiverStream;

let (tx, rx) = tokio::sync::mpsc::channel(32);
// ... spawn producer that sends messages into tx ...
let count = source.publish_stream(ReceiverStream::new(rx)).await?;

Both publish_all and publish_stream are available on EmergentSource and EmergentHandler.

Building Messages

EmergentMessage::new and create_message return a builder with fluent methods for constructing messages:

use emergent_client::{EmergentMessage, create_message};
use serde_json::json;

let msg = create_message("sensor.reading")
    .with_payload(json!({"value": 42.5, "unit": "celsius"}))
    .with_metadata(json!({"sensor_id": "temp-01", "location": "room-a"}));

Link messages into traceable chains with with_causation_from_message and with_correlation_id:

use emergent_client::types::CorrelationId;

let reply = EmergentMessage::new("order.confirmed")
    .with_causation_from_message(original_msg.id())
    .with_correlation_id(CorrelationId::new())
    .with_payload(json!({"confirmed": true}));

The builder sets id (TypeID format msg_<uuidv7>) and timestamp_ms automatically.

Subscribing to Messages

subscribe accepts any type that implements IntoSubscription -- a single &str, an array, a slice, or a Vec:

// Single topic
let stream = sink.subscribe("timer.tick").await?;

// Array of topics
let stream = sink.subscribe(["timer.tick", "timer.filtered"]).await?;

// Slice of topics
let stream = sink.subscribe(&["timer.tick", "timer.filtered"]).await?;

// From a Vec
let topics = vec!["timer.tick".to_string()];
let stream = sink.subscribe(topics).await?;

Iterate over the returned MessageStream with while let:

while let Some(msg) = stream.next().await {
    let data: MyPayload = msg.payload_as()?;
    println!("{data:?}");
}

MessageStream implements futures::Stream, so you can use StreamExt combinators (re-exported in the prelude):

use emergent_client::prelude::*;

stream
    .filter(|msg| futures::future::ready(
        msg.message_type().as_str().starts_with("timer.")
    ))
    .for_each(|msg| async move {
        println!("{:?}", msg);
    })
    .await;

Typed payloads with serde

payload_as deserializes the JSON payload into any type that implements serde::DeserializeOwned:

use serde::Deserialize;

#[derive(Debug, Deserialize)]
struct SensorReading {
    value: f64,
    unit: String,
}

while let Some(msg) = stream.next().await {
    let reading: SensorReading = msg.payload_as()?;
    println!("{} {}", reading.value, reading.unit);
}

Resource Cleanup

Call disconnect() to cleanly close the connection. The SDK sends an unsubscribe-all message so the server sees a normal EOF rather than a connection reset:

let source = EmergentSource::connect("my_source").await?;
// ... use source ...
source.disconnect().await?;

The SDK subscribes to system.shutdown internally. When the Emergent engine signals a graceful shutdown, active message streams close automatically.

Helper Functions

run_source, run_handler, and run_sink eliminate connection, signal handling, and shutdown boilerplate. Each helper connects, sets up SIGTERM handlers, runs your async closure, and disconnects on completion.

Import them from the helpers module:

use emergent_client::helpers::{run_source, run_handler, run_sink};

Source -- custom event loop with shutdown signal

use emergent_client::helpers::run_source;
use emergent_client::EmergentMessage;
use serde_json::json;
use std::time::Duration;

run_source(Some("my_timer"), |source, mut shutdown| async move {
    let mut interval = tokio::time::interval(Duration::from_secs(3));
    let mut count = 0u64;

    loop {
        tokio::select! {
            _ = shutdown.changed() => break,
            _ = interval.tick() => {
                count += 1;
                let msg = EmergentMessage::new("timer.tick")
                    .with_payload(json!({"count": count}));
                source.publish(msg).await.map_err(|e| e.to_string())?;
            }
        }
    }
    Ok(())
}).await?;

Handler -- called once per message

use emergent_client::helpers::run_handler;
use emergent_client::EmergentMessage;
use serde_json::json;

run_handler(
    Some("my_handler"),
    &["timer.tick"],
    |msg, handler| async move {
        let output = EmergentMessage::new("timer.processed")
            .with_causation_from_message(msg.id())
            .with_payload(json!({"processed": true}));
        handler.publish(output).await.map_err(|e| e.to_string())
    }
).await?;

Sink -- called once per message

use emergent_client::helpers::run_sink;

run_sink(
    Some("my_sink"),
    &["timer.tick"],
    |msg| async move {
        println!("Received: {:?}", msg.payload());
        Ok(())
    }
).await?;

The name argument is optional. When set to None, the helper reads from the EMERGENT_NAME environment variable, falling back to a default.

Error Handling

All SDK operations return emergent_client::Result<T>, which uses ClientError as the error type. Match on specific variants for precise control:

use emergent_client::{EmergentSource, ClientError};

match EmergentSource::connect("my_source").await {
    Ok(source) => { /* connected */ }
    Err(ClientError::SocketNotFound(path)) => {
        eprintln!("Engine not running at: {path}");
    }
    Err(ClientError::Timeout) => {
        eprintln!("Connection timed out");
    }
    Err(ClientError::ConnectionFailed(reason)) => {
        eprintln!("Connection failed: {reason}");
    }
    Err(e) => {
        eprintln!("Unexpected error: {e}");
    }
}

Error Variants

Helper functions use a separate HelperError type with variants for connection, subscription, signal setup, and user-function errors.

Message Shape

Every message flowing through Emergent follows the same envelope:

All identifier types (MessageId, CorrelationId, CausationId) use the TypeID format and are available from emergent_client::types.

Use msg.payload_as::<T>() to deserialize the payload into any serde::DeserializeOwned type.

System Events

The Emergent engine broadcasts lifecycle events that your primitives can subscribe to:

Use the typed payload structs for safe access:

use emergent_client::types::{SystemEventPayload, SystemShutdownPayload};

if msg.message_type().as_str().starts_with("system.started.") {
    let event: SystemEventPayload = msg.payload_as()?;
    println!("{} ({}) started", event.name(), event.kind());
}

if msg.message_type().as_str().starts_with("system.error.") {
    let event: SystemEventPayload = msg.payload_as()?;
    if let Some(error) = event.error() {
        eprintln!("{} failed: {error}", event.name());
    }
}

Requirements

• Rust 2024 edition (1.85+)
• Tokio async runtime
• A running Emergent engine with the EMERGENT_SOCKET environment variable set
• Unix platform (Linux or macOS) -- the SDK communicates over Unix domain sockets

License

MIT OR Apache-2.0