Client

A Client sends requests to a Service and receives replies. It can also subscribe to messages that are published by the Service.

from msl.network import Client


def captured(data: bytes) -> None:
    """Handle image data that is published by the Service."""
    ...


with Client(host="192.168.1.25") as client:  # (1)!
    print(client.services())  # (2)!

    camera = client.link("Camera")  # (3)!
    print(camera.signatures())  # (4)!

    camera.subscribe(captured)  # (5)!

    camera.set_resolution(width=1280, height=720)  # (6)!
    resolution = camera.get_resolution()  # (7)!

    # do more stuff...

1. Connect to a Broker. By using a context manager (i.e., a with statement), the connection is automatically closed after you are done using it. Otherwise, call disconnect.
2. Get the names of the Services that are available to link with.
3. Establish a Link with a Service. Tip: The Service does not actually need to be running in order to establish a Link. The Service only needs to be running when a request is sent in order to get a reply.
4. Request what functions are available from the linked Service. This will return a mapping of the function name to the input/output parameters that the function requires/returns.
5. Subscribe to captured images that are published by the Service.
6. Send a request. No reply expected, but the Service still sends None as a reply.
7. Send a request. Wait for reply.

Client

Client(
    *,
    host: str = "127.0.0.1",
    port: int = BROKER_PORT,
    flag: Flag = PICKLE,
    curve: AuthCurve | None = None,
    plain: AuthPlain | None = None,
    xpub_port: int | None = None
)

A Client.

Parameters:

Name	Type	Description	Default
host	str	The hostname (or IP address) that the Broker is running on.	'127.0.0.1'
port	int	The network port that the Broker is running on.	BROKER_PORT
flag	Flag	The serialisation and compression algorithms to apply to a request before sending the message.	PICKLE
curve	AuthCurve | None	The CURVE authentication to use.	None
plain	AuthPlain | None	The PLAIN authentication to use.	None
xpub_port	int | None	The port on the Broker that is publishing messages. Typically, this value is port + 1 and does not need to be specified.	None

Source code in src/msl/network/client.py

def __init__(  # noqa: PLR0913
    self,
    *,
    host: str = "127.0.0.1",
    port: int = BROKER_PORT,
    flag: Flag = Flag.PICKLE,
    curve: AuthCurve | None = None,
    plain: AuthPlain | None = None,
    xpub_port: int | None = None,
) -> None:
    """A Client.

    Args:
        host: The hostname (or IP address) that the [Broker][] is running on.
        port: The network port that the [Broker][] is running on.
        flag: The serialisation and compression algorithms to apply to a
            request before sending the message.
        curve: The [CURVE](https://rfc.zeromq.org/spec/25/) authentication to use.
        plain: The [PLAIN](https://rfc.zeromq.org/spec/24/) authentication to use.
        xpub_port: The port on the [Broker][] that is publishing messages.
            Typically, this value is `port + 1` and does not need to be specified.
    """
    self.flag: Flag = flag
    """The serialisation and compression algorithms to apply to a request before sending the message."""

    self._host_port: tuple[str, int] = (host, port)
    self._xpub_port: int = xpub_port or port + 1  # Link connects via SUBscribe: SUB -> XPUB
    self._id: str = os.urandom(8).hex()
    self._transaction: int = 0
    self._async_client: _AsyncClient | None = None
    self._is_connected: Event = Event()
    self._links: list[Link] = []
    self._curve: AuthCurve | None = curve
    self._plain: AuthPlain | None = plain

    if curve is not None and plain is not None:
        msg = "Cannot use both PLAIN and CURVE authentication, select only one authentication mechanism"
        raise ValueError(msg)

    self._thread: Thread = Thread(
        target=run_event_loop, daemon=True, args=(_create_async_client(self, curve, plain),)
    )
    self._thread.start()

    while self._async_client is None:
        continue

flag instance-attribute

flag: Flag

