[−][src]Trait kompact::prelude::SystemHandle

pub trait SystemHandle: Dispatching + CanCancelTimers {
    fn create<C, F>(&self, f: F) -> Arc<Component<C>>
    where
        F: FnOnce() -> C,
        C: ComponentDefinition + 'static;

    fn register(
        &self, 
        c: &Arc<impl AbstractComponent + ?Sized>
    ) -> KFuture<RegistrationResult>ⓘ
Notable traits for KFuture<T>
impl<T: Send + Sized> Future for KFuture<T>    type Output = Result<T, PromiseDropped>;
;

    fn create_and_register<C, F>(
        &self, 
        f: F
    ) -> (Arc<Component<C>>, KFuture<RegistrationResult>)
    where
        F: FnOnce() -> C,
        C: ComponentDefinition + 'static;

    fn register_by_alias<A>(
        &self, 
        c: &Arc<impl AbstractComponent + ?Sized>, 
        alias: A
    ) -> KFuture<RegistrationResult>ⓘ
Notable traits for KFuture<T>
impl<T: Send + Sized> Future for KFuture<T>    type Output = Result<T, PromiseDropped>;

    where
        A: Into<String>;

    fn update_alias_registration<A>(
        &self, 
        c: &Arc<impl AbstractComponent + ?Sized>, 
        alias: A
    ) -> KFuture<RegistrationResult>ⓘ
Notable traits for KFuture<T>
impl<T: Send + Sized> Future for KFuture<T>    type Output = Result<T, PromiseDropped>;

    where
        A: Into<String>;

    fn set_routing_policy<P>(
        &self, 
        policy: P, 
        path: &str, 
        update: bool
    ) -> KFuture<RegistrationResult>ⓘ
Notable traits for KFuture<T>
impl<T: Send + Sized> Future for KFuture<T>    type Output = Result<T, PromiseDropped>;

    where
        P: Into<StorePolicy>;

    fn start(&self, c: &Arc<impl AbstractComponent + ?Sized>);

    fn start_notify(
        &self, 
        c: &Arc<impl AbstractComponent + ?Sized>
    ) -> KFuture<()>ⓘ
Notable traits for KFuture<T>
impl<T: Send + Sized> Future for KFuture<T>    type Output = Result<T, PromiseDropped>;
;

    fn stop(&self, c: &Arc<impl AbstractComponent + ?Sized>);

    fn stop_notify(
        &self, 
        c: &Arc<impl AbstractComponent + ?Sized>
    ) -> KFuture<()>ⓘ
Notable traits for KFuture<T>
impl<T: Send + Sized> Future for KFuture<T>    type Output = Result<T, PromiseDropped>;
;

    fn kill(&self, c: Arc<impl AbstractComponent + ?Sized>);

    fn kill_notify(
        &self, 
        c: Arc<impl AbstractComponent + ?Sized>
    ) -> KFuture<()>ⓘ
Notable traits for KFuture<T>
impl<T: Send + Sized> Future for KFuture<T>    type Output = Result<T, PromiseDropped>;
;

    fn throughput(&self) -> usize;

    fn max_messages(&self) -> usize;

    fn shutdown_async(&self);

    fn system_path(&self) -> SystemPath;

    fn deadletter_ref(&self) -> ActorRef<Never>;

    fn spawn<R: Send + 'static>(
        &self, 
        future: impl Future<Output = R> + 'static + Send
    ) -> JoinHandle<R>;
}

A limited version of a KompactSystem

This is meant for use from within components, where blocking APIs are unacceptable.

Required methods

fn create<C, F>(&self, f: F) -> Arc<Component<C>> where
    F: FnOnce() -> C,
    C: ComponentDefinition + 'static, 

Create a new component

Uses f to create an instance of a ComponentDefinition, which is the initialised to form a Component. Since components are shared between threads, the created component is wrapped into an Arc.

Newly created components are not started automatically. Use start or start_notify to start a newly created component, once it is connected properly.

If you need address this component via the network, see the register function.

Example

let c = system.create(TestComponent1::new);

fn register(
    &self,
    c: &Arc<impl AbstractComponent + ?Sized>
) -> KFuture<RegistrationResult>ⓘ

Notable traits for KFuture<T>

