Introduction

Remotia is a modular, open-source framework for building and benchmarking remote rendering and video streaming pipelines. Designed with cloud gaming as a primary use case, it provides a small set of abstractions (DTOs, processors, components, and pipelines) that compose into sequential processing chains. The framework links the building blocks and handles their execution so that researchers can focus on experimentation rather than infrastructure. Its platform-agnostic design supports any ffmpeg-compliant codec and protocol, with no hard dependencies on operating systems or proprietary protocols, making it suitable both for academic research and practical application.

It is released as open-source, built upon the Rust programming language. While being designed with remote rendering and cloud gaming in mind, its components are versatile and can be used in a variety of contexts. Bindings and plug-and-play modules for well-established libraries such as ffmpeg and the SRT protocol are available, while ready-to-plug-in components for audio streaming and user input handling are in development; custom modules can already be developed to implement additional features if needed. See the API documentation for the full module reference.

The framework is actively used in the research made by the IPLab group of the University of Catania;

Bibliography

If you use remotia in your work, please cite the following paper:

@article{remotia,
  title={An open source framework for video streaming in cloud gaming},
  author={Catania, Lorenzo and Giudice, Oliver and Battiato, Sebastiano and Stanco, Filippo and Allegra, Dario},
  journal={Multimedia Tools and Applications},
  volume={84},
  pages={41381--41404},
  year={2025},
  doi={10.1007/s11042-025-20798-y},
  publisher={Springer}
}

Quickstart

This example demonstrates a two-pipeline architecture: a main pipeline processes frames by tagging each with a unique frame_id, encoding them with a dummy codec, and checking quality against a PSNR threshold, while an error pipeline catches and logs any frames that fall below the threshold.

Full source: code-samples/examples/quickstart.rs

1. Data Transfer Object (DTO)

The DTO carries a frame_id, a psnr score, an encoded_data buffer, and an optional error field. It implements FrameError<String> so pipeline stages can report errors on it.

#![allow(unused)]
fn main() {
use remotia::{
    traits::{FrameError, FrameProcessor},
    pipeline::{Pipeline, component::Component, registry::PipelineRegistry},
    processors::{
        ticker::Ticker,
        error_switch::OnErrorSwitch,
    },
};
use async_trait::async_trait;
use rand::Rng;

#[derive(Debug, Default)]
struct MyDto {
    frame_id: u64,
    psnr: f64,
    encoded_data: Vec<u8>,
    error: Option<String>,
}

impl FrameError<String> for MyDto {
    fn report_error(&mut self, error: String) { self.error = Some(error); }
    fn get_error(&self) -> Option<String> { self.error.clone() }
}
}

2. Pipeline Processors

Each processor implements FrameProcessor<MyDto> and transforms or inspects the DTO as it moves through the pipeline.

FrameTagger — assigns a unique frame ID on each invocation:

#![allow(unused)]
fn main() {
struct FrameTagger {
    current_frame_id: u64,
}

#[async_trait]
impl FrameProcessor<MyDto> for FrameTagger {
    async fn process(&mut self, mut dto: MyDto) -> Option<MyDto> {
        self.current_frame_id += 1;
        dto.frame_id = self.current_frame_id;
        Some(dto)
    }
}
}

DummyCodec — populates the DTO with random PSNR and encoded data:

