Struct NetMessage 

pub struct NetMessage {
    pub sender: ActorPath,
    pub receiver: ActorPath,
    pub data: NetData,
    pub session: Option<SessionId>,
}

An incoming message from the distributed dispatch subsystem

Provides actor paths for the sender of the message and the receiver in addition to the actual data.

This type stays in kompact rather than kompact-net, because local path-based dispatch and alternative transport backends use the same envelope type even when no socket-backed transport is involved.

It is recommend to use try_deserialise or try_deserialise_unchecked to unpack the contained data to the appropriate type. These methods abstract over whether or not the data is actually serialised or simply heap-allocated.

Fields

sender: ActorPath

The sender of the message

More concretely, this is the actor path that was supplied as a source by the original sender.

receiver: ActorPath

The receiver of the message

More concretely, this is the actor path that was used as a destination for the message. In particular, of the message was sent to a named path for the current component, instead of the unique path, then this will be the named path.

data: NetData

The actual data of the message

session: Option<SessionId>

Each physical end-to-end session between two systems is assigned a unique identifier. When a connection is lost/closed and then re-established, the connection will have a new unique identifier. The unique identifier is negotiated and is the same on both ends of the connection.

The field is only set by the Network-layer in incoming messages. All other messages will have the field set to None.

For any sequence of two messages received from a remote actor over a session:

• If the session of the messages differs, an intermediate message may have been lost.
• Conversely, if the session does not differ no intermediate message was lost.

Implementations

impl NetMessage

pub fn with_box( ser_id: u64, sender: ActorPath, receiver: ActorPath, data: Box<dyn Serialisable>, ) -> NetMessage

Create a network message with heap-allocated data

pub fn with_bytes( ser_id: u64, sender: ActorPath, receiver: ActorPath, data: Bytes, ) -> NetMessage

Create a network message with serialised data

pub fn with_chunk_lease( ser_id: u64, sender: ActorPath, receiver: ActorPath, data: ChunkLease, ) -> NetMessage

Create a network message with a ChunkLease, pooled buffers. For outgoing network messages the data inside the data should be both serialised and (framed)crate::net::frames.

pub fn with_chunk_ref( ser_id: u64, sender: ActorPath, receiver: ActorPath, data: ChunkRef, ) -> NetMessage

Create a network message with a ChunkLease, pooled buffers. For outgoing network messages the data inside the data should be both serialised and (framed)crate::net::frames.

pub fn sender(&self) -> &ActorPath

Return a reference to the sender field

pub fn session(&self) -> Option<SessionId>

Returns the session of the NetMessage

pub fn set_session(&mut self, session: SessionId)

Sets the SessionId of the NetMessage

pub fn try_into_deserialised<T, D>( self, ) -> Result<DeserialisedMessage<T>, UnpackError<NetMessage>>

where T: 'static, D: Deserialiser<T>,

Try to deserialise the data into a value of type T wrapped into a message

This method attempts to deserialise the contents into an instance of T using the deserialiser D. It will only do so after verifying that ser_id == D::SER_ID.

If the serialisation id does not match, this message is returned unaltered wrapped in an UnpackError.

Example

use kompact::prelude::*;
use bytes::BytesMut;


let test_str = "Test me".to_string();
// serialise the string
let mut mbuf = BytesMut::with_capacity(test_str.size_hint().expect("size hint"));
test_str.serialise(&mut mbuf).expect("serialise");
// create a net message
let buf = mbuf.freeze();
let msg = NetMessage::with_bytes(String::SER_ID, some_path, some_path2, buf);
// try to deserialise it again
match msg.try_into_deserialised::<u64, u64>() {
    Ok(_) => unreachable!("It's definitely not a u64..."),
    Err(UnpackError::NoIdMatch(msg_again)) => {
        match msg_again.try_into_deserialised::<String, String>() {
            Ok(test_msg) => assert_eq!(test_str, test_msg.content),
            Err(_) => unreachable!("It's definitely a string..."),
        }   
    }
    Err(error) => panic!("Not the error we expected: {:?}", error),  
}

pub fn try_deserialise<T, D>(self) -> Result<T, UnpackError<NetMessage>>

where T: 'static, D: Deserialiser<T>,

Try to deserialise the data into a value of type T

This method attempts to deserialise the contents into an instance of T using the deserialiser D. It will only do so after verifying that ser_id == D::SER_ID.

If the serialisation id does not match, this message is returned unaltered wrapped in an UnpackError.

Example

use kompact::prelude::*;
use bytes::BytesMut;


