Streaming Adapters

Overview

Most adapters are event-driven: an InputModule (e.g. file_watcher, mqtt_watcher) detects a discrete event, and the interpreter’s measurement() is invoked once per event to parse and return a single result. This works well when events are infrequent or when each call is cheap.

Some equipment doesn’t fit that model - a source that changes too fast, or too continuously, for a per-event callback to keep up (e.g. a rapidly-appended log file, where each modification event forces a full re-read). For these cases, StreamingAdapter lets the interpreter own a persistent read loop instead: the interpreter implements stream(filename), a generator that is started once per run and yields measurements at its own pace for as long as it runs - decoupling measurement throughput entirely from any input event rate.

Use StreamingAdapter only when a per-event adapter (ContinuousExperimentAdapter, UploadAdapter, etc.) genuinely can’t keep up. It’s a different contract, not a drop-in replacement - see Requirements below.

---

Requirements

1. Interpreter: Must implement stream(self, filename) -> Iterator[Any], in addition to (not instead of) the usual measurement(). StreamingAdapter checks for stream() at construction time and raises AdapterBuildError if it’s missing.
2. stream() yields finished measurements: Each yielded item should already be in the standard structure (dict, InfluxPoint, etc.) that measurement() would normally return.
3. The watcher’s start event supplies the filename: stream(filename) isn’t given a fixed target at construction time - it’s called with whatever data the watcher’s start event carries. With FileWatcher, that means constructing it with return_data=False, so on_created passes the created filepath rather than reading its content. A matching stop event (e.g. on_deleted) halts the stream.

---

1. Implement stream(filename) on the interpreter

from leaf.adapters.equipment_adapter import AbstractInterpreter

class CustomStreamingInterpreter(AbstractInterpreter):
    def measurement(self, data):
        # Still required by AbstractInterpreter, even if StreamingAdapter
        # never calls it directly.
        return super().measurement(data)

    def stream(self, filename):
        with open(filename) as f:
            while True:
                line = f.readline()
                if not line:
                    time.sleep(0.1)
                    continue
                yield self._parse(line)

Note there’s no filepath passed into __init__ here - unlike a per-event interpreter, this one doesn’t know what to read until stream() is actually called with it.

Track your own resume point

If stream() ends (see Restart behaviour below) and is restarted for the same filename, it starts from scratch - there is no automatic resume. If your source shouldn’t be re-read from the beginning, persist the read position (e.g. a file offset) on self, and have stream() pick up from there each time it’s (re)started.

2. Construct the adapter

from leaf.adapters.core_adapters.streaming_adapter import StreamingAdapter
from leaf.modules.input_modules.file_watcher import FileWatcher
from leaf.register.metadata import MetadataManager

class CustomStreamingAdapter(StreamingAdapter):
    def __init__(self, instance_data, output, watch_dir, error_holder=None,
                 maximum_message_size=1, metadata_manager=None, experiment_timeout=None,
                 external_watcher=None):
        metadata_manager = metadata_manager or MetadataManager()
        # return_data=False: on_created passes the filepath, not file content -
        # that filepath becomes the argument to interpreter.stream().
        watcher = FileWatcher(watch_dir, metadata_manager, error_holder=error_holder, return_data=False)
        interpreter = CustomStreamingInterpreter(error_holder=error_holder)

        super().__init__(
            instance_data,
            watcher,
            output,
            interpreter,
            maximum_message_size=maximum_message_size,
            error_holder=error_holder,
            metadata_manager=metadata_manager,
            experiment_timeout=experiment_timeout,
            external_watcher=external_watcher,
        )

maximum_message_size behaves the same way it does for MeasurePhase.update(): yielded items are buffered and transmitted in chunks of that size, rather than one message per item.

---

How it works

• StreamingAdapter does not launch stream() when the adapter starts. It waits for the watcher to dispatch a real start event (e.g. FileWatcher.on_created), reads the filename out of that event’s data, and launches stream(filename) on a dedicated background thread from there.
• If another start event arrives while a stream is already running (e.g. a new file appears), the current stream is signalled to stop and a new one is launched for the new filename - the newer run supersedes the older one.
• A matching stop event (e.g. on_deleted) halts the current stream without starting a new one.
• Each chunk of yielded items is transmitted directly to the OutputModule from the background thread, as soon as it’s ready - it does not wait for stream() to finish.
• A reentrancy lock shared with update() ensures the interpreter is never entered twice concurrently. If a watcher event happened to also target the measurement topic (e.g. on_modified on the same file stream() is already tailing), it would be dropped (logged as a warning) rather than racing with the stream - see Interpreter Guidelines for why that’s safe: stream() doesn’t depend on those notifications at all.

Restart behaviour and backoff

stream() is expected to run indefinitely. If it ever ends - by returning normally, or by raising an exception - StreamingAdapter treats that as unexpected and restarts it automatically, with exponential backoff:

• First retry after 1s, doubling each subsequent failure (2s, 4s, 8s, …), capped at 60s.
• If a run stayed up at least 60s before ending, the backoff resets to 1s for the next incident - so one isolated failure after a long healthy run isn’t penalised with an already-escalated delay.

No retry limit

There is no cap on the number of restart attempts. If stream() keeps failing, StreamingAdapter retries forever at the 60s ceiling - it will never give up on its own. If you need a hard stop after N failures, that logic must be added in your interpreter or adapter subclass.

A blocked generator can’t be interrupted

Stopping a stream - via stop(), a real stop event, or a newer start event superseding it - sets a flag that stops the loop from restarting, and waits up to 5s for the current call to exit. It can’t forcibly interrupt stream() if it’s currently blocked inside a system call (e.g. a blocking socket read) rather than suspended at a yield. The background thread is a daemon thread, so it won’t prevent process shutdown, but it may keep running past that 5s wait until the blocking call returns on its own - and if a newer stream was already launched, both may briefly overlap (the reentrancy lock just ensures they don’t run concurrently; the older one’s chunks are simply dropped as “busy” until it exits).

Every restart and shutdown is logged: a warning when a stream ends and is about to retry (with elapsed uptime and the next delay), and an info line when the thread exits for good.

---

Related: generators from measurement()

If you’re using a regular per-event adapter and only need a single measurement() call to produce many results without building the whole result in memory up front, see Returning a generator from measurement() - that’s a lighter-weight option that doesn’t require StreamingAdapter at all. It’s a different tool though: it still returns from measurement() once all chunks are built, so it doesn’t help when the event rate itself is the bottleneck - only stream() does that.