#![allow(unused)]
fn main() {
struct DummyCodec;

#[async_trait]
impl FrameProcessor<MyDto> for DummyCodec {
    async fn process(&mut self, mut dto: MyDto) -> Option<MyDto> {
        let mut rng = rand::thread_rng();
        dto.psnr = rng.gen_range(20.0..50.0);
        let len = rng.gen_range(100..1000);
        dto.encoded_data = (0..len).map(|_| rng.r#gen()).collect();
        Some(dto)
    }
}
}

QualityGate — checks whether the PSNR is below a threshold and reports an error if so:

#![allow(unused)]
fn main() {
struct QualityGate {
    threshold: f64,
}

#[async_trait]
impl FrameProcessor<MyDto> for QualityGate {
    async fn process(&mut self, mut dto: MyDto) -> Option<MyDto> {
        if dto.psnr < self.threshold {
            dto.report_error(
                format!("psnr {:.2} below threshold {}", dto.psnr, self.threshold),
            );
        }
        println!(
            "main: frame_id = {}, psnr = {:.2}, data_len = {}, error = {}",
            dto.frame_id,
            dto.psnr,
            dto.encoded_data.len(),
            dto.error.is_some(),
        );
        Some(dto)
    }
}
}

ErrorLogger — logs any error on a frame and discards it:

#![allow(unused)]
fn main() {
struct ErrorLogger;

#[async_trait]
impl FrameProcessor<MyDto> for ErrorLogger {
    async fn process(&mut self, dto: MyDto) -> Option<MyDto> {
        if let Some(ref err) = dto.error {
            println!("error pipeline: frame {} failed — {}", dto.frame_id, err);
        }
        None
    }
}
}

3. Pipeline Assembly

Register the error pipeline first, making it feedable so OnErrorSwitch can push frames into it. Then build the main pipeline with a tick rate-limiter, frame tagger, dummy codec, quality gate, and error switch. Finally register and run everything:

#[tokio::main]
async fn main() {
    let mut registry = PipelineRegistry::<MyDto, &str>::new();

    let error_pipeline = Pipeline::new()
        .link(Component::singleton(ErrorLogger))
        .feedable()
        .tag("error");
    registry.register("error", error_pipeline);

    let error_switch = {
        let error_pipe = registry.get_mut(&"error");
        OnErrorSwitch::new(error_pipe)
    };

    let main_pipeline = Pipeline::new()
        .link(
            Component::new()
                .append(Ticker::new(100))
                .append(FrameTagger { current_frame_id: 0 })
                .append(DummyCodec)
                .append(QualityGate { threshold: 30.0 })
                .append(error_switch)
                .tag("main-step")
        )
        .tag("main");

    registry.register("main", main_pipeline);

    registry.run().await;
}

What happens at runtime

1. The Ticker paces the loop at ~10 Hz.
2. FrameTagger increments frame_id by 1 each tick.
3. DummyCodec fills psnr with a random value between 20.0 and 50.0 and encoded_data with random bytes (100–1000 bytes).
4. QualityGate checks whether psnr is below the threshold (30.0); if so, it reports an error on the DTO. It always passes the frame forward.
5. OnErrorSwitch inspects the DTO's error field and splits execution into two branches:
   • Main branch — no error is present: the frame is consumed within the main pipeline.
   • Error branch — an error is present: the frame is forwarded to the registered error pipeline.
6. In the error pipeline, ErrorLogger prints the failure details and the frame is discarded (returns None).

Next steps

• Read the Key concepts for the full trait and API reference.
• Browse the Processors catalog to find built-in processors for your pipeline.
• See Pipelines & lifecycle for shutdown, feedable pipelines, and the registry.
• Check the Crate map for feature flags and optional crates.

Key concepts

Remotia is built around four abstractions that compose into streaming pipelines. This page defines each one and shows the Rust APIs you implement and use. Browse the API documentation for the full type and trait reference.

Pipeline
 ┌─────────────────────────────┐       ┌─────────────────────────────┐
 │ Component A                 │       │ Component B                 │
 │  ┌───────────┐ ┌───────────┐│  ch   │  ┌───────────┐ ┌───────────┐│
 │  │Processor 1│→│Processor 2││──────▶│  │Processor 3│→│Processor 4││
 │  └───────────┘ └───────────┘│       │  └───────────┘ └───────────┘│
 └─────────────────────────────┘       └─────────────────────────────┘

• A DTO (Data Transfer Object) carries frame data and metadata through the pipeline.
• A processor performs one atomic operation on a DTO.
• A component groups processors into a single async task, connected to neighbors by channels.
• A pipeline links components into a directed chain; multiple pipelines are connected by switches.

Data Transfer Object (DTO)

At the heart of every Remotia pipeline is the Data Transfer Object (DTO) — the envelope that carries frame data and metadata between processing stages. The DTO is intentionally generic: each application defines its own struct with the fields and traits that its processors require. This decouples processors from the data they work on — a processor only knows about the DTO's trait implementations, not its concrete layout — and these abstractions are likely to be optimized at compile time. As a consequence, different modules can work together without direct dependency on one another's data structures.

The DTO is the data structure that flows through the pipeline. It typically holds frame buffers and per-frame statistics. You define your own DTO type and implement the traits that the processors in your pipeline require.

Core trait: FrameProcessor input

Every processor operates on a generic type F. Your DTO is that type F. The framework does not mandate a specific struct — you define one adapted to your use case.

Trait: FrameProperties<K, V>

Used by switches that read or write routing keys on the DTO (e.g. PoolingSwitch, DepoolingSwitch).

#![allow(unused)]
fn main() {
pub trait FrameProperties<K, V> {
    fn set(&mut self, key: K, value: V);
    fn get(&self, key: &K) -> Option<V>;
}
}

Trait: FrameError<E>

Used by OnErrorSwitch to inspect error state on the DTO.

#![allow(unused)]
fn main() {
pub trait FrameError<E> {
    fn report_error(&mut self, error: E);
    fn get_error(&self) -> Option<E>;
}
}

Other available traits

See the traits module documentation for the full trait reference.

Minimal DTO example

#![allow(unused)]
fn main() {
use remotia::traits::{FrameProperties, FrameError};

#[derive(Debug, Default)]
struct MyDto {
    buffer: Vec<u8>,
    frame_id: u64,
    error: Option<String>,
}

impl FrameProperties<String, u64> for MyDto {
    fn set(&mut self, key: String, value: u64) {
        if key == "frame_id" { self.frame_id = value; }
    }
    fn get(&self, key: &String) -> Option<u64> {
        if key == "frame_id" { Some(self.frame_id) } else { None }
    }
}

impl FrameError<String> for MyDto {
    fn report_error(&mut self, error: String) { self.error = Some(error); }
    fn get_error(&self) -> Option<String> { self.error.clone() }
}
}

Custom processor modules may define additional traits. Your DTO must implement them if you use those modules in your pipeline.

Processors

Processors are the smallest unit of work in a Remotia pipeline. Each processor performs one atomic operation on the DTO — encoding a frame, adding a timestamp, routing to another pipeline, or any other self-contained transformation. Because every processor implements the same simple interface, steps can be added, removed, or reordered with minimal code changes, making it straightforward to ablate different parts of the computation and analyze their impact on the overall system.

A processor is a single unit of work applied to a DTO. The core trait is:

#![allow(unused)]
fn main() {
#[async_trait]
pub trait FrameProcessor<F> {
    async fn process(&mut self, frame_data: F) -> Option<F>;
}
}

See the processors module documentation for the full list of built-in processor types and traits.

Return contract:

• Some(dto) — the DTO is passed to the next processor in the component.
• None — the DTO is consumed. The pipeline interprets this as "this frame is done here" — it may have been redirected to another pipeline (by a switch), stored, or dropped.

Processors have full ownership of the DTO while processing it. This avoids borrowing conflicts and makes the data flow explicit.

See the Processors page for the catalog of built-in processor types.

Components

A component is an asynchronous execution context that runs a sequence of processors as a single Tokio task. By grouping related processors into a component, you control which work shares a task (sequential execution within the component) and which runs concurrently with other components. Each component receives DTOs from an input channel, passes them through its processors, and sends the result to an output channel. This design enables fine-grained concurrency: you can tune parallelism by splitting or merging processor sequences across components without changing the processors themselves.

A component groups an ordered sequence of processors into a single async task (a Tokio task). Each component:

1. Receives a DTO from an input channel (or allocates one itself).
2. Passes it through its processors sequentially.
3. Sends the resulting DTO (if any) to the next component via an output channel.

Components are the unit of concurrency. By grouping processors into different components, you control which work shares a task and which runs in parallel. The framework uses unbounded mpsc channels to connect adjacent components within a pipeline.

Builder API

#![allow(unused)]
fn main() {
Component::new()
    .append(processor_a)
    .append(processor_b)
    .tag("encoder")
}

Or, for a single-processor component:

#![allow(unused)]
fn main() {
Component::singleton(processor)
}

See the pipeline module documentation for the full component API reference.

Pipelines

A pipeline is a directed chain of components connected by message channels. Pipelines define the top-level structure of a Remotia application: they wire components together, spawn each as a separate async task, and manage the flow of DTOs from the first component to the last. While a simple application may use a single pipeline, complex systems often combine multiple pipelines — for example, a main streaming pipeline, a separate error-handling pipeline, and a profiling pipeline — connected by switch processors that route frames between them.

A pipeline is a chain of components connected by channels. Components within a pipeline run concurrently (each is a separate Tokio task), while processors within a component run sequentially.

Builder API

#![allow(unused)]
fn main() {
Pipeline::new()
    .link(component_a)
    .link(component_b)
    .tag("main")
    .run()
}

Or, for a single-component pipeline:

#![allow(unused)]
fn main() {
Pipeline::singleton(component)
}

Calling .run() automatically creates the channels between adjacent components and spawns each component as a Tokio task. It returns Vec<JoinHandle<()>>.

Feedable pipelines

Mark a pipeline as .feedable() to allow external code to inject DTOs into its head:

#![allow(unused)]
fn main() {
let mut pipeline = Pipeline::new().link(component).feedable();
let feeder = pipeline.get_feeder();
feeder.feed(my_dto);
}

Multi-pipeline architectures

Complex systems use multiple pipelines connected by switches. For example, a main streaming pipeline and a separate error-handling pipeline, linked by an OnErrorSwitch. See the Pipelines & Lifecycle page for the full API and lifecycle details. See the pipeline module documentation for the full Pipeline API reference.

Switches

Switches are a special category of processor that redirect DTOs from one pipeline to another. Instead of returning Some(dto) to continue in the current pipeline, a switch sends the DTO to a different pipeline and returns None, signalling that the frame's processing continues elsewhere. Switches are the mechanism for building multi-pipeline architectures — they enable error handling and profiling with limited impact on streaming performance, auxiliary pipelines for logging or debugging, and scaling to multi-user streaming with multiple frame data sources and sinks. The framework provides concrete switch implementations for these patterns, including load balancing via PoolingSwitch/DepoolingSwitch and cloning side-channels via CloneSwitch.

Switches are processors that move DTOs between pipelines. Instead of returning Some(dto) to continue in the current pipeline, they send the DTO to a different pipeline and return None.

The framework provides several switch types:

See the Processors page for constructor signatures and usage details. See the processors module documentation for the full switch type reference.

Processors

Processors are the building blocks of a Remotia pipeline. Each processor performs one atomic operation on the frame data by implementing the FrameProcessor<F> trait with a single process method. The method takes ownership of a DTO and returns Option<F>: Some(dto) passes the data forward to the next processor, while None signals that processing for that frame is complete — it may have been redirected to another pipeline, stored, or dropped. This simple contract makes it easy to add, remove, or reorder processing steps without affecting the rest of the system. See the processors module documentation for the full API reference.

Processors implement the FrameProcessor<F> trait and perform a single atomic operation on the DTO. This page catalogs every built-in processor in remotia-core.

#![allow(unused)]
fn main() {
#[async_trait]
pub trait FrameProcessor<F> {
    async fn process(&mut self, frame_data: F) -> Option<F>;
}
}

Routing switches

Switches move DTOs between pipelines. They return None (or Some with the original) so the current pipeline can continue or terminate for that frame.

Switch

Redirects the DTO into a different pipeline. Returns None — the current pipeline stops processing this frame.

#![allow(unused)]
fn main() {
use remotia::processors::switch::Switch;

let switch = Switch::new(&mut destination_pipeline);
}

Use when: branching — e.g. sending frames down an error or debug pipeline instead of the main one.

CloneSwitch

Clones the DTO, sends the clone to another pipeline, and passes the original forward. The DTO type must implement Clone.

#![allow(unused)]
fn main() {
use remotia::processors::clone_switch::CloneSwitch;

let clone_switch = CloneSwitch::new(&mut profiling_pipeline);
}

Use when: parallel side-channels — e.g. a profiling pipeline that receives every frame while the main pipeline continues uninterrupted.

OnErrorSwitch

Redirects the DTO to a destination pipeline if it carries a matching error. If no error is present (or the error does not match), the frame passes through with Some(frame). The DTO must implement FrameError<E>.

#![allow(unused)]
fn main() {
use remotia::processors::error_switch::OnErrorSwitch;

let error_switch = OnErrorSwitch::new::<MyDto, MyError>(&mut error_pipeline)
    .detect(MyError::Timeout)
    .detect(MyError::ConnectionError);
}

Use when: conditional error routing — only certain error variants are diverted.

PoolingSwitch

Picks a random destination from a registered pool, stamps the DTO with the chosen pool key (via FrameProperties), and sends it. Returns None. The DTO must implement FrameProperties<P, K>.

#![allow(unused)]
fn main() {
use remotia::processors::pool_switch::PoolingSwitch;

let pool_switch = PoolingSwitch::<MyDto, &str, usize>::new("worker_id")
    .entry(0, &mut worker_pipeline_0)
    .entry(1, &mut worker_pipeline_1)
    .entry(2, &mut worker_pipeline_2);
}

Use when: fan-out to a pool of workers — e.g. distributing encoding across N encoder pipelines.

DepoolingSwitch

Routes the DTO to the destination matching its pool key (read via FrameProperties). Returns None. The DTO must implement FrameProperties<P, K>.

#![allow(unused)]
fn main() {
use remotia::processors::pool_switch::DepoolingSwitch;

let depool_switch = DepoolingSwitch::<MyDto, &str, usize>::new("worker_id")
    .entry(0, &mut merger_pipeline_0)
    .entry(1, &mut merger_pipeline_1);
}

Use when: fan-in — merging results from a worker pool back into per-worker downstream pipelines.

Timing

Ticker

Waits for the configured interval, then passes the DTO forward unchanged.

#![allow(unused)]
fn main() {
use remotia::processors::ticker::Ticker;

let ticker = Ticker::new(16); // 16 ms ≈ 60 FPS
}

Use when: frame-rate pacing at the head of a capture pipeline, or throttling any component.

Inline processors

These wrap bare functions, closures, or async functions as FrameProcessor implementors. Use them for quick logic without defining a dedicated struct.

Function

Wraps a function pointer.

#![allow(unused)]
fn main() {
use remotia::processors::functional::Function;

let proc = Function::new(|mut dto: MyDto| -> Option<MyDto> {
    dto.frame_id += 1;
    Some(dto)
});
}

Closure

Wraps a capturing closure.

#![allow(unused)]
fn main() {
use remotia::processors::functional::Closure;

let counter = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0));
let counter_clone = counter.clone();

