Crate rusteron_archive

Source
Expand description

§rusteron-archive

rusteron-archive is a module within the rusteron project that provides functionalities for interacting with Aeron’s archive capabilities in a Rust environment. This module extends rusteron-client by offering features for recording streams, managing archives, and handling replay capabilities.

§Overview

The rusteron-archive module is intended to help Rust developers leverage Aeron’s archive functionalities, including the recording and replaying of messages. However, this module is currently in an early stage and has not been thoroughly tested.

Important: The module is in an early stage and might undergo changes. Ensure that you test thoroughly when incorporating it in production-grade systems.

§Installation

Add rusteron-archive to your Cargo.toml:

dynamic lib

[dependencies]
rusteron-archive = "0.1"

static lib

[dependencies]
rusteron-archive = { version = "0.1", features= ["static"] }

You must also ensure that you include Aeron C libraries required by rusteron-archive when using default feature. (Using static will automatically include these dependancies in binary).

§Features

  • Stream Recording: Enables recording of Aeron streams.
  • Replay Handling: Provides capabilities for replaying recorded messages.
  • Publication: Send messages to various Aeron channels.
  • Subscription: Receive messages from Aeron channels.
  • Callbacks: Handle events such as new publications, new subscriptions, and errors.
  • Automatic Resource Management (new method only): The wrappers attempt to automatically manage resources, specifically when using the new method. This includes calling the appropriate xxx_init method during initialisation and automatically invoking xxx_close or xxx_destroy methods (if one exists) during cleanup. However, this management is partial. For other methods, such as AeronArchive::set_aeron, it is the developer’s responsibility to ensure that the arguments remain valid and alive during their use. Proper resource management beyond initialisation requires manual handling by the user to avoid undefined behaviour or resource leaks.

§General Patterns

Much like rusteron-client, the rusteron-archive module follows several general patterns to simplify usage of Aeron functionalities in Rust:

  • Cloneable Wrappers: All Rust wrappers in rusteron-archive can be cloned, and they will refer to the same underlying Aeron C instance/resource. This allows safe use of multiple references to the same object. If you need a shallow copy, use clone_struct(), which copies only the underlying C struct.

  • Mutable and Immutable Operations: Modifications can be performed directly with &self, allowing flexibility without needing additional ownership complexities.

  • Automatic Resource Management (new method only): The wrappers attempt to automatically manage resources, clearing objects and calling the appropriate close, destroy, or remove methods when needed.

  • Manual Handler Management: Callbacks and handlers require manual management. Handlers are passed into the C bindings using Handlers::leak(xxx), and need to be explicitly released by calling release(). This manual process is required due to the complexity of determining when these handlers should be cleaned up once handed off to C. For methods where the callback is not stored and only used there and then e.g. poll, you can pass in a closure directory e.g.

ⓘ
  subscription.poll_once(|msg, header| { println!("msg={:?}, header={:?}", msg, header) })

§Handlers and Callbacks

Handlers within rusteron-archive work just like those in rusteron-client. You can attach and manage them using two main approaches:

§1. Implementing a Trait

The recommended approach is to define a trait for your handler and implement it within your own struct. This pattern is performant and safe as it does not require additional allocations. For instance:

use rusteron_archive::*;

pub trait AeronErrorHandlerCallback {
    fn handle_aeron_error_handler(&mut self, errcode: ::std::os::raw::c_int, message: &str) -> ();
}

pub struct AeronErrorHandlerLogger;

impl AeronErrorHandlerCallback for AeronErrorHandlerLogger {
    fn handle_aeron_error_handler(&mut self, _errcode: ::std::os::raw::c_int, _message: &str) -> () {
        println!("{}", stringify!(handle_aeron_error_handler));
    }
}

By passing instances of this trait to the archive context, you gain a reusable and safe way to respond to errors or other events without incurring unnecessary runtime overhead.

§2. Using a Closure

