Struct EmergentHandler

Source

pub struct EmergentHandler { /* private fields */ }

Expand description

A Handler primitive that subscribes to and publishes messages.

Handlers are the transformation layer in a workflow. They receive messages from Sources or other Handlers, process them, and emit new messages.

§Example

use emergent_client::{EmergentHandler, EmergentMessage};
use futures::StreamExt;

let handler = EmergentHandler::connect("my_filter").await?;
let mut stream = handler.subscribe("timer.tick").await?;

while let Some(msg) = stream.next().await {
    let output = EmergentMessage::new("timer.filtered")
        .with_causation_id(msg.id());
    handler.publish(output).await?;
}

Implementations§

Source §

impl EmergentHandler

Source

pub async fn connect(name: &str) -> Result<Self>

Connect to the Emergent engine as a Handler.

§Errors

Returns an error if the connection fails.

Source

pub async fn connect_to(name: &str, socket_path: &Path) -> Result<Self>

Connect to the Emergent engine at a specific socket path.

This is useful for testing or when connecting to a non-default socket.

§Errors

Returns an error if the connection fails.

Subscribe to message types and return a stream of incoming messages.

The SDK automatically handles system.shutdown messages - when the engine signals shutdown for handlers, the stream will close gracefully.

§Examples

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

// Multiple topics with array
let stream = handler.subscribe(["timer.tick", "timer.filtered"]).await?;

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

§Errors

Returns an error if the subscription fails.

Source

pub async fn messages( name: impl Into<String>, _types: impl IntoSubscription, ) -> Result<(Self, MessageStream)>

Convenience method that connects, gets configured subscriptions, and returns (handler, stream) for the common one-liner pattern.

§Errors

Returns an error if connection or subscription fails.

Source

pub async fn publish(&self, message: EmergentMessage) -> Result<()>

Publish a message to the engine (fire-and-forget).

§Errors

Returns an error if the message cannot be sent.

Source

pub async fn publish_ack(&self, message: EmergentMessage) -> Result<()>

Publish a message with broker acknowledgment (backpressure).

Unlike [publish], this waits for the engine’s message broker to confirm it has processed and forwarded the message before returning. This provides natural backpressure — the caller cannot outpace the broker.

Used internally by [publish_all] and [publish_stream].

§Errors

Returns an error if the broker rejects the message or times out.

Source

pub async fn publish_all( &self, messages: impl IntoIterator<Item = EmergentMessage>, ) -> Result<usize>

Publish all messages from an iterator.

Sends each message individually with broker acknowledgment so subscribers begin consuming immediately. Stops on the first error.

Returns the number of messages successfully published.

§Errors

Returns the first publish error encountered.

Source

pub async fn publish_stream<S>(&self, stream: S) -> Result<usize>
where S: Stream<Item = EmergentMessage> + Unpin,

Publish messages from an async stream.

Consumes the stream, publishing each message individually with broker acknowledgment so subscribers begin consuming immediately. Stops on the first publish error or when the stream ends.

Returns the number of messages successfully published.