Repository files navigation

The purpose of Reactive Streams is to provide a standard for asynchronous stream processing with non-blocking backpressure.

The latest release isavailable on NuGet.

To install Reactive Streams, run the following command in the Package Manager Console

PM> Install-Package Reactive.Streams

Handling streams of data—especially “live” data whose volume is not predetermined—requires special care in an asynchronous system. The most prominent issue is that resource consumption needs to be carefully controlled such that a fast data source does not overwhelm the stream destination. Asynchrony is needed in order to enable the parallel use of computing resources, on collaborating network hosts or multiple CPU cores within a single machine.

The main goal of Reactive Streams is to govern the exchange of stream data across an asynchronous boundary – think passing elements on to another thread or thread-pool — while ensuring that the receiving side is not forced to buffer arbitrary amounts of data. In other words, backpressure is an integral part of this model in order to allow the queues which mediate between threads to be bounded. The benefits of asynchronous processing would be negated if the communication of backpressure were synchronous (see also theReactive Manifesto), therefore care has been taken to mandate fully non-blocking and asynchronous behavior of all aspects of a Reactive Streams implementation.

It is the intention of this specification to allow the creation of many conforming implementations, which by virtue of abiding by the rules will be able to interoperate smoothly, preserving the aforementioned benefits and characteristics across the whole processing graph of a stream application.

It should be noted that the precise nature of stream manipulations (transformation, splitting, merging, etc.) is not covered by this specification. Reactive Streams are only concerned with mediating the stream of data between differentAPI Components. In their development care has been taken to ensure that all basic ways of combining streams can be expressed.

In summary, Reactive Streams .NET is a standard and specification for Stream-oriented libraries for .NET that

• process a potentially unbounded number of elements
• in sequence,
• asynchronously passing elements between components,
• with mandatory non-blocking backpressure.

The Reactive Streams specification consists of the following parts:

The API specifies the types to implement Reactive Streams and achieve interoperability between different implementations.

The Technology Compatibility Kit (TCK) is a standard test suite for conformance testing of implementations.

Implementations are free to implement additional features not covered by the specification as long as they conform to the API requirements and pass the tests in the TCK.

The API consists of the following components that are required to be provided by Reactive Stream implementations:

1. IPublisher
2. ISubscriber
3. ISubscription
4. IProcessor

AnIPublisher is a provider of a potentially unbounded number of sequenced elements, publishing them according to the demand received from its ISubscriber(s).

In response to a call toIPublisher.Subscribe(ISubscriber) the possible invocation sequences for methods on theISubscriber are given by the following protocol:

onSubscribe onNext* (onError | onComplete)?

This means thatonSubscribe is always signalled,followed by a possibly unbounded number ofonNext signals (as requested byISubscriber) followed by anonError signal if there is a failure, or anonComplete signal when no more elements are available—all as long as theISubscription is not cancelled.

• The specifications below use binding words in capital letters fromhttps://www.ietf.org/rfc/rfc2119.txt
• The termsemit,signal orsend are interchangeable. The specifications below will usesignal.
• The termssynchronously orsynchronous refer to executing in the callingThread.
• The term "return normally" means "only throws exceptions that are explicitly allowed by the rule".

publicinterfaceIPublisher<outT>{voidSubscribe(ISubscriber<T>subscriber);}

[1] : A stateful IPublisher can be overwhelmed, bounded by a finite number of underlying resources, exhausted, shut-down or in a failed state.

publicinterfaceISubscriber<inT>{publicvoidOnSubscribe(ISubscriptionsubscription);publicvoidOnNext(Telement);publicvoidOnError(Exceptioncause);publicvoidOnComplete();}

[1] : See JMM definition of Happen-Before in section 17.4.5. onhttp://docs.oracle.com/javase/specs/jls/se7/html/jls-17.html

publicinterfaceISubscription{publicvoidRequest(longn);publicvoidCancel();}

[1] : An example for undesirable synchronous, open recursion would beISubscriber.OnNext ->ISubscription.Request ->ISubscriber.OnNext -> …, as it very quickly would result in blowing the calling Thread´s stack.

[2] : Avoid heavy computations and other things that would stall the caller´s thread of execution

[3] : As it is not feasibly reachable with current or foreseen hardware within a reasonable amount of time (1 element per nanosecond would take 292 years) to fulfill a demand of 2^63-1, it is allowed for anIPublisher to stop tracking demand beyond this point.