impl<T: Send + Sized> Future for KFuture<T> type Output = Result<T, PromiseDropped>;

Attempts to register c with the dispatcher using its unique id

The returned future will contain the unique id ActorPath for the given component, once it is completed by the dispatcher.

Once the future completes, the component can be addressed via the network, even if it has not been started, yet (in which case messages will simply be queued up).

Example

use kompact::prelude::*;
use std::time::Duration;
let mut cfg = KompactConfig::new();
cfg.system_components(DeadletterBox::new, {
    let net_config = NetworkConfig::new("127.0.0.1:0".parse().expect("Address should work"));
    net_config.build()
});
let system = cfg.build().expect("KompactSystem");
let c = system.create(TestComponent1::new);
system.register(&c).wait_expect(Duration::from_millis(1000), "Failed to register TestComponent1");

fn create_and_register<C, F>(
    &self,
    f: F
) -> (Arc<Component<C>>, KFuture<RegistrationResult>) where
    F: FnOnce() -> C,
    C: ComponentDefinition + 'static, 

Creates a new component and registers it with the dispatcher

This function is simply a convenience shortcut for create followed by register, as this combination is very common in networked Kompact systems.

Example

use kompact::prelude::*;
use std::time::Duration;
let mut cfg = KompactConfig::new();
cfg.system_components(DeadletterBox::new, {
    let net_config = NetworkConfig::new("127.0.0.1:0".parse().expect("Address should work"));
    net_config.build()
});
let system = cfg.build().expect("KompactSystem");
let (c, registration_future) = system.create_and_register(TestComponent1::new);
registration_future.wait_expect(Duration::from_millis(1000), "Failed to register TestComponent1");

fn register_by_alias<A>(
    &self,
    c: &Arc<impl AbstractComponent + ?Sized>,
    alias: A
) -> KFuture<RegistrationResult>ⓘ

Notable traits for KFuture<T>

impl<T: Send + Sized> Future for KFuture<T> type Output = Result<T, PromiseDropped>;
 where
    A: Into<String>, 

Attempts to register the provided component with a human-readable alias.

The returned future will contain the named ActorPath for the given alias, once it is completed by the dispatcher.

Alias registration will fail if a previous registration already exists. Use update_alias_registration to override an existing registration.

Note

While aliases are easier to read, lookup by unique ids is significantly more efficient. However, named aliases allow services to be taken over by another component when the original registrant failed, something that is not possible with unique paths. Thus, this kind of addressing lends itself to lookup-service style components, for example.

Example

use kompact::prelude::*;
use std::time::Duration;
let mut cfg = KompactConfig::new();
cfg.system_components(DeadletterBox::new, {
    let net_config = NetworkConfig::new("127.0.0.1:0".parse().expect("Address should work"));
    net_config.build()
});
let system = cfg.build().expect("KompactSystem");
let (c, unique_registration_future) = system.create_and_register(TestComponent1::new);
unique_registration_future.wait_expect(Duration::from_millis(1000), "Failed to register TestComponent1");
let alias_registration_future = system.register_by_alias(&c, "test");
alias_registration_future.wait_expect(Duration::from_millis(1000), "Failed to register TestComponent1 by alias");

fn update_alias_registration<A>(
    &self,
    c: &Arc<impl AbstractComponent + ?Sized>,
    alias: A
) -> KFuture<RegistrationResult>ⓘ

Notable traits for KFuture<T>

impl<T: Send + Sized> Future for KFuture<T> type Output = Result<T, PromiseDropped>;
 where
    A: Into<String>, 

Attempts to register the provided component with a human-readable alias.

The returned future will contain the named ActorPath for the given alias, once it is completed by the dispatcher.

This registration will replace any previous registration, if it exists.

Note

While aliases are easier to read, lookup by unique ids is significantly more efficient. However, named aliases allow services to be taken over by another component when the original registrant failed, something that is not possible with unique paths. Thus, this kind of addressing lends itself to lookup-service style components, for example.

Example

