org.osgi.util.pushstream

Class PushStreamProvider

java.lang.Object

org.osgi.util.pushstream.PushStreamProvider

public final class PushStreamProvider
extends Object

A factory for PushStream instances, and utility methods for handling PushEventSources and PushEventConsumers

Constructor Summary

Constructors

Constructor and Description
PushStreamProvider()

Method Summary

Modifier and Type	Method and Description
<T,U extends BlockingQueue<PushEvent<? extends T>>> BufferBuilder<PushEventConsumer<T>,T,U>	buildBufferedConsumer(PushEventConsumer<T> delegate) Build a buffered PushEventConsumer with custom configuration.
<T,U extends BlockingQueue<PushEvent<? extends T>>> BufferBuilder<PushEventSource<T>,T,U>	buildEventSourceFromStream(PushStream<T> stream) Convert an PushStream into an PushEventSource.
<T,U extends BlockingQueue<PushEvent<? extends T>>> BufferBuilder<SimplePushEventSource<T>,T,U>	buildSimpleEventSource(Class<T> type) Build a SimplePushEventSource with the supplied type and custom buffering behaviors.
<T,U extends BlockingQueue<PushEvent<? extends T>>> PushStreamBuilder<T,U>	buildStream(PushEventSource<T> eventSource) Builds a push stream with custom configuration.
<T> PushEventConsumer<T>	createBufferedConsumer(PushEventConsumer<T> delegate) Create a buffered PushEventConsumer with the default configured buffer, executor size, queue, queue policy and pushback policy.
<T> PushEventSource<T>	createEventSourceFromStream(PushStream<T> stream) Convert an PushStream into an PushEventSource.
<T> SimplePushEventSource<T>	createSimpleEventSource(Class<T> type) Create a SimplePushEventSource with the supplied type and default buffering behaviors.
<T> PushStream<T>	createStream(PushEventSource<T> eventSource) Create a stream with the default configured buffer, executor size, queue, queue policy and pushback policy.
<T> PushStream<T>	streamOf(Executor executor, ScheduledExecutorService scheduler, Stream<T> items) Create an Unbuffered PushStream from a Java Stream The data from the stream will be pushed into the PushStream asynchronously using the supplied Executor.
<T> PushStream<T>	streamOf(Stream<T> items) Create an Unbuffered PushStream from a Java Stream The data from the stream will be pushed into the PushStream synchronously as it is opened.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

PushStreamProvider

public PushStreamProvider()

Method Detail

createStream

public <T> PushStream<T> createStream(PushEventSource<T> eventSource)

Create a stream with the default configured buffer, executor size, queue, queue policy and pushback policy. This is equivalent to calling

 buildStream(source).create();
 

This stream will be buffered from the event producer, and will honor back pressure even if the source does not.

Buffered streams are useful for "bursty" event sources which produce a number of events close together, then none for some time. These bursts can sometimes overwhelm downstream processors. Buffering will not, however, protect downstream components from a source which produces events faster (on average) than they can be consumed.

Event delivery will not begin until a terminal operation is reached on the chain of PushStreams. Once a terminal operation is reached the stream will be connected to the event source.

Parameters:

eventSource -

Returns:

A PushStream with a default initial buffer

buildStream

public <T,U extends BlockingQueue<PushEvent<? extends T>>> PushStreamBuilder<T,U> buildStream(PushEventSource<T> eventSource)

Builds a push stream with custom configuration.

The resulting PushStream may be buffered or unbuffered depending on how it is configured.

Parameters:

eventSource - The source of the events

Returns:

A PushStreamBuilder for the stream

createEventSourceFromStream

public <T> PushEventSource<T> createEventSourceFromStream(PushStream<T> stream)

Convert an PushStream into an PushEventSource. The first call to PushEventSource.open(PushEventConsumer) will begin event processing. The PushEventSource will remain active until the backing stream is closed, and permits multiple consumers to PushEventSource.open(PushEventConsumer) it. This is equivalent to:

 buildEventSourceFromStream(stream).create();
 

Parameters:

stream -

Returns:

a PushEventSource backed by the PushStream

buildEventSourceFromStream