let proc = Closure::new(move |dto: MyDto| -> Option<MyDto> {
    counter_clone.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    Some(dto)
});
}

Convenience method on Component:

#![allow(unused)]
fn main() {
component.closure(|dto: MyDto| Some(dto))
}

AsyncFunction

Wraps an async function pointer. Use the async_func! macro to create the pinned future.

#![allow(unused)]
fn main() {
use remotia::processors::async_functional::AsyncFunction;
use remotia::async_func;

async fn fetch_frame(dto: MyDto) -> Option<MyDto> {
    // async I/O ...
    Some(dto)
}

let proc = AsyncFunction::new(|dto| async_func!(async move {
    fetch_frame(dto).await
}));
}

Containers

Sequential

Runs a sequence of processors in order inside a single component. Each processor receives the output of the previous one. If any processor returns None, the sequence stops.

#![allow(unused)]
fn main() {
use remotia::processors::containers::sequential::Sequential;

let seq = Sequential::new()
    .append(processor_a)
    .append(processor_b)
    .append(processor_c);
}

Use when: grouping processors that must run in the same async task — e.g. a tick-then-capture pattern, or a sequence of buffer operations that should not be split across component boundaries.

Profilation

The profilation utilities (behind the profilation feature flag, in the separate remotia-profilation-utils crate) collect execution statistics during frame processing. Remotia does not force a specific method to collect statistics, but provides profilation processors that are injected into components at the points where the phase to profile begins or ends, so that collection points are equal between executions and reproducibility is strengthened.

