Struct Route

Source

pub struct Route {
    pub input: Endpoint,
    pub output: Endpoint,
    pub options: RouteOptions,
}

Expand description

Defines a single message processing route from an input to an output.

Fields§

§input: Endpoint

The input/source endpoint for the route.

§output: Endpoint

The output/sink endpoint for the route.

§options: RouteOptions

(Optional) Fine-tuning options for the route’s execution.

Implementations§

Source §

impl Route

Source

pub fn new(input: Endpoint, output: Endpoint) -> Self

Creates a new route with default concurrency (1) and batch size (128).

§Arguments

input - The input/source endpoint for the route
output - The output/sink endpoint for the route

Source

pub fn get(name: &str) -> Option<Self>

Retrieves a registered (and running) route by name.

Source

pub fn list() -> Vec<String>

Returns a list of all registered route names.

Source

pub async fn deploy(&self, name: &str) -> Result<()>

Registers the route and starts it. If a route with the same name is already running, it will be stopped first.

§Examples

use mq_bridge::{Route, models::Endpoint};

let route = Route::new(Endpoint::new_memory("in", 10), Endpoint::new_memory("out", 10));
route.deploy("global_route").await.unwrap();
assert!(Route::get("global_route").is_some());

Source

pub async fn stop(name: &str) -> bool

Stops a running route by name and removes it from the registry.

Source

pub async fn create_publisher(&self) -> Result<Publisher>

Creates a new Publisher configured for this route’s output. This is useful if you want to send messages to the same destination as this route.

§Examples

use mq_bridge::{Route, models::Endpoint};

let route = Route::new(Endpoint::new_memory("in", 10), Endpoint::new_memory("out", 10));
let publisher = route.create_publisher().await;
assert!(publisher.is_ok());

Source

pub async fn connect_to_output( &self, name: &str, ) -> Result<Box<dyn MessageConsumer>>

Creates a consumer connected to the route’s output. This is primarily useful for integration tests to verify messages reaching the destination.

Source

pub fn check( &self, name: &str, allowed_endpoints: Option<&[&str]>, ) -> Result<()>

Validates the route configuration, checking if endpoints are supported and correctly configured. Core types like file, memory, and response are always supported.

§Arguments

name - The name of the route
allowed_endpoints - An optional list of allowed endpoint types

Source

pub async fn run(&self, name_str: &str) -> Result<RouteHandle>

Runs the message processing route with concurrency, error handling, and graceful shutdown.

This function spawns the necessary background tasks to process messages. It waits asynchronously until the route is successfully initialized (i.e., connections are established) or until a timeout occurs. The name_str parameter is just used for logging and tracing.

It returns a JoinHandle for the main route task and a Sender channel that can be used to signal a graceful shutdown. The result is typically converted into a RouteHandle for easier management.

§Examples

let route = Route::new(Endpoint::new_memory("in", 10), Endpoint::new_memory("out", 10));

// Start the route (blocks until initialized) and convert to RouteHandle
let handle: RouteHandle = route.run("my_route").await?.into();

// Stop the route later
handle.stop().await;
handle.join().await?;

Source

pub async fn run_until_err( &self, name: &str, shutdown_rx: Option<Receiver<()>>, ready_tx: Option<Sender<()>>, ) -> Result<bool>

The core logic of running the route, designed to be called within a reconnect loop.

Source

pub fn add_handler<T, H, Args>(self, type_name: &str, handler: H) -> Self
where T: DeserializeOwned + Send + Sync + 'static, H: IntoTypedHandler<T, Args>, Args: Send + Sync + 'static,

Registers a typed handler for the route.

The handler can accept either:

fn(T) -> Future<Output = Result<Handled, HandlerError>>
fn(T, MessageContext) -> Future<Output = Result<Handled, HandlerError>>

§Examples


#[derive(Deserialize)]
struct MyData {
    id: u32,
}

async fn my_handler(data: MyData) -> Result<Handled, HandlerError> {
    Ok(Handled::Ack)
}

let route = Route::new(Endpoint::new_memory("in", 10), Endpoint::new_memory("out", 10))
    .add_handler("my_type", my_handler);