Streaming and object contexts¶

This page covers the C++ streaming model: objects as stream items, pull consumers that read a completed collection, and push consumers that drain a bounded live queue. It also distinguishes that model from byte-oriented DataSource/DataSink I/O and from object conversion with DataBuffer.

Choose the right primitive¶

Need	Use	What crosses the boundary
Own and organize application objects	`ObjectsContext`	Object identity and parent/child membership.
Process objects one at a time	`IStreamProducer` and `StreamInput`	`Object*` values.
Decouple a producer and consumer, cap memory, or trigger live work	`DataBuffer` as a stream producer	A bounded queue of `Object*`.
Read bytes from a file, memory, or an embedded scene payload	`DataSource`	`std::istream` and decoded object/frame data.
Write bytes to a file, memory, or an embedded scene payload	`DataSink`	`std::ostream` and encoded object/frame data.
Re-encode an object between registered codecs	`DataBuffer` as a codec node	An encoded `uprops::DataFrame`.

DataBuffer deliberately participates in both rows: it derives from DataSource to be a codec-node output, and it also implements the stream producer API to be a bounded object queue. Select its behavior by how it is wired and read; do not assume an encoded byte payload is a stream of individual objects.

Object lifetime comes first¶

The context owns the objects it contains. A stream normally borrows pointers to those objects, so the context must outlive both the producer and consumer. For a temporary value that is not part of a scene, give its owning scope an equally clear lifetime.

#include <Core/ObjectsContext.h>
#include <Core/Stream.h>
#include <Math/ContainerBox.h>

uLib::ObjectsContext context;
auto *first = context.CreateObject<uLib::ContainerBox>("first");
auto *second = context.CreateObject<uLib::ContainerBox>("second");

uLib::GenericStreamProducer completed;
completed.AddItem(first);
completed.AddItem(second);

GenericStreamProducer is the small in-memory pull producer. It does not take ownership of the pointers it holds.

Pull: read a completed collection on demand¶

Use pull when all items exist before the consumer begins. The consumer controls when it runs and reads from the start of the collection. It is the simplest and most reproducible form of a stream.

uLib::StreamInput input;
input.Connect(&completed);
input.SetLimit(0);  // zero means no limit

while (uLib::Object *item = input.Read()) {
  auto *box = dynamic_cast<uLib::ContainerBox *>(item);
  if (!box) continue;
  // Process box.
}

StreamInput is the preferred consumer API. It derives whether the producer is pull or push at Connect time, handles cursor reset/waiting rules, and lets a consumer use the same loop for a collection, a queue, or a live run.

For recipes, the shipped C++ helper builds the equivalent port connection:

#include <Core/RecipeStreamChain.h>

uLib::ObjectsContext scene;
const auto chain = uLib::BuildRecipeStreamChain(
    scene, uLib::RecipeStreamChainKind::Pull);

// The scene now contains:
// chain.producer.Out --(connected)--> chain.consumer.In
// Run the producer, then explicitly run the consumer.

The helper uses ReferencePropertyBase::ConnectFrom, not a one-time assignment. The input therefore follows future values placed on the output port and the wire is saved in a SceneDocument.

Push: producer and consumer with a bounded buffer¶

Use push when items arrive during production or the producer and consumer should run independently. A DataBuffer holds at most Capacity items. Its overflow policy determines whether a full queue blocks, discards the newest item, or discards the oldest item.

upstream producer ──> DataBuffer ──> StreamInput consumer
                       capacity N       drains new items
                       NotifyEvery K

The lowest-level producer calls PushItem and closes the stream when it is done:

#include <chrono>
#include <Core/DataBuffer.h>

uLib::DataBuffer buffer{"live-items"};
buffer.Capacity.Set(32);
buffer.OverflowPolicy.Set("Block");
buffer.NotifyEvery.Set(8);

for (uLib::Object *item : producedItems) {
  if (!buffer.PushItem(item, std::chrono::milliseconds{5000})) {
    // The queue remained full, was closed, or the item was invalid.
    break;
  }
}
buffer.CloseStream();

The consumer does not need to know the queue implementation:

uLib::StreamInput input;
input.Connect(&buffer);
input.SetTimeout(std::chrono::milliseconds{200});
input.SetInterrupt([&] {
  return !stopRequested;
});

while (uLib::Object *item = input.Read()) {
  // Process item. Read() waits briefly when a live source is temporarily empty.
}

Read returns null only when the producer is genuinely finished, the configured limit is reached, or the interrupt requests stopping. An empty queue alone is not the end of a push stream.

Async relay: stream-to-stream¶

When an existing IStreamProducer is the upstream side, point the buffer at it and set StreamMode to Async. The buffer starts a worker that relays items into its bounded queue; back-pressure comes from Capacity.

