Struct ockam_node::Context

pub struct Context { /* private fields */ }

Context contains Node state and references to the runtime.

Implementations

impl Context

pub fn runtime(&self) -> &Handle

Return runtime clone

impl Context

pub fn address(&self) -> Address

Return the primary address of the current worker

pub fn aliases(&self) -> Vec<Address>

Return all addresses of the current worker

pub fn mailboxes(&self) -> &Mailboxes

Return a reference to the mailboxes of this context

pub async fn new_detached_with_mailboxes(
    &self,
    mailboxes: Mailboxes
) -> Result<DetachedContext>

TODO basically we can just rename Self::new_detached_impl()

pub async fn new_detached(
    &self,
    address: impl Into<Address>,
    incoming: impl IncomingAccessControl,
    outgoing: impl OutgoingAccessControl
) -> Result<DetachedContext>

Create a new detached Context without spawning a full worker

Note: this function is very low-level. For most users start_worker() is the recommended way to create a new worker context.

pub async fn start_worker<NM, NW>(
    &self,
    address: impl Into<Address>,
    worker: NW,
    incoming: impl IncomingAccessControl,
    outgoing: impl OutgoingAccessControl
) -> Result<()>where
    NM: Message + Send + 'static,
    NW: Worker<Context = Context, Message = NM>,

Start a new worker instance at the given address

A worker is an asynchronous piece of code that can send and receive messages of a specific type. This type is encoded via the Worker trait. If your code relies on a manual run-loop you may want to use start_processor() instead!

Each address in the set must be unique and unused on the current node. Workers must implement the Worker trait and be thread-safe. Workers run asynchronously and will be scheduled independently of each other. To wait for the initialisation of your worker to complete you can use wait_for().

use ockam_core::{AllowAll, Result, Worker, worker};
use ockam_node::Context;

struct MyWorker;

#[worker]
impl Worker for MyWorker {
    type Context = Context;
    type Message = String;
}

async fn start_my_worker(ctx: &mut Context) -> Result<()> {
    ctx.start_worker("my-worker-address", MyWorker, AllowAll, AllowAll).await
}

pub async fn start_processor<P>(
    &self,
    address: impl Into<Address>,
    processor: P,
    incoming: impl IncomingAccessControl,
    outgoing: impl OutgoingAccessControl
) -> Result<()>where
    P: Processor<Context = Context>,

Start a new processor instance at the given address

A processor is an asynchronous piece of code that runs a custom run loop, with access to a worker context to send and receive messages. If your code is built around responding to message events, consider using start_worker() instead!

pub async fn stop_worker<A: Into<Address>>(&self, addr: A) -> Result<()>

Shut down a local worker by its primary address

pub async fn stop_processor<A: Into<Address>>(&self, addr: A) -> Result<()>

Shut down a local processor by its address

pub async fn stop_now(&mut self) -> Result<()>

Signal to the local runtime to shut down immediately

WARNING: calling this function may result in data loss. It is recommended to use the much safer Context::stop function instead!

pub async fn stop(&mut self) -> Result<()>

Signal to the local runtime to shut down

This call will hang until a safe shutdown has been completed. The default timeout for a safe shutdown is 1 second. You can change this behaviour by calling Context::stop_timeout directly.

pub async fn stop_timeout(&mut self, seconds: u8) -> Result<()>

Signal to the local runtime to shut down

This call will hang until a safe shutdown has been completed or the desired timeout has been reached.

pub async fn send_and_receive<R, M, N>(&self, route: R, msg: M) -> Result<N>where
    R: Into<Route>,
    M: Message + Send + 'static,
    N: Message,

Using a temporary new context, send a message and then receive a message

This helper function uses new_detached, send, and receive internally. See their documentation for more details.

pub async fn send_and_receive_with_timeout<R, M, N>(
    &self,
    route: R,
    msg: M,
    timeout: Duration
) -> Result<N>where
    R: Into<Route>,
    M: Message + Send + 'static,
    N: Message,

Using a temporary new context, send a message and then receive a message with custom timeout

This helper function uses new_detached, send, and receive internally. See their documentation for more details.

pub async fn send_to_self<A, M>(&self, from: A, addr: A, msg: M) -> Result<()>where
    A: Into<Address>,
    M: Message + Send + 'static,

Send a message to another address associated with this worker

This function is a simple wrapper around Self::send() which validates the address given to it and will reject invalid addresses.

pub async fn send<R, M>(&self, route: R, msg: M) -> Result<()>where
    R: Into<Route>,
    M: Message + Send + 'static,

Send a message to an address or via a fully-qualified route

