Ohara Lab

gRPC Frame: EventBus and Event Subscriptions

gRPC development guide / gRPC Quick Start / StreamCall and image transfer

Architecture

The gRPC Frame combines an in-process EventBus with the Events/Subscribe gRPC service. StreamCall returns values produced by an invoked streaming method; Subscribe receives notifications published to a named topic.

Server process                                     Client
Event source → EventBus.publish() → Events/Subscribe → GrpcClient.subscribe()

create_grpc_server(event_bus=...) accepts an existing bus and creates one when omitted. The server passes the bus to the Events service. If ctrl_obj exposes _set_event_bus, the server calls it with the bus.

Publishing an event

from grpc_frame.event_bus import EventBus
from grpc_frame.grpc_server import create_grpc_server

bus = EventBus()
server = create_grpc_server(ctrl_obj=ctrl, event_bus=bus)

# Called by server-side integration code when an event occurs:
bus.publish("measurement_ready", {"value": 12.3}, source="sensor")

The publish() call illustrates the server-side interface; actual device integrations invoke it from their event-producing code. DeviceClass does not need gRPC-specific types or subscription logic. EventBus.subscribe() is an internal iterator with per-subscriber queues and a condition variable, not the remote client API. The bus does not store events published before subscription.

Subscribing from a client

from grpc_frame.grpc_client import GrpcClient

client = GrpcClient(server_ip="127.0.0.1", server_port=50051)
try:
    for payload in client.subscribe(topic="measurement_ready", once=True):
        print(payload)
finally:
    client.close()

subscribe(topic=..., once=False, source="") yields deserialized payloads, not full event envelopes. once=True ends the client subscription after one event; it does not disable event generation on the instrument. The source parameter is sent in the request, but the supplied EventBus implementation does not apply source filtering.

A&D reader: threshold-triggered events

The aandd_reader package combines instrument-specific threshold evaluation with gRPC Frame event subscription. Its AanddReaderGrpcClient.prepare_event_measurement() wraps these remote DeviceClass calls:

self.stream_start()
self.event_set_config(threshold_g=threshold_g, above=above, topic=topic)
self.set_trigger_mode(
    mode=trigger_mode,
    window=trigger_window,
    min_samples=trigger_min_samples,
    alpha=trigger_alpha,
    stable_only=trigger_stable_only,
)
self.event_arm(once=once)

Here stream_start() starts the aandd_reader measurement stream, not gRPC StreamCall. event_set_config, set_trigger_mode, and event_arm are device-specific methods. The once argument of event_arm controls device-side arming, independently of the once argument of subscribe.

from aandd_reader import Config
from aandd_reader.server_grpc import AanddReaderGrpcClient

Config.load_yaml("config_server1.yaml")
client = AanddReaderGrpcClient(config=Config)
try:
    client.prepare_event_measurement(
        topic="weight_threshold",
        threshold_g=10.0,
        above=True,
        once=True,
        trigger_mode="raw",
    )
    print(client.wait_event())
finally:
    client.stream_stop()
    client.close()

wait_event() calls subscribe(topic=the_prepared_topic, once=True) and returns the first event. Trigger modes include raw, mean, median, and ema; the dedicated client also accepts window, minimum-sample, EMA-alpha, and stable-only options. Subscription timing should be designed with the instrument’s event-generation behavior in mind.

Source implementations

Further gRPC implementation examples