Heartbeat

Example Service that publishes data to all subscribed Clients.

This example shows how to add a task to the event loop of the Service.

Start the Broker

Start the Broker by running the following command.

msl-network start

Start the Heartbeat Service

Open another terminal and start the Heartbeat Service by running the following command.

python -c "from msl.examples.network import run_heartbeat; run_heartbeat()"

Expand to show the source code of the Heartbeat Service

"""Example Service that publishes data to all subscribed Clients.

This example also shows how to add a task to the event loop of the Service.
"""

import asyncio

from msl.network import Service


class Heartbeat(Service):
    """A Service that publishes a counter value."""

    _sleep: float = 1.0
    _counter: int = 0

    def reset(self) -> None:
        """Reset the heartbeat counter."""
        self._counter = 0

    def set_heart_rate(self, beats_per_second: int) -> None:
        """Change the rate that the value of the counter is published."""
        self._sleep = 1.0 / float(beats_per_second)

    async def emit(self) -> None:
        """This coroutine is also run in the event loop."""
        # Wait for the connection to the Broker to be established
        _ = await self.connected.wait()

        # Loop while connected to the Broker
        while self.connected.is_set():
            self.publish(self._counter)
            self._counter += 1
            await asyncio.sleep(self._sleep)


if __name__ == "__main__":
    heartbeat = Heartbeat()

    # Add a task to the event loop of the Service
    heartbeat.add_tasks(heartbeat.emit())

    # Connect the Service to the Broker (runs the event loop "forever")
    heartbeat.connect()

Run the Client

Connect to the Broker as a Client, link with the Heartbeat Service, subscribe to publications from the Service and also send requests to the Service.

from msl.network import Client


def heartbeat_handler(counter: int) -> None:
    """Receives published data from the Heartbeat Service."""
    print(f"Heartbeat {counter=} (Press ENTER to perform the next task)")


with Client() as client:
    link = client.link("Heartbeat")

    # Subscribe to publications with a callback function to receive published data
    link.subscribe(heartbeat_handler)

    # Wait until ENTER is pressed to reset the counter
    _ = input()
    link.reset()

    # Wait until ENTER is pressed to change the rate that data is published
    _ = input()
    link.set_heart_rate(10)

    # Wait until ENTER is pressed to end the script
    _ = input()