The serialisation and compression algorithms to apply to a request before sending the message.

is_connected property

is_connected: bool

bool — Whether the client is connected to the Broker.

disconnect

disconnect() -> None

Close the connection to the Broker.

Source code in src/msl/network/client.py

def disconnect(self) -> None:
    """Close the connection to the [Broker][]."""
    if self._async_client is None:
        return

    for link in self._links:
        link.unlink()

    self._async_client.disconnect()
    self._is_connected.clear()
    self._async_client = None
    self._thread.join()
    logger.debug("%s disconnected", self)

flag_at

flag_at(flag: Flag) -> Generator[None, None, None]

Use as a context manager to temporarily change the flag value.

Example

from msl.network import Client, Flag

client = Client(flag=Flag.JSON)  # (1)!
link = client.link("MyService")

# uses JSON to serialise the request
link.do_something()

with link.flag_at(Flag.ZSTD):  # (2)!
    # uses ZSTD to compress the image bytes in the request
    link.send_picture(open("image.jpg", "rb").read())

# uses JSON to serialise the request
link.do_something_else(n=100)

client.disconnect()

1. Use JSON to serialised each request (no compression).
2. Temporarily change the serialisation and compression algorithms to send the request.

Parameters:

Name	Type	Description	Default
flag	Flag	The temporary flag to use while within the context. Once the context exits, the value is reset to the original value.	required

Source code in src/msl/network/client.py

@contextmanager
def flag_at(self, flag: Flag) -> Generator[None, None, None]:
    """Use as a context manager to temporarily change the [flag][..flag] value.

    !!! example
        ```python hl_lines="9-11"
        from msl.network import Client, Flag

        client = Client(flag=Flag.JSON)  # (1)!
        link = client.link("MyService")

        # uses JSON to serialise the request
        link.do_something()

        with link.flag_at(Flag.ZSTD):  # (2)!
            # uses ZSTD to compress the image bytes in the request
            link.send_picture(open("image.jpg", "rb").read())

        # uses JSON to serialise the request
        link.do_something_else(n=100)

        client.disconnect()
        ```

        1. Use JSON to serialised each request (no compression).
        2. Temporarily change the serialisation and compression algorithms to send the request.

    Args:
        flag: The temporary flag to use while within the context. Once the
            context exits, the value is reset to the original value.
    """
    original = self.flag
    self.flag = flag
    try:
        yield
    finally:
        self.flag = original

link

link(service_name: str) -> Link

Link with a Service.

Parameters:

Name	Type	Description	Default
service_name	str	The name of a Service to create a Link with.	required

Returns:

Type	Description
Link	The Link instance.

Source code in src/msl/network/client.py

def link(self, service_name: str) -> Link:
    """Link with a [Service][].

    Args:
        service_name: The name of a [Service][] to create a [Link][msl.network.client.Link] with.

    Returns:
        The [Link][msl.network.client.Link] instance.
    """
    link = Link(
        client=self,
        curve=self._curve,
        host=self._host_port[0],
        plain=self._plain,
        service_name=service_name,
        xpub_port=self._xpub_port,
    )
    self._links.append(link)
    return link

services

services(timeout: float | None = None) -> list[str]

Request the names of the Services that are available from the Broker.

Parameters:

Name	Type	Description	Default
timeout	float | None	The maximum number of seconds to wait for the reply. If None, there is no limit on the wait time.	None

Returns:

Type	Description
list[str]	The names of the Services that are available to be linked with.

Source code in src/msl/network/client.py

def services(self, timeout: float | None = None) -> list[str]:
    """Request the names of the [Service][]s that are available from the [Broker][].

    Args:
        timeout: The maximum number of seconds to wait for the reply.
            If `None`, there is no limit on the wait time.

    Returns:
        The names of the [Service][]s that are available to be [link][..link]ed with.
    """
    return sorted(self._request("Broker", "SERVICES").result(timeout))