Routes can be constructed from a set of Addresses, or via the RouteBuilder type. Routes can contain middleware router addresses, which will re-address messages that need to be handled by specific domain workers.

use ockam_core::Message;
use serde::{Serialize, Deserialize};

#[derive(Message, Serialize, Deserialize)]
struct MyMessage(String);

impl MyMessage {
    fn new(s: &str) -> Self {
        Self(s.into())
    }
}

ctx.send("my-test-worker", MyMessage::new("Hello you there :)")).await?;
Ok(())

pub async fn send_with_local_info<R, M>(
    &self,
    route: R,
    msg: M,
    local_info: Vec<LocalInfo>
) -> Result<()>where
    R: Into<Route>,
    M: Message + Send + 'static,

Send a message to an address or via a fully-qualified route after attaching the given LocalInfo to the message.

pub async fn send_from_address<R, M>(
    &self,
    route: R,
    msg: M,
    sending_address: Address
) -> Result<()>where
    R: Into<Route>,
    M: Message + Send + 'static,

Send a message to an address or via a fully-qualified route

Routes can be constructed from a set of Addresses, or via the RouteBuilder type. Routes can contain middleware router addresses, which will re-address messages that need to be handled by specific domain workers.

This function additionally takes the sending address parameter, to specify which of a worker’s (or processor’s) addresses should be used.

pub async fn forward(&self, local_msg: LocalMessage) -> Result<()>

Forward a transport message to its next routing destination

Similar to Context::send, but taking a TransportMessage, which contains the full destination route, and calculated return route for this hop.

Note: you most likely want to use Context::send instead, unless you are writing an external router implementation for ockam node.

pub async fn forward_from_address(
    &self,
    local_msg: LocalMessage,
    sending_address: Address
) -> Result<()>

Forward a transport message to its next routing destination

Similar to Context::send, but taking a TransportMessage, which contains the full destination route, and calculated return route for this hop.

Note: you most likely want to use Context::send instead, unless you are writing an external router implementation for ockam node.

pub async fn receive_block<M: Message>(&mut self) -> Result<Cancel<'_, M>>

Block the current worker to wait for a typed message

Warning this function will wait until its running ockam node is shut down. A safer variant of this function is receive and receive_timeout.

pub async fn receive<M: Message>(&mut self) -> Result<Cancel<'_, M>>

Block the current worker to wait for a typed message

This function may return a Err(FailedLoadData) if the underlying worker was shut down, or Err(Timeout) if the call was waiting for longer than the default timeout. Use receive_timeout to adjust the timeout period.

Will return None if the corresponding worker has been stopped, or the underlying Node has shut down.

pub async fn receive_duration_timeout<M: Message>(
    &mut self,
    timeout_duration: Duration
) -> Result<Cancel<'_, M>>

Wait to receive a message up to a specified timeout

See receive for more details.

pub async fn receive_timeout<M: Message>(
    &mut self,
    timeout_secs: u64
) -> Result<Cancel<'_, M>>

Wait to receive a message up to a specified timeout

See receive for more details.

pub async fn receive_match<M, F>(&mut self, check: F) -> Result<Cancel<'_, M>>where
    M: Message,
    F: Fn(&M) -> bool,

Block the current worker to wait for a message satisfying a conditional

Will return Err if the corresponding worker has been stopped, or the underlying node has shut down. This operation has a default timeout.

Internally this function uses receive, so is subject to the same timeout.

pub async fn set_cluster<S: Into<String>>(&self, label: S) -> Result<()>

Assign the current worker to a cluster

A cluster is a set of workers that should be stopped together when the node is stopped or parts of the system are reloaded. This is not to be confused with supervisors!

By adding your worker to a cluster you signal to the runtime that your worker may be depended on by other workers that should be stopped first.

Your cluster name MUST NOT start with _internals. or ockam.!

Clusters are de-allocated in reverse order of their initialisation when the node is stopped.

pub async fn list_workers(&self) -> Result<Vec<Address>>

Return a list of all available worker addresses on a node

pub async fn register<A: Into<Address>>(
    &self,
    type_: TransportType,
    addr: A
) -> Result<()>

Register a router for a specific address type

pub async fn wait_for<A: Into<Address>>(&mut self, addr: A) -> Result<()>

Wait for a particular address to become “ready”

Trait Implementations

impl AsyncTryClone for Context

fn async_try_clone<'life0, 'async_trait>(
    &'life0 self
) -> Pin<Box<dyn Future<Output = Result<Self>> + Send + 'async_trait>>where
    Self: 'async_trait,
    'life0: 'async_trait,

Try cloning a object and return an Err in case of failure.

impl Drop for Context

fn drop(&mut self)

Executes the destructor for this type.