aea.mail.base
Mail module abstract base classes.
AEAConnectionError
class AEAConnectionError(Exception)
Exception class for connection errors.
Empty
class Empty(Exception)
Exception for when the inbox is empty.
URI
class URI()
URI following RFC3986.
__init__
| __init__(uri_raw: str)
Initialize the URI.
Must follow: https://tools.ietf.org/html/rfc3986.html
Arguments:
uri_raw: the raw form uri
Raises:
ValueError: if uri_raw is not RFC3986 compliant
scheme
| @property
| scheme() -> str
Get the scheme.
netloc
| @property
| netloc() -> str
Get the netloc.
path
| @property
| path() -> str
Get the path.
params
| @property
| params() -> str
Get the params.
query
| @property
| query() -> str
Get the query.
fragment
| @property
| fragment() -> str
Get the fragment.
username
| @property
| username() -> Optional[str]
Get the username.
password
| @property
| password() -> Optional[str]
Get the password.
host
| @property
| host() -> Optional[str]
Get the host.
port
| @property
| port() -> Optional[int]
Get the port.
__str__
| __str__()
Get string representation.
__eq__
| __eq__(other)
Compare with another object.
EnvelopeContext
class EnvelopeContext()
Extra information for the handling of an envelope.
__init__
| __init__(connection_id: Optional[PublicId] = None, uri: Optional[URI] = None)
Initialize the envelope context.
Arguments:
connection_id: the connection id used for routing the outgoing envelope in the multiplexer.uri: the URI sent with the envelope.
uri_raw
| @property
| uri_raw() -> str
Get uri in string format.
__str__
| __str__()
Get the string representation.
__eq__
| __eq__(other)
Compare with another object.
EnvelopeSerializer
class EnvelopeSerializer(ABC)
Abstract class to specify the serialization layer for the envelope.
encode
| @abstractmethod
| encode(envelope: "Envelope") -> bytes
Encode the envelope.
Arguments:
envelope: the envelope to encode
Returns:
the encoded envelope
decode
| @abstractmethod
| decode(envelope_bytes: bytes) -> "Envelope"
Decode the envelope.
Arguments:
envelope_bytes: the encoded envelope
Returns:
the envelope
ProtobufEnvelopeSerializer
class ProtobufEnvelopeSerializer(EnvelopeSerializer)
Envelope serializer using Protobuf.
encode
| encode(envelope: "Envelope") -> bytes
Encode the envelope.
Arguments:
envelope: the envelope to encode
Returns:
the encoded envelope
decode
| decode(envelope_bytes: bytes) -> "Envelope"
Decode the envelope.
Arguments:
envelope_bytes: the encoded envelope
Returns:
the envelope
Envelope
class Envelope()
The top level message class for agent to agent communication.
__init__
| __init__(to: Address, sender: Address, protocol_id: ProtocolId, message: bytes, context: Optional[EnvelopeContext] = None)
Initialize a Message object.
Arguments:
to: the address of the receiver.sender: the address of the sender.protocol_id: the protocol id.message: the protocol-specific message.context: the optional envelope context.
to
| @to.setter
| to(to: Address) -> None
Set address of receiver.
sender
| @sender.setter
| sender(sender: Address) -> None
Set address of sender.
protocol_id
| @protocol_id.setter
| protocol_id(protocol_id: ProtocolId) -> None
Set the protocol id.
message
| @message.setter
| message(message: bytes) -> None
Set the protocol-specific message.
context
| @property
| context() -> EnvelopeContext
Get the envelope context.
skill_id
| @property
| skill_id() -> Optional[SkillId]
Get the skill id from an envelope context, if set.
Returns:
skill id
__eq__
| __eq__(other)
Compare with another object.
encode
| encode(serializer: Optional[EnvelopeSerializer] = None) -> bytes
Encode the envelope.
Arguments:
serializer: the serializer that implements the encoding procedure.
Returns:
the encoded envelope.
decode
| @classmethod
| decode(cls, envelope_bytes: bytes, serializer: Optional[EnvelopeSerializer] = None) -> "Envelope"
Decode the envelope.
Arguments:
envelope_bytes: the bytes to be decoded.serializer: the serializer that implements the decoding procedure.
Returns:
the decoded envelope.
__str__
| __str__()
Get the string representation of an envelope.
Multiplexer
class Multiplexer()
This class can handle multiple connections at once.
__init__
| __init__(connections: Sequence["Connection"], default_connection_index: int = 0, loop: Optional[AbstractEventLoop] = None)
Initialize the connection multiplexer.
Arguments:
connections: a sequence of connections.default_connection_index: the index of the connection to use as default. | this information is used for envelopes which | don't specify any routing context.loop: the event loop to run the multiplexer. If None, a new event loop is created.
in_queue
| @property
| in_queue() -> AsyncFriendlyQueue
Get the in queue.
out_queue
| @property
| out_queue() -> asyncio.Queue
Get the out queue.
connections
| @property
| connections() -> Tuple["Connection"]
Get the connections.
is_connected
| @property
| is_connected() -> bool
Check whether the multiplexer is processing envelopes.
default_routing
| @default_routing.setter
| default_routing(default_routing: Dict[PublicId, PublicId])
Set the default routing.
connection_status
| @property
| connection_status() -> ConnectionStatus
Get the connection status.
connect
| connect() -> None
Connect the multiplexer.
disconnect
| disconnect() -> None
Disconnect the multiplexer.
get
| get(block: bool = False, timeout: Optional[float] = None) -> Optional[Envelope]
Get an envelope within a timeout.
Arguments:
block: make the call blocking (ignore the timeout).timeout: the timeout to wait until an envelope is received.
Returns:
the envelope, or None if no envelope is available within a timeout.
async_get
| async async_get() -> Envelope
Get an envelope async way.
Returns:
the envelope
async_wait
| async async_wait() -> None
Get an envelope async way.
Returns:
the envelope
put
| put(envelope: Envelope) -> None
Schedule an envelope for sending it.
Notice that the output queue is an asyncio.Queue which uses an event loop running on a different thread than the one used in this function.
Arguments:
envelope: the envelope to be sent.
Returns:
None
InBox
class InBox()
A queue from where you can only consume envelopes.
__init__
| __init__(multiplexer: Multiplexer)
Initialize the inbox.
Arguments:
multiplexer: the multiplexer
empty
| empty() -> bool
Check for a envelope on the in queue.
Returns:
boolean indicating whether there is an envelope or not
get
| get(block: bool = False, timeout: Optional[float] = None) -> Envelope
Check for a envelope on the in queue.
Arguments:
block: make the call blocking (ignore the timeout).timeout: times out the block after timeout seconds.
Returns:
the envelope object.
Raises:
Empty: if the attempt to get an envelope fails.
get_nowait
| get_nowait() -> Optional[Envelope]
Check for a envelope on the in queue and wait for no time.
Returns:
the envelope object
async_get
| async async_get() -> Envelope
Check for a envelope on the in queue.
Returns:
the envelope object.
async_wait
| async async_wait() -> None
Check for a envelope on the in queue.
Returns:
the envelope object.
OutBox
class OutBox()
A queue from where you can only enqueue envelopes.
__init__
| __init__(multiplexer: Multiplexer)
Initialize the outbox.
Arguments:
multiplexer: the multiplexer
empty
| empty() -> bool
Check for a envelope on the in queue.
Returns:
boolean indicating whether there is an envelope or not
put
| put(envelope: Envelope) -> None
Put an envelope into the queue.
Arguments:
envelope: the envelope.
Returns:
None
put_message
| put_message(to: Address, sender: Address, protocol_id: ProtocolId, message: bytes) -> None
Put a message in the outbox.
This constructs an envelope with the input arguments.
Arguments:
to: the recipient of the envelope.sender: the sender of the envelope.protocol_id: the protocol id.message: the content of the message.
Returns:
None