public <T,U extends BlockingQueue<PushEvent<? extends T>>> BufferBuilder<PushEventSource<T>,T,U> buildEventSourceFromStream(PushStream<T> stream)

Convert an PushStream into an PushEventSource. The first call to PushEventSource.open(PushEventConsumer) will begin event processing.

The PushEventSource will remain active until the backing stream is closed, and permits multiple consumers to PushEventSource.open(PushEventConsumer) it. Note that this means the caller of this method is responsible for closing the supplied stream if it is not finite in length.

Late joining consumers will not receive historical events, but will immediately receive the terminal event which closed the stream if the stream is already closed.

Parameters:

stream -

Returns:

a PushEventSource backed by the PushStream

createSimpleEventSource

public <T> SimplePushEventSource<T> createSimpleEventSource(Class<T> type)

Create a SimplePushEventSource with the supplied type and default buffering behaviors. The SimplePushEventSource will respond to back pressure requests from the consumers connected to it. This is equivalent to:

 buildSimpleEventSource(type).create();
 

Parameters:

type -

Returns:

a SimplePushEventSource

buildSimpleEventSource

public <T,U extends BlockingQueue<PushEvent<? extends T>>> BufferBuilder<SimplePushEventSource<T>,T,U> buildSimpleEventSource(Class<T> type)

Build a SimplePushEventSource with the supplied type and custom buffering behaviors. The SimplePushEventSource will respond to back pressure requests from the consumers connected to it.

Parameters:

type -

Returns:

a SimplePushEventSource

createBufferedConsumer

public <T> PushEventConsumer<T> createBufferedConsumer(PushEventConsumer<T> delegate)

Create a buffered PushEventConsumer with the default configured buffer, executor size, queue, queue policy and pushback policy. This is equivalent to calling

 buildBufferedConsumer(delegate).create();
 

The returned consumer will be buffered from the event source, and will honor back pressure requests from its delegate even if the event source does not.

Buffered consumers are useful for "bursty" event sources which produce a number of events close together, then none for some time. These bursts can sometimes overwhelm the consumer. Buffering will not, however, protect downstream components from a source which produces events faster than they can be consumed.

Parameters:

delegate -

Returns:

a PushEventConsumer with a buffer directly before it

buildBufferedConsumer

public <T,U extends BlockingQueue<PushEvent<? extends T>>> BufferBuilder<PushEventConsumer<T>,T,U> buildBufferedConsumer(PushEventConsumer<T> delegate)

Build a buffered PushEventConsumer with custom configuration.

The returned consumer will be buffered from the event source, and will honor back pressure requests from its delegate even if the event source does not.

Buffered consumers are useful for "bursty" event sources which produce a number of events close together, then none for some time. These bursts can sometimes overwhelm the consumer. Buffering will not, however, protect downstream components from a source which produces events faster than they can be consumed.

Buffers are also useful as "circuit breakers". If a QueuePolicyOption.FAIL is used then a full buffer will request that the stream close, preventing an event storm from reaching the client.

Note that this buffered consumer will close when it receives a terminal event, or if the delegate returns negative backpressure. No further events will be propagated after this time.

Parameters:

delegate -

Returns:

a PushEventConsumer with a buffer directly before it

streamOf

public <T> PushStream<T> streamOf(Stream<T> items)

Create an Unbuffered PushStream from a Java Stream The data from the stream will be pushed into the PushStream synchronously as it is opened. This may make terminal operations blocking unless a buffer has been added to the PushStream. Care should be taken with infinite Streams to avoid blocking indefinitely.

Parameters:

items - The items to push into the PushStream

Returns:

A PushStream containing the items from the Java Stream

streamOf

public <T> PushStream<T> streamOf(Executor executor,
                                  ScheduledExecutorService scheduler,
                                  Stream<T> items)

Create an Unbuffered PushStream from a Java Stream The data from the stream will be pushed into the PushStream asynchronously using the supplied Executor.

Parameters:

executor - The worker to use to push items from the Stream into the PushStream

scheduler - The scheduler to use to trigger timed events in the PushStream

items - The items to push into the PushStream

Returns:

A PushStream containing the items from the Java Stream