Timestamp pairs

A TimestampAdder("encoding_start") processor at the start of the phase saves the current time into a DTO field. A time-diff processor at the end writes the elapsed time into another DTO field:

#![allow(unused)]
fn main() {
use remotia::profilation::{TimestampAdder, TimestampDiffCalculator};

Pipeline::new()
    .link(
        Component::new()
            .append(TimestampAdder("encoding_start"))
            .append(AV1Preprocessor())
            .append(AV1Compressor())
            .append(AV1Serializer())
            .append(TimestampDiffCalculator("encoding_start", "encoding_time"))
    )
    .run();
}

The same measurement works when the phase is split across multiple components, making it simple to verify whether asynchronous preprocessing improves performance:

#![allow(unused)]
fn main() {
Pipeline::new()
    .link(
        Component::new()
            .append(TimestampAdder("encoding_start"))
            .append(AV1Preprocessor())
    )
    .link(
        Component::new()
            .append(AV1Compressor())
            .append(AV1Serializer())
            .append(TimestampDiffCalculator("encoding_start", "encoding_time"))
    )
    .run();
}

ProfiledSequential

When only the overall processing time matters and the computation runs in a single thread, ProfiledSequential wraps a sequence of processors and measures their total execution time:

#![allow(unused)]
fn main() {
use remotia::profilation::ProfiledSequential;

Component::new()
    .append(ProfiledSequential("encoding_time"))
    .append(AV1Preprocessor())
    .append(AV1Compressor())
    .append(AV1Serializer());
}