Alternatively, you can use closures as handlers. This approach may be less efficient due to potential allocations, but it is often more convenient for quick, less performance-critical tasks. For example:

use rusteron_archive::*;

pub struct AeronErrorHandlerClosure<F: FnMut(::std::os::raw::c_int, &'static str) -> ()> {
    closure: F,
}

impl<F: FnMut(::std::os::raw::c_int, &'static str) -> ()> AeronErrorHandlerCallback for AeronErrorHandlerClosure<F> {
    fn handle_aeron_error_handler(&mut self, errcode: ::std::os::raw::c_int, message: &'static str) -> () {
        (self.closure)(errcode, message)
    }
}

Closures are wrapped here in a struct so they may be passed into C safely.

§Wrapping Callbacks with Handler

Regardless of the approach, callbacks must be wrapped in a Handler. This ensures proper integration with the Aeron C API. Use Handlers::leak(xxx) to pass a handler into C bindings. When your handler is no longer needed, call release() to free up resources and avoid memory leaks.

§Handler Convenience Methods

If you do not need to set a particular handler, you can pass None. However, doing so manually can be awkward due to static type requirements. To simplify this, rusteron-archive (like rusteron-client) provides convenience methods prefixed with Handlers::no_..., returning None with the correct type signature. For example:

ⓘ
use rusteron_archive::*;
impl Handlers {
    #[doc = r" No handler is set i.e. None with correct type"]
    pub fn no_error_handler_handler() -> Option<&'static Handler<AeronErrorHandlerLogger>> {
        None::<&Handler<AeronErrorHandlerLogger>>
    }
}

These methods make it easy to specify that no handler is required, keeping your code concise.

§Error Handling with Aeron C Bindings

rusteron-archive relies on the same Aeron C bindings as rusteron-client, using i32 error codes to indicate the outcome of operations. In Rust, these are wrapped within a Result<i32, AeronCError> to provide clearer, more idiomatic error handling.

§Error Type Enum

The AeronErrorType enum defines various error types that may occur:

Error TypeDescription
NullOrNotConnectedNull value or not connected
ClientErrorDriverTimeoutDriver timeout error
ClientErrorClientTimeoutClient timeout error
ClientErrorConductorServiceTimeoutConductor service timeout error
ClientErrorBufferFullBuffer is full
PublicationBackPressuredBack pressure on publication
PublicationAdminActionAdmin action during publication
PublicationClosedPublication has been closed
PublicationMaxPositionExceededMaximum position exceeded for publication
PublicationErrorGeneral publication error
TimedOutOperation timed out
Unknown(i32)Unknown error code

These error types help provide more context on the underlying issues when working with Aeron. For example, if a publication is closed or back-pressured, these specific errors can be captured and managed accordingly.

The AeronCError struct encapsulates the error code and provides methods to retrieve the corresponding error type and a human-readable description. Error handling in rusteron-client is designed to make working with Aeron C bindings more ergonomic by providing clear error types and descriptions for easier debugging.

§Safety Considerations

Resource Management:

  1. Lifetime of Aeron: The AeronArchive does not take full ownership or manage the lifetime of the Aeron instance. Instead, it calls AeronArchive::set_aeron, meaning you must ensure the Aeron object remains valid throughout the archive’s usage. Dropping or losing reference to the Aeron too soon can lead to segmentation faults or undefined behaviour.

  2. Unsafe Bindings: Since rusteron-archive relies on Aeron C bindings, you must carefully manage resources (publishers, subscriptions, handlers, etc.) to avoid crashes or undefined behaviour. This includes ensuring you do not publish messages after closing the Aeron client or the associated archive context.

  3. Partial Automatic Resource Management: While constructors aim to manage resources automatically, many aspects of resource lifecycles remain manual. For instance, handlers require a call to release() to clean up memory. Be especially cautious in multithreaded environments, ensuring synchronisation is properly handled.