AnISubscription is shared by exactly oneIPublisher and oneISubscriber for the purpose of mediating the data exchange between this pair. This is the reason why theSubscribe() method does not return the createdISubscription, but instead returnsvoid; theISubscription is only passed to theISubscriber via theOnSubscribe callback.

publicinterfaceIProcessor<inT1,outT2>:ISubscriber<T1>,IPublisher<T2>{}

While not mandated, it can be a good idea to cancel anIProcessors upstreamISubscription when/if its lastISubscriber cancels theirISubscription,to let the cancellation signal propagate upstream.

The Reactive Streams API prescribes that all processing of elements (onNext) or termination signals (onError,onComplete) MUST NOTblock theIPublisher. However, each of theOn* handlers can process the events synchronously or asynchronously.

Take this example:

nioSelectorThreadOrigin map(f) filter(p) consumeTo(toNioSelectorOutput)

It has an async origin and an async destination. Let's assume that both origin and destination are selector event loops. TheISubscription.Request(n) must be chained from the destination to the origin. This is now where each implementation can choose how to do this.

The following uses the pipe| character to signal async boundaries (queue and schedule) andR# to represent resources (possibly threads).

nioSelectorThreadOrigin | map(f) | filter(p) | consumeTo(toNioSelectorOutput)-------------- R1 ----  | - R2 - | -- R3 --- | ---------- R4 ----------------

In this example each of the 3 consumers,map,filter andconsumeTo asynchronously schedule the work. It could be on the same event loop (trampoline), separate threads, whatever.

nioSelectorThreadOrigin map(f) filter(p) | consumeTo(toNioSelectorOutput)------------------- R1 ----------------- | ---------- R2 ----------------

Here it is only the final step that asynchronously schedules, by adding work to the NioSelectorOutput event loop. Themap andfilter steps are synchronously performed on the origin thread.

Or another implementation could fuse the operations to the final consumer:

nioSelectorThreadOrigin | map(f) filter(p) consumeTo(toNioSelectorOutput)--------- R1 ---------- | ------------------ R2 -------------------------

All of these variants are "asynchronous streams". They all have their place and each has different tradeoffs including performance and implementation complexity.

The Reactive Streams contract allows implementations the flexibility to manage resources and scheduling and mix asynchronous and synchronous processing within the bounds of a non-blocking, asynchronous, dynamic push-pull stream.

In order to allow fully asynchronous implementations of all participating API elements—IPublisher/ISubscription/ISubscriber/IProcessor—all methods defined by these interfaces returnvoid.

One of the underlying design principles is that all buffer sizes are to be bounded and these bounds must beknown andcontrolled by the subscribers. These bounds are expressed in terms ofelement count (which in turn translates to the invocation count of onNext). Any implementation that aims to support infinite streams (especially high output rate streams) needs to enforce bounds all along the way to avoid out-of-memory errors and constrain resource usage in general.

Since back-pressure is mandatory the use of unbounded buffers can be avoided. In general, the only time when a queue might grow without bounds is when the publisher side maintains a higher rate than the subscriber for an extended period of time, but this scenario is handled by backpressure instead.

Queue bounds can be controlled by a subscriber signaling demand for the appropriate number of elements. At any point in time the subscriber knows:

• the total number of elements requested:P
• the number of elements that have been processed:N

Then the maximum number of elements that may arrive—until more demand is signaled to the IPublisher—isP - N. In the case that the subscriber also knows the number of elements B in its input buffer then this bound can be refined toP - B - N.

These bounds must be respected by a publisher independent of whether the source it represents can be backpressured or not. In the case of sources whose production rate cannot be influenced—for example clock ticks or mouse movement—the publisher must choose to either buffer or drop elements to obey the imposed bounds.

ISubscribers signaling a demand for one element after the reception of an element effectively implement a Stop-and-Wait protocol where the demand signal is equivalent to acknowledgement. By providing demand for multiple elements the cost of acknowledgement is amortized. It is worth noting that the subscriber is allowed to signal demand at any point in time, allowing it to avoid unnecessary delays between the publisher and the subscriber (i.e. keeping its input buffer filled without having to wait for full round-trips).

This project is a collaboration between engineers from Kaazing, Lightbend, Netflix, Pivotal, Red Hat, Twitter and many others. This project is licensed under MIT No Attribution (SPDX: MIT-0).