It can later be replaced by a plain Sequential (which runs the same sequence without measuring) with minimal edits.

Frame droppers and loggers

The profilation feature also includes utilities for dropping stale frames and logging statistics:

The delay threshold used by frame droppers is grounded in the paper's experimental setup, where a threshold on delay is added before rendering on the client screen to drop stale frames.

Browse the API documentation for the full profilation module reference.

Pipelines & lifecycle

A pipeline is a directed chain of components that process frames sequentially. Each component runs concurrently as its own async task, while processors within a component execute in order. Pipelines can be connected via switches — special processors that route DTOs from one pipeline to another — enabling complex architectures such as a main streaming pipeline with a separate error-handling pipeline and a profiling side-channel. The framework manages channel wiring and task spawning automatically: calling .run() creates unbounded mpsc channels between adjacent components, spawns each as a Tokio task, and returns join handles for lifecycle management. See the pipeline module documentation for the full API reference.

This page covers the full Pipeline, PipelineFeeder, PipelineRegistry, and PipelineHandle APIs, plus the shutdown/drain lifecycle.

Pipeline

A Pipeline<F> is an ordered chain of Component<F> instances. When you call .run(), the framework:

1. Binds — creates unbounded mpsc channels between each pair of adjacent components.
2. Spawns — launches each component as a separate Tokio task.
3. Returns Vec<JoinHandle<()>> you can await.

Builder API

#![allow(unused)]
fn main() {
use remotia::pipeline::Pipeline;

let handles = Pipeline::new()
    .link(capture_component)
    .link(encoder_component)
    .link(transmission_component)
    .tag("main")
    .run();
}

PipelineFeeder

A PipelineFeeder<F> lets you inject DTOs into the head of a feedable pipeline from external code.

#![allow(unused)]
fn main() {
let mut pipeline = Pipeline::new()
    .link(component)
    .feedable();

let feeder = pipeline.get_feeder();

// Later, from any task:
feeder.feed(my_dto); // panics if the channel is closed
}

PipelineFeeder is Clone-safe — you can share it across tasks.

PipelineHandle

A PipelineHandle lets you request graceful shutdown of a pipeline.

#![allow(unused)]
fn main() {
let mut pipeline = Pipeline::new().link(component).tag("main");
let handle = pipeline.get_handle();
let handles = pipeline.run();

// From another task:
handle.request_shutdown();
}

When request_shutdown() is called, the shared AtomicBool is set. Every component in the pipeline checks this signal each iteration and exits when it becomes true.

PipelineHandle is Clone — multiple callers can hold a handle to the same pipeline. See the pipeline module documentation for the full PipelineHandle API reference.

PipelineRegistry

PipelineRegistry<F, K> manages multiple named pipelines and runs them together. This is the standard way to launch a multi-pipeline architecture.

Builder API

#![allow(unused)]
fn main() {
use remotia::pipeline::registry::PipelineRegistry;

let mut registry = PipelineRegistry::<MyDto, &str>::new();

registry.register("main", main_pipeline);
registry.register("errors", error_pipeline);

registry.run().await; // runs all pipelines, blocks until all finish
}

