- Type Parameters:
T- type of the values in the sequence
- All Superinterfaces:
public interface Observable<T> extends java.lang.Iterable<T>Represents a flowing sequence of events produced by observable sinks. To observe the events, call
Observableis backed by a
Ringbuffer, which, once created, has a fixed capacity for storing messages. It supports reading by multiple
Observer Observers, which will all observe the same sequence of messages. A new
Observerwill start reading automatically from the oldest sequence available. Once the capacity is full, the oldest messages will be overwritten as new ones arrive.
In addition to data events, the
Observercan also observe completion and failure events. Completion means that no further values will appear in the sequence. Failure means that something went wrong during the job execution .
When talking about the lifecycle of an
Observable(which is basically just a client side object and has a lifecycle just like any other POJO) it's better to actually consider the lifecycle of the underlying
Ringbuffer, since that is the significant distributed entity.
The lifecycle of the
Ringbufferis decoupled from the lifecycle of the job. The
Ringbufferis created either when the user gets a reference to its equivalent
JetService.getObservable()) and registers the first
Observeron it (through
Observable.addObserver()) or when the job containing the sink for it starts executing.
Ringbuffermust be explicitly destroyed when it's no longer in use, or data will be retained in the cluster. This is done via the
Observable.destroy()method. Note: even if the
ObservablePOJO gets lost and its underlying
Ringbufferis leaked in the cluster, it's still possible to manually destroy it later by creating another
Observableinstance with the same name and calling
Important: The same
Observablemust not be used again in a new job since this will cause completion events interleaving and causing data loss or other unexpected behaviour. Using one observable name in multiple
observable sinksin the same job is allowed, this will not produce multiple completion or error events (just an intermingling of the results from the two sinks, but that should be fine in some use cases).
- Jet 4.0
All Methods Instance Methods Abstract Methods Default Methods Modifier and Type Method Description
addObserver(Observer<T> observer)Registers an
destroy()Removes all previously registered observers and destroys the backing
getConfiguredCapacity()Returns the configured capacity of the underlying
iterator()Returns an iterator over the sequence of events produced by this
name()Name of this instance.
removeObserver(java.util.UUID registrationId)Removes a previously added
Observeridentified by its assigned registration ID.
default <R> java.util.concurrent.CompletableFuture<R>
toFuture(java.util.function.Function<java.util.stream.Stream<T>,R> fn)Allows you to post-process the results of a Jet job on the client side using the standard Java
@Nonnull java.lang.String name()Name of this instance.
Observable. It will receive all events currently in the backing
Ringbufferand then continue receiving any future events.
- registration ID associated with the added
Observer, can be used to remove the
void removeObserver(@Nonnull java.util.UUID registrationId)Removes a previously added
Observeridentified by its assigned registration ID. A removed
Observerwill not get notified about further events.
Observable<T> configureCapacity(int capacity)Set the capacity of the underlying
Ringbuffer, which defaults to 10000.
This method can be called only before the
Ringbuffergets created. This means before any
Observersare added to the
Observableand before any jobs containing
observable sinks(with the same observable name) are submitted for execution.
Important: only configure capacity once, multiple configuration are currently not supported.
java.lang.IllegalStateException- if the
Ringbufferhas already been created
int getConfiguredCapacity()Returns the configured capacity of the underlying
This method only works if the backing
Ringbufferhas already been created. If so, it will be queried for its actual capacity, which can't be changed any longer. (Reminder: the
Ringbuffergets created either when the first
Observeris added or when the job containing the
observable sink(with the same observable name) is submitted for execution.)
java.lang.IllegalStateException- if the backing
Ringbufferhas not yet been created
@Nonnull default java.util.Iterator<T> iterator()Returns an iterator over the sequence of events produced by this
Observable. If there are currently no events to observe, the iterator's
next()methods will block. A completion event completes the iterator (
hasNext()will return false) and a failure event makes the iterator's methods throw the underlying exception.
If used against an
Observablepopulated from a streaming job, the iterator will complete only in the case of an error or job cancellation.
The iterator is not thread-safe.
The iterator is backed by a blocking concurrent queue which stores all events until consumed.
- Specified by:
@Nonnull default <R> java.util.concurrent.CompletableFuture<R> toFuture(@Nonnull java.util.function.Function<java.util.stream.Stream<T>,R> fn)Allows you to post-process the results of a Jet job on the client side using the standard Java
Stream API. You provide a function that will receive the job results as a
Stream<T>and return a single result (which can in fact be another
Stream, if so desired).
CompletableFuturethat will become completed once your function has received all the job results through its
Streamand returned the final result.
A trivial example is counting, like this:
observable.toFuture(Stream::count), however the Stream API is quite rich and you can perform arbitrary transformations and aggregations.
This feature is intended to be used only on the results of a batch job. On an unbounded streaming job the stream-collecting operation will never reach the final result.
fn- transform function which takes the stream of observed values and produces an altered value from it, which could also be a stream
void destroy()Removes all previously registered observers and destroys the backing
Note: if you call this while a job that publishes to this
Observableis still active, it will silently create a new
Ringbufferand go on publishing to it.