uLib::GenericStreamProducer upstream;
// Populate upstream from context-owned objects.

uLib::DataBuffer buffer{"relay"};
buffer.SetInput(&upstream);
buffer.StreamMode.Set("Async");
buffer.Capacity.Set(32);
buffer.NotifyEvery.Set(8);
buffer.StartWorker();

uLib::StreamInput input;
input.Connect(&buffer);
while (uLib::Object *item = input.Read()) {
  // Items become available while the worker relays them.
}

Inline streams directly from the upstream producer, without a thread or queue. Auto pulls directly unless an asynchronous worker is already appropriate. Use Async when pacing and back-pressure are the point.

Notification-driven consumer: drain, do not drive¶

An algorithm with AutoRun can be scheduled by buffer updates. Set NotifyEvery to a positive batch size, wire the buffer to an input port, and enable AutoRun on the consumer. When that consumer runs because of a buffer notification, it should read what is already queued and return:

uLib::StreamInput input;
input.Connect(&buffer);
input.SetDrainOnly(true);

while (uLib::Object *item = input.Read()) {
  // Fold in this notified batch.
}

SetDrainOnly(true) is important in this case: waiting for more can block the producer that triggered the consumer, and asking an activatable source for more can initiate nested production. Leave it false for a manual consumer whose job is to drive a live source to completion.

The ready-made recipe scene shows the full wiring:

#include <Core/RecipeStreamChain.h>

uLib::ObjectsContext scene;
const auto chain = uLib::BuildRecipeStreamChain(
    scene, uLib::RecipeStreamChainKind::Push);

// producer.Out -> buffer.Input
// consumer.In refers to buffer
// buffer: Async, capacity 32, NotifyEvery 8
// consumer: AutoRun true

The sources are src/Core/examples/01_recipes_pull.cpp and src/Core/examples/02_recipes_push.cpp. They create the scene files used by gcompose, so the code, node graph, and examples share the same construction helper.

Live muon reconstruction¶

MuonScatterSource applies the same push shape to detector data. A Geant scene or other producer sends scattering events to its IMuonScatterSink interface, then a MuonTomographyAlgorithm reads the source through its Stream port.

#include <HEP/Detectors/MuonScatterSource.h>
#include <HEP/MuonTomography/MuonTomographyImaging.h>

uLib::MuonScatterSource source{"scatters"};
source.Capacity.Set(1024);
source.BatchSize.Set(64);
source.NotifyEvery.Set(64);

uLib::PocaImageAlgorithm reconstruction;
reconstruction.SetStream(&source);
reconstruction.DriveStream.Set(false);  // this run is notification-driven
reconstruction.ResetImage.Set(false);   // retain earlier batches

For a manually started reconstruction, leave DriveStream true so that an idle, activatable producer can be started and waited on. For a notification-driven reconstruction, set it false so the run only drains the batch that caused it.

Data sources, sinks, and embedded payloads¶

DataSource and DataSink are byte-payload endpoints, not just file path holders. Both can use a disk file, memory, or an embedded payload that travels when a scene is serialized.

#include <Core/DataSource.h>
#include <Core/DataSink.h>

uLib::DataSource source{"input"};
source.SetFilePath("measurement.vti");
source.SetFormat("vti");
if (!source.LoadFile()) {
  // Handle unreadable input.
}
source.Embed(true);  // save the payload inside a scene document

uLib::DataSink sink{"output"};
sink.SetFilePath("result.vti");
sink.SetFormat("vti");
sink.SetEmbedded(false);

{
  auto out = sink.OpenStream();
  // Write a format-valid payload to *out.
}  // destruction commits the buffered payload and writes the file

For object codecs, wire a source's Output or a sink's Input to the domain object and call Publish or Capture. A DataBuffer can sit between them to re-encode through a registered backend:

uLib::DataBuffer conversion{"conversion"};
conversion.SetBackend("native-bin");  // or another registered backend
conversion.SetInput(domainObject);
conversion.Refresh();

uLib::DataSink sink{"encoded"};
sink.WriteSource(&conversion);

Use DataSource::EmbedAll / UnpackAll and DataSink::EmbedAll / FlushAll for an entire ObjectsContext.

Failure and shutdown rules¶

Always set a positive Capacity and choose an OverflowPolicy deliberately. Block gives back-pressure; dropping policies make loss explicit.
Always close a manually pushed stream with CloseStream; a waiting reader otherwise cannot distinguish an idle producer from one that has finished.
Supply an interrupt and finite timeout for consumers that may be stopped.
Keep items owned by a context or another longer-lived owner for as long as the stream can expose them.
Use GetQueuedCount, GetStatus, GetDroppedCount (on MuonScatterSource), and the algorithm's processed-count/status APIs to diagnose pacing before increasing capacity.

See the Core guide for properties and port wiring, and the class catalog for all related public types.