let test_str = "Test me".to_string();
// serialise the string
let mut mbuf = BytesMut::with_capacity(test_str.size_hint().expect("size hint"));
test_str.serialise(&mut mbuf).expect("serialise");
// create a net message
let buf = mbuf.freeze();
let msg = NetMessage::with_bytes(String::SER_ID, some_path, some_path2, buf);
// try to deserialise it again
match msg.try_deserialise::<u64, u64>() {
    Ok(_) => unreachable!("It's definitely not a u64..."),
    Err(UnpackError::NoIdMatch(msg_again)) => {
        match msg_again.try_deserialise::<String, String>() {
            Ok(test_res) => assert_eq!(test_str, test_res),
            Err(_) => unreachable!("It's definitely a string..."),
        }   
    }
    Err(error) => panic!("Not the error we expected: {:?}", error),  
}

Note

If you need the sender or the receiver to be owned after deserialisation, either use msg.data.try_deserialise<...>(...) instead, or use try_into_deserialised.

pub fn try_deserialise_unchecked<T, D>( self, ) -> Result<T, UnpackError<NetMessage>>

where T: 'static, D: Deserialiser<T>,

Try to deserialise the data into a value of type T

This method attempts to deserialise the contents into an instance of T using the deserialiser D without checking the ser_id first for a match.

Only use this, if you have already verified that ser_id == D::SER_ID! Otherwise use try_deserialise.

Example

use kompact::prelude::*;
use bytes::BytesMut;


let test_str = "Test me".to_string();
// serialise the string
let mut mbuf = BytesMut::with_capacity(test_str.size_hint().expect("size hint"));
test_str.serialise(&mut mbuf).expect("serialise");
// create a net message
let buf = mbuf.freeze();
let msg = NetMessage::with_bytes(String::SER_ID, some_path, some_path2, buf);
// try to deserialise it again
match msg.ser_id() {
    &u64::SER_ID => unreachable!("It's definitely not a u64..."),
    &String::SER_ID => {
        let test_res = msg.try_deserialise_unchecked::<String, String>().expect("deserialised");
        assert_eq!(test_str, test_res);
    }
    _ => unreachable!("It's definitely not...whatever this is..."),
}

Note

The match_deser macro generates code that is approximately equivalent to the example above with some nicer syntax.

If you need the sender or the receiver to be owned after deserialisation, either use msg.data.try_deserialise_unchecked<...>(...) instead, or use try_into_deserialised.

pub fn ser_id(&self) -> &u64

Returns a reference to the serialisation id of this message

Trait Implementations

impl Debug for NetMessage

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

Formats the value using the given formatter.

impl RoutingPolicy<DynActorRef, NetMessage> for BroadcastRouting

fn route( &self, members: &[&DynActorRef], msg: NetMessage, logger: &Logger<Arc<Fuse<Async>>>, )

Route the msg to the appropriate members

fn boxed_clone( &self, ) -> Box<dyn RoutingPolicy<DynActorRef, NetMessage> + Send + Sync>

Create a boxed copy of this policy

fn broadcast( &self, ) -> Option<&(dyn RoutingPolicy<DynActorRef, NetMessage> + Send + Sync)>

Provide the broadcast part of this policy, if any

fn select( &self, ) -> Option<&(dyn RoutingPolicy<DynActorRef, NetMessage> + Send + Sync)>

Provide the select part of this policy, if any

impl<H> RoutingPolicy<DynActorRef, NetMessage> for FieldHashBucketRouting<NetMessage, ActorPath, H>

where H: BuildHasher + Clone + Send + Sync + 'static,

fn route( &self, members: &[&DynActorRef], msg: NetMessage, logger: &Logger<Arc<Fuse<Async>>>, )

Route the msg to the appropriate members

fn boxed_clone( &self, ) -> Box<dyn RoutingPolicy<DynActorRef, NetMessage> + Send + Sync>

Create a boxed copy of this policy

fn broadcast( &self, ) -> Option<&(dyn RoutingPolicy<DynActorRef, NetMessage> + Send + Sync)>

Provide the broadcast part of this policy, if any

fn select( &self, ) -> Option<&(dyn RoutingPolicy<DynActorRef, NetMessage> + Send + Sync)>

Provide the select part of this policy, if any

impl RoutingPolicy<DynActorRef, NetMessage> for RoundRobinRouting

fn route( &self, members: &[&DynActorRef], msg: NetMessage, logger: &Logger<Arc<Fuse<Async>>>, )

Route the msg to the appropriate members

fn boxed_clone( &self, ) -> Box<dyn RoutingPolicy<DynActorRef, NetMessage> + Send + Sync>

Create a boxed copy of this policy

fn broadcast( &self, ) -> Option<&(dyn RoutingPolicy<DynActorRef, NetMessage> + Send + Sync)>

Provide the broadcast part of this policy, if any

fn select( &self, ) -> Option<&(dyn RoutingPolicy<DynActorRef, NetMessage> + Send + Sync)>

Provide the select part of this policy, if any

impl TryClone for NetMessage

fn try_clone(&self) -> Result<NetMessage, SerError>

Tries to produce a copy of self or returns an error