See the pipeline registry module documentation for the full PipelineRegistry API reference.

Connecting pipelines with switches

When using a registry, you typically get feeders from destination pipelines before calling .run():

#![allow(unused)]
fn main() {
let mut registry = PipelineRegistry::<MyDto, &str>::new();

let error_pipeline = Pipeline::new()
    .link(error_component)
    .tag("errors");

registry.register("errors", error_pipeline);

let error_switch = {
    let error_pipe = registry.get_mut(&"errors");
    OnErrorSwitch::new(error_pipe).detect(MyError::Timeout)
};

let main_pipeline = Pipeline::new()
    .link(capture_component.append(ticker).append(error_switch))
    .tag("main");

registry.register("main", main_pipeline);
registry.run().await;
}

Lifecycle: shutdown and drain

Understanding the component lifecycle is important for graceful shutdown.

Normal operation

Component receives DTO from input channel
  → passes through processors
  → sends result to output channel
  → repeats

Shutdown signal

When PipelineHandle::request_shutdown() is called, the shared AtomicBool is set. On the next iteration, each component observes the signal and breaks out of its loop.

Drain mode

If a component's input channel is closed (e.g. the upstream component has exited), the component enters drain mode:

1. It stops reading from the channel (it will always be None).
2. It continues the loop, yielding F::default() as the DTO, and running processors.
3. When a processor returns None (frame consumed), the component checks if it should exit.
4. If the input channel was closed and the last processor returned None, the component exits.

This ensures that in-flight DTOs in downstream components can finish processing before shutdown.

Component without an input channel

A component with no receiver (e.g. the head of a non-feedable pipeline that generates its own DTOs) relies solely on the shutdown signal. It checks is_shutdown() each iteration and exits when the signal fires.

Channel binding details

When .run() is called on a pipeline that has not been bound, the framework automatically creates an unbounded mpsc channel between each pair of adjacent components:

Component[i]  ──sender──▶  channel  ──receiver──▶  Component[i+1]

The last component has no sender; the first component has no receiver unless the pipeline is feedable (in which case the feeder's sender is wired to the first component's receiver).

Buffer pools

Buffer pools manage memory reuse across processing steps. Memory buffers are stored in pools from where they are extracted and returned once used, avoiding redundant allocation and consistently improving performance.

Borrower/redeemer flow

A borrower processor pulls a memory buffer from the selected pool, if any is available, and inserts it into the current DTO. Subsequent processors fill the buffer with data. A redeemer processor sends the buffer back to the pool once it is no longer needed by the current processing step. The borrower and redeemer may belong to different components — a buffer can travel across components before being returned to the pool.

The policy when the pool is empty, for instance locking until a buffer is available or dropping the frame, must be decided based on the use case.

Implementation

Pools are backed by asynchronous message queues, bounded by allocating a fixed amount of buffers at run-time. The number of buffers allocated in each pool tunes how much data each processing step can handle in parallel.

#![allow(unused)]
fn main() {
use remotia::buffers::BuffersPool;

let encoding_pool = BuffersPool::new("encoded_frame", 4);

Pipeline::new()
    .link(
        Component::new()
            .append(encoding_pool.borrower())
            .append(AV1Encoder())
    )
    .link(
        Component::new()
            .append(TCPSender())
            .append(encoding_pool.redeemer())
    )
    .run();
}

The borrower pulls a buffer from the pool (if available) and injects it into the DTO. The redeemer returns the buffer from the DTO to the pool so it can be reused for a new frame, making the encoding process more efficient by avoiding repeated memory allocations.

All buffers should be redeemed from the DTO before it is deallocated at the end of a terminal pipeline, or the pool will empty causing a deadlock. The framework provides debugging facilities to locate such problems.

Feature flag