Failure to follow these guidelines can lead to unstable or unpredictable results.

§Example Usage: Recording and Replaying a Stream with Aeron Archive

Below is an example of how to use AeronArchive to set up a recording, publish messages, and replay the recorded stream.

use rusteron_archive::*;
use rusteron_archive::bindings::*;
use std::time::Duration;
use std::time::Instant;
use std::cell::Cell;
use std::thread::sleep;

let request_port = find_unused_udp_port(8000).expect("Could not find port");
let response_port = find_unused_udp_port(request_port + 1).expect("Could not find port");
let request_control_channel = &format!("aeron:udp?endpoint=localhost:{}", request_port);
let response_control_channel = &format!("aeron:udp?endpoint=localhost:{}", response_port);
let recording_events_channel = &format!("aeron:udp?endpoint=localhost:{}", response_port+1);
assert_ne!(request_control_channel, response_control_channel);

let error_handler = Handler::leak(AeronErrorHandlerClosure::from(|error_code, msg| {
panic!("error {} {}", error_code, msg)
}));

let aeron_context = AeronContext::new()?;
aeron_context.set_client_name("test")?;
aeron_context.set_publication_error_frame_handler(Some(&Handler::leak(
AeronPublicationErrorFrameHandlerLogger,
)))?;
aeron_context.set_error_handler(Some(&error_handler))?;
let aeron = Aeron::new(&aeron_context)?;
aeron.start()?;
println!("connected to aeron");

let archive_context = AeronArchiveContext::new_with_no_credentials_supplier(
    &aeron,
    request_control_channel,
    response_control_channel,
    recording_events_channel,
)?;
let found_recording_signal = Cell::new(false);
archive_context.set_recording_signal_consumer(Some(&Handler::leak(
    AeronArchiveRecordingSignalConsumerFuncClosure::from(
        |signal: AeronArchiveRecordingSignal| {
            println!("signal {:?}", signal);
            found_recording_signal.set(true);
        },
    ),
)))?;
archive_context.set_idle_strategy(Some(&Handler::leak(
    AeronIdleStrategyFuncClosure::from(|_work_count| {}),
)))?;
archive_context.set_error_handler(Some(&error_handler))?;


let connect = AeronArchiveAsyncConnect::new(&archive_context.clone())?;
let archive = connect.poll_blocking(Duration::from_secs(5))?;

// Start recording a channel and stream
let channel = "aeron:ipc";
let stream_id = 10;

let subscription_id = archive.start_recording(
    channel,
    stream_id,
    aeron_archive_source_location_t::AERON_ARCHIVE_SOURCE_LOCATION_LOCAL,
    true,
)?;

println!("subscription id {}", subscription_id);

// Publish messages to be recorded
let publication = aeron
    .async_add_exclusive_publication(channel, stream_id)?
    .poll_blocking(Duration::from_secs(5))?;

let start = Instant::now();
while !found_recording_signal.get() && start.elapsed().as_secs() < 5 {
    sleep(Duration::from_millis(50));
    archive.poll_for_recording_signals()?;
    let err = archive.poll_for_error_response_as_string(4096)?;
    if !err.is_empty() {
        panic!("{}", err);
    }
}
assert!(start.elapsed().as_secs() < 5);

for _ in 0..11 {
    while publication.offer(
        "123456".as_bytes(),
        Handlers::no_reserved_value_supplier_handler(),
    ) <= 0
    {
        sleep(Duration::from_millis(50));
        archive.poll_for_recording_signals()?;
        let err = archive.poll_for_error_response_as_string(4096)?;
        if !err.is_empty() {
            panic!("{}", err);
        }
    }
    println!("sent message");
}
archive.stop_recording_channel_and_stream(channel, stream_id)?;
drop(publication);

