Struct MainRuntime 

pub struct MainRuntime(/* private fields */);

The runtime the app runs on, as a handler reaches it: Ctx(main): Ctx<MainRuntime> as a parameter, or Context::main_runtime.

A handler on dedicated threads (threads(n)) runs on its own thread, and so does everything it leaves behind: a plain tokio::spawn, a timer, an and_after continuation, an after(..) hook and the timer of a retry_after copy. MainRuntime is the one explicit way to send work to the app’s runtime instead: spawn starts it there and moves on, run runs it there and hands its output back. What is sent must be Send. A plain tokio::spawn on a dedicated thread lives as long as that thread: a task still pending when the subscription ends is dropped with the thread’s runtime. On any other placement the handler already runs on this runtime.

Examples

use ruststream::prelude::*;

#[derive(serde::Deserialize)]
struct Job {
    id: u64,
}

async fn notify(_id: u64) {}

async fn lookup(id: u64) -> u64 {
    id * 2
}

#[subscriber("jobs", threads(4))]
async fn index(job: &Job, Ctx(main): Ctx<MainRuntime>) -> HandlerOutcome {
    // The notification leaves the thread; the handler does not wait for it.
    main.spawn(notify(job.id));
    // The lookup runs on the app's runtime, and its answer comes back here.
    match main.run(lookup(job.id)).await {
        Ok(_answer) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

Implementations

impl MainRuntime

pub fn spawn<Work>(&self, future: Work) -> JoinHandle<<Work as Future>::Output>

where Work: Future + Send + 'static, <Work as Future>::Output: Send + 'static,

Starts future as a task of the app’s runtime and returns its handle; the handler does not wait for it unless it awaits the handle.

Examples

use ruststream::IncomingMessage;
use ruststream::runtime::{Context, HandlerOutcome};

async fn notify(_id: u64) {}

async fn handle<M: IncomingMessage>(_msg: &M, ctx: &mut Context<'_>) -> HandlerOutcome {
    ctx.main_runtime().spawn(notify(7));
    HandlerOutcome::ack()
}

pub fn run<Work>( &self, future: Work, ) -> impl Future<Output = Result<<Work as Future>::Output, JoinError>> + Send + 'static

where Work: Future + Send + 'static, <Work as Future>::Output: Send + 'static,

Runs future as a task of the app’s runtime and resolves to its output.

Errors

Returns the JoinError of a task that panicked, or that the runtime cancelled because it is shutting down.

Cancel safety

Dropping the returned future stops waiting, not the work: the task runs to its end on the app’s runtime.

Examples

use ruststream::IncomingMessage;
use ruststream::runtime::{Context, HandlerOutcome};

async fn lookup(id: u64) -> u64 {
    id * 2
}

async fn handle<M: IncomingMessage>(_msg: &M, ctx: &mut Context<'_>) -> HandlerOutcome {
    match ctx.main_runtime().run(lookup(7)).await {
        Ok(14) => HandlerOutcome::ack(),
        _ => HandlerOutcome::retry(),
    }
}

pub const fn as_handle(&self) -> &Handle

The Tokio handle of the app’s runtime, for what the two methods above do not cover (block_on from a thread outside any runtime, the runtime’s metrics).

Examples

use ruststream::IncomingMessage;
use ruststream::runtime::{Context, HandlerOutcome};

async fn handle<M: IncomingMessage>(_msg: &M, ctx: &mut Context<'_>) -> HandlerOutcome {
    // The app's runtime metrics, read from a handler on a thread of its own.
    let metrics = ctx.main_runtime().as_handle().metrics();
    tracing::info!(
        workers = metrics.num_workers(),
        alive = metrics.num_alive_tasks(),
        "app runtime"
    );
    HandlerOutcome::ack()
}

Trait Implementations

impl Clone for MainRuntime

fn clone(&self) -> MainRuntime

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl CtxKey for MainRuntime

type Value = MainRuntime

The owned value the extractor binds.

impl Debug for MainRuntime

fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error>

Formats the value using the given formatter.