From Event Pipelining to Stream Processing

Subscriber

The Substrates API is an example of an Event Driven Architecture (EDA) applied to Observability. Instruments emit a data value into an Inlet part of a Conduit connected to a Circuit. Suppose there are Subscribers registered against the Source associated with the Conduit. In that case, the data value becomes the emitted value, and is dispatched to all Outlets registered by a Subscriber for the Subject (Instrument). Below is an example of boilerplate code for event pipeline processing.

var circuit =

cortex.circuit (

"counters"

);

var conduit =

circuit.conduit (

Counters.counters ()

);

var subscription =

conduit

.source ()

.subscribe ( /* ... */ );

With an event-driven architectural approach as above, anytime a value needs to be calculated from a series of Events, a stateful consumer, typically an Outlet, invariably must use another Circuit Component, such as a Conduit or Adjunct, to publish the derived result and continuously after each Source Event receipt and processing. However, there is an alternative option in the Source interface.

Reservoir

Instead of registering a Subscriber with a Source, the reservoir method can be called on a Source to create a Reservoir that temporarily stores Captures in an order-maintaining buffer between calls to the drain method, which, when called, returns a Stream of Captures, consisting of the Captures published since the last call to the drain method or the creation of the Reservoir instance.

var reservoir =

conduit

.source ()

.reservoir ();

var stream =

reservoir.drain ();

Because the Circuit turns an Instrument’s data emission into a Capture, it is typical to call await on the Circuit’s Current before draining a Reservoir. For each call to the drain method, the queued-up Captures in the buffer are removed and returned as a Stream, with the buffer continuing to fill up after the drain method is executed. Below is an example of such Substrates API usage.

var counter =

counters.instrument (

"name"

);

for ( int i = 10; i > 0; i-- )

counter.inc ();

circuit

.current ()

.await ();

var stream =

reservoir.drain ();

Stream

Now that the buffered Captures are represented as a Stream, standard functional library operations provided by the Java SDK, such as grouping and summing, can be easily employed, as shown in the following code. Much of the unit test code uses Reservoirs as such.

stream.collect (

Collectors.groupingBy (

event -> event

.subject ()

.reference ()

.name (),

Collectors.summingLong (

Capture::emission

)

).forEach (

( name, total ) ->

System.out.printf (

"%s : %d%n",

name,

total

)

);

Running the above code, the following is printed to a terminal.

>_
name : 10