// Locate the recorded stream in the archive
println!("list recordings");
let found_recording_id = Cell::new(-1);
let start_pos = Cell::new(-1);
let end_pos = Cell::new(-1);
let handler = Handler::leak(
    AeronArchiveRecordingDescriptorConsumerFuncClosure::from(
        |d: AeronArchiveRecordingDescriptor| {
            println!("found recording {:?}", d);
            found_recording_id.set(d.recording_id);
            start_pos.set(d.start_position);
            end_pos.set(d.stop_position);
        },
    ),
);
let start = Instant::now();
while start.elapsed() < Duration::from_secs(5)
    && found_recording_id.get() == -1
    && archive.list_recordings_for_uri(0, i32::MAX, channel, stream_id, Some(&handler))?
        <= 0
{
    sleep(Duration::from_millis(50));
    archive.poll_for_recording_signals()?;
    let err = archive.poll_for_error_response_as_string(4096)?;
    if !err.is_empty() {
        panic!("{}", err);
    }
}
assert!(start.elapsed() < Duration::from_secs(5));

// Replay the recorded stream on a new stream_id
println!("start replay");
let params = AeronArchiveReplayParams::new(
    0,
    i32::MAX,
    start_pos.get(),
    end_pos.get() - start_pos.get(),
    0,
    0,
)?;
let replay_stream_id = 45;
let replay_session_id =
    archive.start_replay(found_recording_id.get(), channel, replay_stream_id, &params)?;
let session_id = replay_session_id as i32;

println!("replay session id {}", replay_session_id);
println!("session id {}", session_id);
let channel_replay = format!("{}?session-id={}", channel, session_id);
println!("archive id: {}", archive.get_archive_id());

println!("add subscription {}", channel_replay);
let subscription = aeron
    .async_add_subscription(
        &channel_replay,
        replay_stream_id,
        Some(&Handler::leak(AeronAvailableImageLogger)),
        Some(&Handler::leak(AeronUnavailableImageLogger)),
    )?
    .poll_blocking(Duration::from_secs(10))?;

let count = Cell::new(0);
let poll = Handler::leak(AeronFragmentHandlerClosure::from(|msg, header| {
    assert_eq!(msg, "123456".as_bytes().to_vec());
    count.set(count.get() + 1);
}));

let start = Instant::now();
while start.elapsed() < Duration::from_secs(5) && subscription.poll(Some(&poll), 100)? <= 0
{
    archive.poll_for_recording_signals()?;
    let err = archive.poll_for_error_response_as_string(4096)?;
    if !err.is_empty() {
        panic!("{}", err);
    }
}
assert!(start.elapsed() < Duration::from_secs(5));
println!("aeron {:?}", aeron);
println!("ctx {:?}", archive_context);
assert_eq!(11, count.get());
Ok::<(), AeronCError>(())

§Workflow Overview

  1. Initialise Contexts: Set up archive and client contexts.
  2. Start Recording: Begin recording a specified channel and stream.
  3. Publish Messages: Send messages to be captured by the archive.
  4. Stop Recording: Conclude the recording session.
  5. Locate Recording: Identify and retrieve details about the recorded stream.
  6. Replay Setup: Configure replay parameters and replay the recorded messages on a new stream.
  7. Subscribe and Receive: Subscribe to the replay stream, receiving the replayed messages as they appear.

§Building This Project

For full details on building the rusteron project, please refer to the HOW_TO_BUILD.md file.

§Benchmarks

You can view the benchmarks for this project by visiting BENCHMARKS.md.

§Contributing

Contributions are welcome! Please see our contributing guidelines for more information on how to get involved.

§License

This project is dual-licensed under either the MIT License or Apache License 2.0. You may choose which one to use.

Feel free to reach out with any questions or suggestions via GitHub Issues!

§Features

  • static: When enabled, this feature statically links the Aeron C code. By default, the library uses dynamic linking to the Aeron C libraries.

Modules§

Structs§

Enums§

Constants§

Traits§

Functions§

Type Aliases§