Pub/Sub Communication

For dashboards, services and notebooks to work seamlessly together, OrangeQS Juice provides a framework for publishing messages and logging them to the database.

This tutorial shows how to publish and subscribe to events and write/query them from the database usingorangeqs.juice.client.pubsub and orangeqs.juice.client.influxdb2.

In particular, we use publish_event() to publish an event and write it to InfluxDB. See protocol and database for the full specification. The tutorial will link to the full specification where relevant.

This tutorial assumes you have set up a Juice instance with a task manager service and a service where you can write code. For this tutorial, we suggest opening a Proxy Kernel in the default “shared” service to act as a publisher. We will create two clients, one for publishing events from a service and one for subscribing/querying these events from a personal notebook.

Step 1: Start the publisher

Start by opening a new Kernel in your service. Next, you can start publishing events from this notebook using the following code snippet:

import asyncio
import random

from typing import ClassVar, Annotated

from orangeqs.juice.client.influxdb2 import influxdb2_write_api
from orangeqs.juice.client.pubsub import publish_event, publisher_async
from orangeqs.juice.messaging import Event

class QuantityUpdate(Event):
    """Dummy event representing a value of a physical quantity."""

    measurement: ClassVar[str] = "dummy_quantity"

    def topic(self) -> str:
        """Topic of the quantity event, equal to `quantity`."""
        return self.quantity

    quantity: Annotated[str, "tag"]
    value: int | float
    unit: str | None

    @classmethod
    def generate_random(cls) -> "QuantityUpdate":
        """Generate a random `QuantityUpdate` event"""
        quantity, unit = random.choice([("voltage", "V"), ("current", "A"), ("power", "W")])
        value = round(random.uniform(0.0, 100.0), 2)
        return cls(quantity=quantity, value=value, unit=unit)

publisher = publisher_async()
write_api = influxdb2_write_api()
bucket = "shared"

while True:
    event = QuantityUpdate.generate_random()
    print(f"Publishing: {event.quantity} = {event.value} {event.unit}")

    await publish_event(
        event=event,
        publisher=publisher,
        write_api=write_api,
        bucket=bucket,
    )
    await asyncio.sleep(1)

This snippet creates a publisher with publisher_async() and uses publish_event() to publish each event. For the event type we will use QuantityUpdate as an example. Every second it prints the event with random data that was published. The event will be published for other services to receive. It will also be logged in the database. As there are no subscribers yet, the events will be silently discarded.

The first notebook is now done and we should leave it running in the background.

Step 2: Prepare the subscriber

Next, open a second notebook within the same Juice instance. Let’s create a subscriber inside that notebook that listens for QuantityUpdate events with the topic "voltage" or "current". In QuantityUpdate.topic() we can see that the topic of the event is set to the name of the quantity.

from typing import ClassVar, Annotated

from orangeqs.juice.client.pubsub import subscriber_async
from orangeqs.juice.messaging import Event

# Usually you would import this from a common location. For demonstration purposes
# we copy the class definition here.
class QuantityUpdate(Event):
    measurement: ClassVar[str] = "dummy_quantity"

    def topic(self) -> str:
        """Topic of the quantity event, equal to `quantity`."""
        return self.quantity

    quantity: Annotated[str, "tag"]
    value: int | float
    unit: str | None

subscriber = subscriber_async()
subscriber.subscribe(QuantityUpdate, topic="voltage")
subscriber.subscribe(QuantityUpdate, topic="current")

This will instantiate a SubscriberAsync that connects to the central proxy to receive events it is subscribed to. See SubscriberAsync.subscribe() to learn more about subscribing to specific event types and topics.

Step 3: Start listening for events

So far we have instantiated the subscriber, so let’s now access incoming events on the subscriber. The coroutine SubscriberAsync.get() can be used to retrieve messages from the subscriber.

while True:
    event = await subscriber.get()
    print(f"Received: {event.quantity} = {event.value} {event.unit}")

The messages that are published from the first notebook that match the subscriptions should be printed.

Discovering published topics

To subscribe to an event you need to know which topics are actually being published in your Juice instance. This is also useful, for example, when configuring the system parameter monitor, which monitors events by topic.

Every event flows through the central pub/sub proxy in the task manager. The task manager keeps an in-memory record of each unique topic it has seen, together with the timestamp it was last seen. A topic is recorded as EventName.subtopic (or just EventName for events without a topic), so the QuantityUpdate events above appear as QuantityUpdate.voltage, QuantityUpdate.current, and so on.

You can retrieve this record with the GetEventTopicList task:

from orangeqs.juice import Client
from orangeqs.juice.task_manager.tasks import GetEventTopicList

client = Client()
result = client.execute_blocking("task-manager", GetEventTopicList(), check=True)
print(result.result)
# {'QuantityUpdate.voltage': '2026-06-05T19:30:24.204446+00:00',
#  'QuantityUpdate.current': '2026-06-05T19:30:24.203630+00:00',
#  ...}

For quick manual inspection there is also a convenience helper that returns the same mapping directly:

from orangeqs.juice.client import topics

topics.get_topics_last_seen_timestamp()

The record is kept for the lifetime of the task manager process, so topics that were published before the task manager last (re)started will not appear until they are published again.