Measuring number of in-flight data in unbounded Stream consuming

[Jernej Jerin]

Is there a way on how to measure number of real time in-flight data in unbounded Stream consuming? I am interested in Broadcaster as well as Processor version. In the end I would like to have a dataset of key:value pairs, where key would represent timestamp when the measurement was taken and value would represent the number of in-flight data at that particular point. I need this so I could draw some graphs in Excel.

[Stephane Maldini]

Streams have the .timestamp() and .elapsed() operations which produce Tuple2<Long, T> where Long is the timestamp or the delta between 2 onNext. You can then play with reduce().

In this test I actually reduce the metrics produced by elapsed to compute a moving time average: 

https://github.com/reactor/reactor/blob/master/reactor-stream/src/test/groovy/reactor/rx/StreamsSpec.groovy#L2448

2.1 will benefit from a more generic MetricsProcessor for these scenarios, we have it in the backlog once we're done migrating reactor-core into its new focused job: toolkit for creating basic Processors (including the current ones), Subscribers, and Publisher. Reactor-Stream being the module using these blocks to produce high order operations. 

On Tue, Jul 14, 2015 at 11:33 AM, Jernej Jerin <jernej...@gmail.com> wrote:

Is there a way on how to measure number of real time in-flight data in unbounded Stream consuming? I am interested in Broadcaster as well as Processor version. In the end I would like to have a dataset of key:value pairs, where key would represent timestamp when the measurement was taken and value would represent the number of in-flight data at that particular point. I need this so I could draw some graphs in Excel.

--
You received this message because you are subscribed to the Google Groups "reactor-framework" group.
To unsubscribe from this group and stop receiving emails from it, send an email to reactor-framew...@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

--

Stephane Maldini | Solutions Architect, CSO EMEA | London | Pivotal

W: pivotal.io | T: @smaldini

[Jernej Jerin]

I think I did not ask appropriate question. I have read again the section about the Backpressure and Overflow and the following is what I really want to measure. Is there a way to measure in-memory overflow buffer size? The documentation states the following:

Reactor will automatically provide for an in-memory overflow buffer when the dispatcher or the capacity differs from one action to another. This will not apply to Core Processors, which handle the overflow in their own way. Dispatchers can be re-used and Reactor must limit the number of dispatches where it can, hence the in-memory buffer added by Action when dispatchers differ.

So if I add explicitly the call onOverflowBuffer(), where can I measure its size during the program execution? I am also using Java Mission Control and Flight Recorder, if this helps anything.