use kompact::prelude::*;
use std::time::Duration;
let mut cfg = KompactConfig::new();
cfg.system_components(DeadletterBox::new, {
    let net_config = NetworkConfig::new("127.0.0.1:0".parse().expect("Address should work"));
    net_config.build()
});
let system = cfg.build().expect("KompactSystem");
let (c, unique_registration_future) = system.create_and_register(TestComponent1::new);
unique_registration_future.wait_expect(Duration::from_millis(1000), "Failed to register TestComponent1");
let alias_registration_future = system.update_alias_registration(&c, "test");
alias_registration_future.wait_expect(Duration::from_millis(1000), "Failed to register TestComponent1 by alias");
let alias_reregistration_future = system.update_alias_registration(&c, "test");
alias_reregistration_future.wait_expect(Duration::from_millis(1000), "Failed to override TestComponent1 registration by alias");

fn set_routing_policy<P>(
    &self,
    policy: P,
    path: &str,
    update: bool
) -> KFuture<RegistrationResult>ⓘ

Notable traits for KFuture<T>

impl<T: Send + Sized> Future for KFuture<T> type Output = Result<T, PromiseDropped>;
 where
    P: Into<StorePolicy>, 

Attempts to set the routing policy at path

Setting a routing policy at a path "a" will include all actors paths registered under paths of the form "a/..." to be included as members of the routing group that the policy applies to.

Having an explicit routing policy at a path will cause routing over group members, even if no routing marker (e.g., "a/*" or "a/?" is given).

Overriding an existing actor or policy at the given path will fail, unless update is set to true.

Provided routing policies can be found in the routing::groups module.

Example

use kompact::prelude::*;
use kompact::routing::groups::*;
use std::time::Duration;
let mut cfg = KompactConfig::new();
cfg.system_components(DeadletterBox::new, {
    let net_config = NetworkConfig::new("127.0.0.1:0".parse().expect("Address should work"));
    net_config.build()
});
let system = cfg.build().expect("KompactSystem");
let policy_registration_future = system.set_routing_policy(BroadcastRouting::default(), "broadcast-me", false);
let broadcast_path = policy_registration_future.wait_expect(Duration::from_millis(1000), "Failed to set broadcast policy");
let c1 = system.create(TestComponent1::new);
let c2 = system.create(TestComponent1::new);
let alias_registration_future1 = system.register_by_alias(&c1, "broadcast-me/test1");
let alias_registration_future2 = system.register_by_alias(&c2, "broadcast-me/something/test2");
alias_registration_future1.wait_expect(Duration::from_millis(1000), "Failed to register TestComponent1 by alias");
alias_registration_future2.wait_expect(Duration::from_millis(1000), "Failed to register TestComponent2 by alias");
// sending to broadcast_path now will send to c1 and c2

fn start(&self, c: &Arc<impl AbstractComponent + ?Sized>)

Start a component

A component only handles events/messages once it is started. In particular, a component that isn't started shouldn't be scheduled and thus access to its definition should always succeed, for example via on_definition.

Example

let c = system.create(TestComponent1::new);
system.start(&c);

fn start_notify(&self, c: &Arc<impl AbstractComponent + ?Sized>) -> KFuture<()>ⓘ

Notable traits for KFuture<T>

impl<T: Send + Sized> Future for KFuture<T> type Output = Result<T, PromiseDropped>;

Start a component and complete a future once it has started

When the returned future completes, the component is guaranteed to have started. However, it is not guaranteed to be in an active state, as it could already have been stopped or could have failed since.

A component only handles events/messages once it is started. In particular, a component that isn't started shouldn't be scheduled and thus access to its definition should always succeed, for example via on_definition.

Example

use std::time::Duration;
let c = system.create(TestComponent1::new);
system.start_notify(&c)
      .wait_timeout(Duration::from_millis(1000))
      .expect("TestComponent1 never started!");

fn stop(&self, c: &Arc<impl AbstractComponent + ?Sized>)

Stop a component

A component does not handle any events/messages while it is stopped, but it does not get deallocated either. It can be started again later with start or start_notify.

A component that is stopped shouldn't be scheduled and thus access to its definition should always succeed, for example via on_definition.

Example

use std::time::Duration;
let c = system.create(TestComponent1::new);
system.start_notify(&c)
      .wait_timeout(Duration::from_millis(1000))
      .expect("TestComponent1 never started!");
system.stop(&c);

fn stop_notify(&self, c: &Arc<impl AbstractComponent + ?Sized>) -> KFuture<()>ⓘ

