Skip to content

academy.exchange.proxystore

ProxyStoreExchangeTransport 

ProxyStoreExchangeTransport(
    transport: ExchangeTransportT,
    store: Store[Any],
    should_proxy: Callable[[Any], bool],
    *,
    resolve_async: bool = False
)

Bases: ExchangeTransportMixin, NoPickleMixin, Generic[ExchangeTransportT]

ProxyStore exchange client bound to a specific mailbox.

Source code in academy/exchange/proxystore.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def __init__(
    self,
    transport: ExchangeTransportT,
    store: Store[Any],
    should_proxy: Callable[[Any], bool],
    *,
    resolve_async: bool = False,
) -> None:
    _assert_proxystore_available()
    self.transport = transport
    self.store = store
    self.should_proxy = should_proxy
    self.resolve_async = resolve_async
    register_store(store, exist_ok=True)

ProxyStoreExchangeFactory 

ProxyStoreExchangeFactory(
    base: ExchangeFactory[ExchangeTransportT],
    store: Store[Any] | None,
    should_proxy: Callable[[Any], bool],
    *,
    resolve_async: bool = False
)

Bases: ExchangeFactory[ProxyStoreExchangeTransport[ExchangeTransportT]]

ProxStore exchange client factory.

A ProxyStore exchange is used to wrap an underlying exchange so large objects may be passed by reference.

Parameters:

  • base (ExchangeFactory[ExchangeTransportT]) –

    Base exchange factory.

  • store (Store[Any] | None) –

    Store to use for proxying data.

  • should_proxy (Callable[[Any], bool]) –

    A callable that returns True if an object should be proxied. This is applied to every positional and keyword argument and result value.

  • resolve_async (bool, default: False ) –

    Resolve proxies asynchronously when received.

Source code in academy/exchange/proxystore.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
def __init__(
    self,
    base: ExchangeFactory[ExchangeTransportT],
    store: Store[Any] | None,
    should_proxy: Callable[[Any], bool],
    *,
    resolve_async: bool = False,
) -> None:
    _assert_proxystore_available()
    self.base = base
    self.store = store
    self.should_proxy = should_proxy
    self.resolve_async = resolve_async

create_agent_client async 

create_agent_client(
    registration: AgentRegistration[AgentT],
    request_handler: RequestHandler[RequestT_co],
) -> AgentExchangeClient[AgentT, ExchangeTransportT]

Create a new agent exchange client.

An agent must be registered with the exchange before an exchange client can be created. For example: 

factory = ExchangeFactory(...)
user_client = factory.create_user_client()
registration = user_client.register_agent(...)
agent_client = factory.create_agent_client(registration, ...)

Parameters:

  • registration (AgentRegistration[AgentT]) –

    Registration information returned by the exchange.

  • request_handler (RequestHandler[RequestT_co]) –

    Agent request message handler.

Returns:

Raises:

  • BadEntityIdError –

    If an agent with registration.agent_id is not already registered with the exchange.

Source code in academy/exchange/factory.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
async def create_agent_client(
    self,
    registration: AgentRegistration[AgentT],
    request_handler: RequestHandler[RequestT_co],
) -> AgentExchangeClient[AgentT, ExchangeTransportT]:
    """Create a new agent exchange client.

    An agent must be registered with the exchange before an exchange
    client can be created. For example:
    ```python
    factory = ExchangeFactory(...)
    user_client = factory.create_user_client()
    registration = user_client.register_agent(...)
    agent_client = factory.create_agent_client(registration, ...)
    ```

    Args:
        registration: Registration information returned by the exchange.
        request_handler: Agent request message handler.

    Returns:
        Agent exchange client.

    Raises:
        BadEntityIdError: If an agent with `registration.agent_id` is not
            already registered with the exchange.
    """
    agent_id: AgentId[AgentT] = registration.agent_id
    transport = await self._create_transport(
        mailbox_id=agent_id,
        registration=registration,
    )
    assert transport.mailbox_id == agent_id
    status = await transport.status(agent_id)
    if status != MailboxStatus.ACTIVE:
        await transport.close()
        raise BadEntityIdError(agent_id)
    return AgentExchangeClient(
        agent_id,
        transport,
        request_handler=request_handler,
    )

create_user_client async 

create_user_client(
    *, name: str | None = None, start_listener: bool = True
) -> UserExchangeClient[ExchangeTransportT]

Create a new user in the exchange and associated client.

Parameters:

  • name (str | None, default: None ) –

    Display name of the client on the exchange.

  • start_listener (bool, default: True ) –

    Start a message listener thread.

Returns:

Source code in academy/exchange/factory.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
async def create_user_client(
    self,
    *,
    name: str | None = None,
    start_listener: bool = True,
) -> UserExchangeClient[ExchangeTransportT]:
    """Create a new user in the exchange and associated client.

    Args:
        name: Display name of the client on the exchange.
        start_listener: Start a message listener thread.

    Returns:
        User exchange client.
    """
    transport = await self._create_transport(mailbox_id=None, name=name)
    user_id = transport.mailbox_id
    assert isinstance(user_id, UserId)
    return UserExchangeClient(
        user_id,
        transport,
        start_listener=start_listener,
    )