Skip to content

aea.connections.base

The base connection package.

ConnectionStates Objects

class ConnectionStates(Enum)

Connection states enum.

Connection Objects

class Connection(Component, ABC)

Abstract definition of a connection.

__init__

def __init__(configuration: ConnectionConfig,
             data_dir: str,
             identity: Optional[Identity] = None,
             crypto_store: Optional[CryptoStore] = None,
             restricted_to_protocols: Optional[Set[PublicId]] = None,
             excluded_protocols: Optional[Set[PublicId]] = None,
             **kwargs: Any) -> None

Initialize the connection.

The configuration must be specified if and only if the following parameters are None: connection_id, excluded_protocols or restricted_to_protocols.

Arguments:

  • configuration: the connection configuration.
  • data_dir: directory where to put local files.
  • identity: the identity object held by the agent.
  • crypto_store: the crypto store for encrypted communication.
  • restricted_to_protocols: the set of protocols ids of the only supported protocols for this connection.
  • excluded_protocols: the set of protocols ids that we want to exclude for this connection.
  • kwargs: keyword arguments passed to component base

loop

@property
def loop() -> asyncio.AbstractEventLoop

Get the event loop.

address

@property
def address() -> "Address"

Get the address.

crypto_store

@property
def crypto_store() -> CryptoStore

Get the crypto store.

has_crypto_store

@property
def has_crypto_store() -> bool

Check if the connection has the crypto store.

data_dir

@property
def data_dir() -> str

Get the data directory.

component_type

@property
def component_type() -> ComponentType

Get the component type.

configuration

@property
def configuration() -> ConnectionConfig

Get the connection configuration.

restricted_to_protocols

@property
def restricted_to_protocols() -> Set[PublicId]

Get the ids of the protocols this connection is restricted to.

excluded_protocols

@property
def excluded_protocols() -> Set[PublicId]

Get the ids of the excluded protocols for this connection.

state

@property
def state() -> ConnectionStates

Get the connection status.

state

@state.setter
def state(value: ConnectionStates) -> None

Set the connection status.

connect

@abstractmethod
async def connect() -> None

Set up the connection.

disconnect

@abstractmethod
async def disconnect() -> None

Tear down the connection.

send

@abstractmethod
async def send(envelope: "Envelope") -> None

Send an envelope.

Arguments:

  • envelope: the envelope to send.

Returns:

None

receive

@abstractmethod
async def receive(*args: Any, **kwargs: Any) -> Optional["Envelope"]

Receive an envelope.

Arguments:

  • args: positional arguments
  • kwargs: keyword arguments

Returns:

the received envelope, or None if an error occurred.

from_dir

@classmethod
def from_dir(cls, directory: str, identity: Identity,
             crypto_store: CryptoStore, data_dir: str,
             **kwargs: Any) -> "Connection"

Load the connection from a directory.

Arguments:

  • directory: the directory to the connection package.
  • identity: the identity object.
  • crypto_store: object to access the connection crypto objects.
  • data_dir: the assets directory.
  • kwargs: keyword arguments passed to connection base

Returns:

the connection object.

from_config

@classmethod
def from_config(cls, configuration: ConnectionConfig, identity: Identity,
                crypto_store: CryptoStore, data_dir: str,
                **kwargs: Any) -> "Connection"

Load a connection from a configuration.

Arguments:

  • configuration: the connection configuration.
  • identity: the identity object.
  • crypto_store: object to access the connection crypto objects.
  • data_dir: the directory of the AEA project data.
  • kwargs: keyword arguments passed to component base

Returns:

an instance of the concrete connection class.

is_connected

@property
def is_connected() -> bool

Return is connected state.

is_connecting

@property
def is_connecting() -> bool

Return is connecting state.

is_disconnected

@property
def is_disconnected() -> bool

Return is disconnected state.

BaseSyncConnection Objects

class BaseSyncConnection(Connection)

Base sync connection class to write connections with sync code.

__init__

def __init__(configuration: ConnectionConfig,
             data_dir: str,
             identity: Optional[Identity] = None,
             crypto_store: Optional[CryptoStore] = None,
             restricted_to_protocols: Optional[Set[PublicId]] = None,
             excluded_protocols: Optional[Set[PublicId]] = None,
             **kwargs: Any) -> None

Initialize the connection.

The configuration must be specified if and only if the following parameters are None: connection_id, excluded_protocols or restricted_to_protocols.

Arguments:

  • configuration: the connection configuration.
  • data_dir: directory where to put local files.
  • identity: the identity object held by the agent.
  • crypto_store: the crypto store for encrypted communication.
  • restricted_to_protocols: the set of protocols ids of the only supported protocols for this connection.
  • excluded_protocols: the set of protocols ids that we want to exclude for this connection.
  • kwargs: keyword arguments passed to connection base

put_envelope

def put_envelope(envelope: Optional["Envelope"]) -> None

Put envelope in to the incoming queue.

connect

async def connect() -> None

Connect connection.

disconnect

async def disconnect() -> None

Disconnect connection.

send

async def send(envelope: "Envelope") -> None

Send envelope to connection.

receive

async def receive(*args: Any, **kwargs: Any) -> Optional["Envelope"]

Get an envelope from the connection.

start_main

def start_main() -> None

Start main function of the connection.

main

def main() -> None

Run main body of the connection in dedicated thread.

on_connect

@abstractmethod
def on_connect() -> None

Run on connect method called.

on_disconnect

@abstractmethod
def on_disconnect() -> None

Run on disconnect method called.

on_send

@abstractmethod
def on_send(envelope: "Envelope") -> None

Run on send method called.