T- type of the values in the sequence
public interface Observable<T> extends Iterable<T>
Observable is backed by a
Ringbuffer, which, once
created, has a fixed capacity for storing messages. It supports reading
Observer Observers, which will all observe the same
sequence of messages. A new
Observer will start reading
automatically from the oldest sequence available. Once the capacity is
full, the oldest messages will be overwritten as new ones arrive.
Ringbuffer's capacity defaults to , but can be changed (via the
configureCapacity(int) method), as long as the
been created yet (see the "Lifecycle" section below).
In addition to data events, the
Observer can 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
Ringbuffer, since that is the significant
The lifecycle of the
Ringbuffer is decoupled from the lifecycle
of the job. The
Ringbuffer is created either when the user
gets a reference to its equivalent
and registers the first
Observer on it (through
or when the job containing the sink for it starts executing.
Ringbuffer must 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
Observable POJO gets lost and its underlying
is leaked in the cluster, it's still possible to manually destroy
it later by creating another
Observable instance with the same
name and calling
destroy() on that.
Important: The same
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 sinks in 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).
|Modifier and Type||Method and Description|
Set the capacity of the underlying
Removes all previously registered observers and destroys the backing
Returns the configured capacity of the underlying
Returns an iterator over the sequence of events produced by this
Name of this instance.
Removes a previously added
Allows you to post-process the results of a Jet job on the client side using the standard Java
Observable. It will receive all events currently in the backing
Ringbufferand then continue receiving any future events.
Observer, can be used to remove the
Observeridentified by its assigned registration ID. A removed
Observerwill not get notified about further events.
Observable<T> configureCapacity(int capacity)
Ringbuffer, which defaults to .
This method can be called only before the
created. This means before any
Observers are added
Observable and before any jobs containing
sinks (with the same observable name) are submitted for execution.
Important: only configure capacity once, multiple configuration are currently not supported.
IllegalStateException- if the
Ringbufferhas already been created
This method only works if the backing
Ringbuffer has already
been created. If so, it will be queried for its actual capacity,
which can't be changed any longer. (Reminder: the
gets created either when the first
Observer is added or when
the job containing the
observable sink (with the same observable name) is submitted for
IllegalStateException- if the backing
Ringbufferhas not yet been created
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
Observable populated from a streaming job,
the iterator will complete only in the case of an error or job
The iterator is not thread-safe.
The iterator is backed by a blocking concurrent queue which stores all events until consumed.
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).
CompletableFuture that will become
completed once your function has received all the job results through
Stream and returned the final result.
A trivial example is counting, like this:
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
Note: if you call this while a job that publishes to this
Observable is still active, it will silently create a new
Ringbuffer and go on publishing to it.
Copyright © 2021 Hazelcast, Inc.. All rights reserved.