Package 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 Details

    • PushStreamProvider

      public PushStreamProvider()

  • Method Details

    • 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