Buffer pool types are behind the buffers feature flag on the remotia crate. See the crate map for the full type list (BuffersPool, BufferAllocator, BufferBorrower, BufferRedeemer, PoolRegistry, #[buffers_map]).

Browse the API documentation for the full module reference.

Crate map

Remotia is organized as a workspace of crates. The remotia umbrella crate re-exports them behind feature flags so you only compile what you need.

Core crate

Optional crates (feature flags on the remotia crate)

External crates (separate Cargo dependencies)

These live in separate repositories and are not part of the remotia umbrella crate. Add them as direct dependencies.

Feature flag usage

[dependencies.remotia]
version = "0.1.0"
default-features = false
features = ["capture", "transmission"]  # only what you need

All features are off by default. Enable only the ones you use to minimize compile time and dependencies.

Examples

This chapter contains worked examples that demonstrate the framework's mechanisms and design patterns.

Each example includes a step-by-step walkthrough with inline code. Full source code is available on GitHub.

Screen snapper — Capture the screen and save to disk

Periodically captures the primary display and saves each frame as a PNG file.

Full source: GitHub — screen-snapper

Architecture

Two components in a single pipeline:

Component 1: Capture                          Component 2: Save
┌──────────────────────────────────┐          ┌─────────────────┐
│ Ticker → BufferAllocator →       │  channel │ PNGBufferSaver  │
│ XCapCapturer                     │─────────▶│                 │
└──────────────────────────────────┘          └─────────────────┘

• Component 1 paces the capture at 1 FPS, allocates a buffer, fills it with screen pixels.
• Component 2 reads the buffer and writes a PNG.

The DTO

The DTO holds a single optional BytesMut buffer identified by an enum key. It implements PullableFrameProperties so processors can pull and push the buffer by key.

#![allow(unused)]
fn main() {
#[derive(Default, Debug)]
pub struct RecorderData {
    screen_buffer: Option<BytesMut>,
}

#[derive(Clone, Copy)]
pub enum Buffers {
    CapturedScreenBuffer,
}

impl PullableFrameProperties<Buffers, BytesMut> for RecorderData {
    fn push(&mut self, key: Buffers, value: BytesMut) {
        match key {
            Buffers::CapturedScreenBuffer => self.screen_buffer.replace(value),
        };
    }
    fn pull(&mut self, key: &Buffers) -> Option<BytesMut> {
        match key {
            Buffers::CapturedScreenBuffer => self.screen_buffer.take(),
        }
    }
}
}

The pull/push pattern ensures the buffer is taken out of the DTO during processing and returned after, avoiding mutable aliasing.

Pipeline construction

#![allow(unused)]
fn main() {
fn capturer(monitor_id: usize, height: u32, width: u32) -> Component<RecorderData> {
    Component::new()
        .append(Ticker::new(1000))       // 1 tick/second
        .append(BufferAllocator::new(    // allocate a fresh BytesMut
            Buffers::CapturedScreenBuffer,
            height as usize * width as usize * 3,
        ))
        .append(
            XCapCapturer::builder()
                .buffer_key(Buffers::CapturedScreenBuffer)
                .monitor_id(monitor_id)
                .build(),
        )
}

fn saver(height: u32, width: u32) -> Component<RecorderData> {
    Component::new().append(
        PNGBufferSaver::builder()
            .buffer_key(Buffers::CapturedScreenBuffer)
            .path("./screenshots/")
            .height(height)
            .width(width)
            .build(),
    )
}

let pipeline = Pipeline::<RecorderData>::new()
    .link(capturer(monitor_id, height, width))
    .link(saver(height, width));

for handle in pipeline.run() {
    handle.await.unwrap();
}
}

Key takeaways

• Ticker at the head of a component paces frame generation.
• BufferAllocator is a processor that allocates a new BytesMut into the DTO each tick — useful when the buffer size is known ahead of time.
• Custom processors (XCapCapturer, PNGBufferSaver) implement FrameProcessor and use PullableFrameProperties to exchange buffers with the DTO.

Platform-dependant screen snapper — Handling platform-specific backends

Same capture-and-save pipeline as the screen snapper, but uses Cargo feature flags to select between X11/XCap and Wayland/libwayshot capture backends at compile time.

Full source: GitHub — platform-dependant-screen-snapper

Architecture

Identical runtime layout — two components in one pipeline. The difference is in how the capturer is selected:

Component 1: Capture                          Component 2: Save
┌──────────────────────────────────┐          ┌─────────────────┐
│ Ticker → BufferAllocator →       │  channel │ PNGBufferSaver  │
│ [xcap OR wayshot] Capturer       │─────────▶│                 │
└──────────────────────────────────┘          └─────────────────┘

Feature flags in Cargo.toml

[features]
default = ["wayshot"]
xcap = ["dep:xcap"]
wayshot = ["dep:libwayshot"]

The lib.rs uses compile-time guards to prevent misconfiguration:

#![allow(unused)]
fn main() {
#[cfg(not(any(feature = "xcap", feature = "wayshot")))]
compile_error!("No snapper backend enabled");

#[cfg(all(feature = "xcap", feature = "wayshot"))]
compile_error!("Compiling with both wayshot and xcap is not currently supported.");

#[cfg(feature = "wayshot")]
pub mod wayshot_capturer;

#[cfg(feature = "xcap")]
pub mod xcap_capturer;
}

Runtime selection with conditional compilation

A capture.rs module provides the same interface regardless of backend:

#![allow(unused)]
fn main() {
#[cfg(feature = "xcap")]
pub mod xcap {
    pub fn fetch_screen_resolution() -> (u32, u32) {
        xcap_utils::display_size(MONITOR_ID)
    }
    pub fn capturer_processor() -> XCapCapturer<Buffers> {
        XCapCapturer::builder()
            .buffer_key(Buffers::CapturedScreenBuffer)
            .monitor_id(MONITOR_ID)
            .build()
    }
}
#[cfg(feature = "xcap")]
pub use xcap::*;

#[cfg(feature = "wayshot")]
pub mod libwayshot {
    pub fn fetch_screen_resolution() -> (u32, u32) {
        wayshot_utils::display_size()
    }
    pub fn capturer_processor() -> WayshotCapturer<Buffers> {
        WayshotCapturer::builder()
            .buffer_key(Buffers::CapturedScreenBuffer)
            .build()
    }
}
#[cfg(feature = "wayshot")]
pub use libwayshot::*;
}

The main.rs calls fetch_screen_resolution() and capturer_processor() without knowing which backend is active.

Building

# Default (wayshot)
cargo run --example autosnapper

# X11/XCap backend
cargo run --example autosnapper --no-default-features --features xcap

Key takeaways

• Cargo feature flags are the idiomatic Rust way to handle platform-specific dependencies. Remotia's FrameProcessor trait makes it easy — both backends implement the same trait, so the pipeline code is backend-agnostic.
• compile_error! guards prevent invalid feature combinations at compile time.
• The DTO and save component are identical across backends — only the capture processor changes.

Multi-pipeline with switches — Connecting pipelines

This example demonstrates one of remotia's key mechanisms: multiple pipelines connected by switch processors, which route frames between them. A main processing pipeline routes frames to different downstream pipelines based on conditions.

Full source: GitHub — examples

Architecture

Three pipelines connected by switches:

Pipeline "main"
┌──────────────────────────────────────────┐
│ Component:                               │
│  Ticker → Closure(print) → CloneSwitch ──┼──▶ Pipeline "profiling"
│                           → Switch ──────┼──▶ Pipeline "errors" (if error)
└──────────────────────────────────────────┘

• Main pipeline: generates frames, prints their state, clones each frame to the profiling pipeline, and conditionally switches errored frames to the error pipeline.
• Profiling pipeline: receives a clone of every frame and logs statistics.
• Error pipeline: receives only frames that carry an error.

The DTO

#![allow(unused)]
fn main() {
#[derive(Debug, Default, Clone)]
struct FrameDto {
    frame_id: u64,
    timestamp: Option<u128>,
    error: Option<String>,
}

impl FrameProperties<String, u64> for FrameDto {
    fn set(&mut self, key: String, value: u64) {
        if key == "frame_id" { self.frame_id = value; }
    }
    fn get(&self, key: &String) -> Option<u64> {
        if key == "frame_id" { Some(self.frame_id) } else { None }
    }
}

impl FrameError<String> for FrameDto {
    fn report_error(&mut self, error: String) { self.error = Some(error); }
    fn get_error(&self) -> Option<String> { self.error.clone() }
}
}

Note that Clone is required because CloneSwitch clones the DTO.

Building the pipelines

#![allow(unused)]
fn main() {
let mut registry = PipelineRegistry::<FrameDto, &str>::new();

// 1. Register profiling pipeline (receives a clone of every frame)
let profiling_pipeline = Pipeline::new()
    .link(Component::singleton(Closure::new(|dto: FrameDto| {
        log::info!("[profiling] frame_id={}", dto.frame_id);
        None
    })))
    .tag("profiling");
registry.register("profiling", profiling_pipeline);

// 2. Register error pipeline (receives only errored frames)
let error_pipeline = Pipeline::new()
    .link(Component::singleton(Closure::new(|dto: FrameDto| {
        log::warn!("[errors] frame_id={} error={:?}", dto.frame_id, dto.error);
        None
    })))
    .tag("errors");
registry.register("errors", error_pipeline);

// 3. Build switches that target the above pipelines
let clone_switch = CloneSwitch::new(registry.get_mut(&"profiling"));
let error_switch = OnErrorSwitch::new(registry.get_mut(&"errors"));

// 4. Build main pipeline
let main_pipeline = Pipeline::new()
    .link(
        Component::new()
            .append(Ticker::new(33))       // ~30 FPS
            .append(Closure::new(|mut dto: FrameDto| {
                dto.frame_id += 1;
                log::info!("[main] frame_id={}", dto.frame_id);
                Some(dto)
            }))
            .append(error_switch)           // redirect errored frames
            .append(clone_switch)           // clone to profiling pipeline
    )
    .tag("main");
registry.register("main", main_pipeline);

registry.run().await;
}

How switches compose

The processor order matters:

1. error_switch checks if the frame has an error. If yes, it sends the frame to the error pipeline and returns None — the frame never reaches clone_switch. If no error, it returns Some(frame) and processing continues.
2. clone_switch clones the frame, sends the clone to profiling, and returns Some(original). The original then exits the component and flows to the next component in the main pipeline (if any).

By placing error_switch before clone_switch, errored frames are not cloned to profiling. Reverse the order if you want profiling to see every frame including errored ones.

Key takeaways

• CloneSwitch is non-destructive: it forwards the original and sends a copy elsewhere. Use it for side-channels like profiling or logging.
• OnErrorSwitch is conditional: it only redirects when the DTO carries a matching error. Otherwise the frame passes through.
• Switch is unconditional: it always redirects and returns None. Use it when the main pipeline should not continue for that frame.
• Processor ordering within a component determines which switches fire and in what sequence. Plan the order carefully when combining multiple switches.