Notable traits for KFuture<T>

impl<T: Send + Sized> Future for KFuture<T> type Output = Result<T, PromiseDropped>;

Stop a component and complete a future once it has stopped

When the returned future completes, the component is guaranteed to have stopped. However, it is not guaranteed to be in a passive state, as it could already have been started again since.

A component does not handle any events/messages while it is stopped, but it does not get deallocated either. It can be started again later with start or start_notify.

A component that is stopped shouldn't be scheduled and thus access to its definition should always succeed, for example via on_definition.

Example

use std::time::Duration;
let c = system.create(TestComponent1::new);
system.start_notify(&c)
      .wait_timeout(Duration::from_millis(1000))
      .expect("TestComponent1 never started!");
system.stop_notify(&c)
      .wait_timeout(Duration::from_millis(1000))
      .expect("TestComponent1 never stopped!");
system.start_notify(&c)
      .wait_timeout(Duration::from_millis(1000))
      .expect("TestComponent1 never re-started!");

fn kill(&self, c: Arc<impl AbstractComponent + ?Sized>)

Stop and deallocate a component

The supervisor will attempt to deallocate c once it is stopped. However, if there are still outstanding references somewhere else in the system this will fail, of course. In that case the supervisor leaves a debug message in the logging output, so that this circumstance can be discovered if necessary.

Example

use std::time::Duration;
let c = system.create(TestComponent1::new);
system.start_notify(&c)
      .wait_timeout(Duration::from_millis(1000))
      .expect("TestComponent1 never started!");
system.kill(c);

fn kill_notify(&self, c: Arc<impl AbstractComponent + ?Sized>) -> KFuture<()>ⓘ

Notable traits for KFuture<T>

impl<T: Send + Sized> Future for KFuture<T> type Output = Result<T, PromiseDropped>;

Stop and deallocate a component, and complete a future once it has stopped

The supervisor will attempt to deallocate c once it is stopped. However, if there are still outstanding references somewhere else in the system this will fail, of course. In that case the supervisor leaves a debug message in the logging output, so that this circumstance can be discovered if necessary.

Note

The completion of the future indicates that the component has been stopped, not that it has been deallocated.

If, for some reason, you really need to know when it has been deallocated, you need to hold on to a copy of the component, use try_unwrap and then call drop once you are successful.

Example

use std::time::Duration;
let c = system.create(TestComponent1::new);
system.start_notify(&c)
      .wait_timeout(Duration::from_millis(1000))
      .expect("TestComponent1 never started!");
system.kill_notify(c)
      .wait_timeout(Duration::from_millis(1000))
      .expect("TestComponent1 never stopped!");

fn throughput(&self) -> usize

Return the configured thoughput value

See also throughput.

fn max_messages(&self) -> usize

Return the configured maximum number of messages per scheduling

This value is based on throughput and msg_priority.

fn shutdown_async(&self)

Shutdown the Kompact system from within a component

Stops all components and then stops the scheduler.

This function may fail to stop in time (or at all), if components hang on to scheduler threads indefinitely.

Example

use kompact::prelude::*;

#[derive(ComponentDefinition, Actor)]
struct Stopper {
   ctx: ComponentContext<Self>,
}
impl Stopper {
    fn new() -> Stopper {
        Stopper {
            ctx: ComponentContext::uninitialised(),
        }
    }    
}
impl ComponentLifecycle for Stopper {
   fn on_start(&mut self) -> Handled {
       self.ctx().system().shutdown_async();
       Handled::Ok
    }
}
let system = KompactConfig::default().build().expect("system");
let c = system.create(Stopper::new);
system.start(&c);
system.await_termination();

fn system_path(&self) -> SystemPath

Return the system path of this Kompact system

The system path forms a prefix for every ActorPath.

fn deadletter_ref(&self) -> ActorRef<Never>

Returns a reference to the system's deadletter box

fn spawn<R: Send + 'static>(
    &self,
    future: impl Future<Output = R> + 'static + Send
) -> JoinHandle<R>

Run a Future on this system's executor pool and return a handle to the result

Handles can be awaited like any other future.

Loading content...

Implementors

impl SystemHandle for ContextSystemHandle[src]

Loading content...