uk.co.strangeskies.observable

Interface Observable<M>

Type Parameters:

M - The message type. This may be Void if no message need be sent.

All Known Subinterfaces:

LocaleManager, LocaleProvider, Localized<T>, ObservableProperty<T>, ObservableValue<T>

All Known Implementing Classes:

ColdObservable, HotObservable, ImmutableObservable, ImmutableObservableValue, LocaleManagerService, ObservablePropertyImpl

public interface Observable<M>

Simple interface for an observable object, with methods to add and remove observers expecting the applicable type of message.

Method Summary

Modifier and Type	Method and Description
default Observable<java.util.List<M>>	aggregateBackpressure()
default Observable<java.util.List<M>>	aggregateBackpressure(long toCapacity)
default <R,A> java.util.concurrent.CompletableFuture<R>	collect(java.util.stream.Collector<? super M,A,? extends R> collector)
default <R,A> Observable<R>	collectBackpressure(java.util.stream.Collector<? super M,A,? extends R> collector) Introduce backpressure by collecting messages until a request is made downstream, then forwarding the collection.
default <T> Observable<T>	compose(java.util.function.Function<Observable<? super M>,Observable<? extends T>> transformation) Derive a new observable by application of the given function.
static <M> Observable<M>	concat(java.util.Collection<? extends Observable<? extends M>> observables)
static <M> Observable<M>	concat(Observable<? extends M>... observables)
default <T> Observable<T>	concatMap(java.util.function.Function<? super M,? extends Observable<? extends T>> mapping) As flatMap(Function, RequestAllocator) using sequential request allocation.
default Observable<M>	dropWhile(java.util.function.Predicate<? super M> condition) Derive an observable which completes and disposes itself after receiving a message which matches the given condition.
static <T> Observable<T>	empty()
default Observable<M>	executeOn(java.util.concurrent.Executor executor) Derive an observable which re-emits messages on the given executor.
static <M> ObservableValue<M>	failingValue(java.lang.Throwable failure)
default Observable<M>	filter(java.util.function.Predicate<? super M> condition) Derive an observable which passes along only those messages which match the given condition.
default <T> Observable<T>	flatMap(java.util.function.Function<? super M,? extends Observable<? extends T>> mapping, RequestAllocator requestAllocator) Derive an observable which maps each message to an intermediate observable, then combines those intermediate observables into one.
default M	get() Block until we either receive the next message event of the next failure event.
default java.util.concurrent.CompletableFuture<M>	getNext() Block until we either receive the next message event of the next failure event.
default <T> Observable<T>	map(java.util.function.Function<? super M,? extends T> mapping) Derive an observable which transforms messages according to the given mapping.
default Observable<Observable<M>>	materialize()
static <M> Observable<M>	merge(java.util.Collection<? extends Observable<? extends M>> observables)
static <M> Observable<M>	merge(Observable<? extends M>... observables)
default <T> Observable<T>	mergeMap(java.util.function.Function<? super M,? extends Observable<? extends T>> mapping) A common case of flatMap(Function, RequestAllocator) using balanced request allocation.
default Disposable	observe() As observe(Observer) with an empty observer.
Disposable	observe(Observer<? super M> observer) Observers added will receive messages from this Observable.
static <M> Observable<M>	of(java.util.Collection<? extends M> messages)
static <M> Observable<M>	of(M... messages)
static <M> Observable<M>	of(java.util.Optional<? extends M> messages)
default <R> java.util.concurrent.CompletableFuture<R>	reduce(java.util.function.Supplier<R> identity, java.util.function.BiFunction<R,? super M,R> accumulator)
default Observable<M>	reduceBackpressure(java.util.function.BinaryOperator<M> accumulator) Introduce backpressure by reducing messages until a request is made downstream, then forwarding the reduction.
default <R> Observable<R>	reduceBackpressure(java.util.function.Function<? super M,? extends R> initial, java.util.function.BiFunction<? super R,? super M,? extends R> accumulator) Introduce backpressure by reducing messages until a request is made downstream, then forwarding the reduction.
default <R> Observable<R>	reduceBackpressure(java.util.function.Supplier<? extends R> identity, java.util.function.BiFunction<? super R,? super M,? extends R> accumulator) Introduce backpressure by reducing messages until a request is made downstream, then forwarding the reduction.
default Observable<M>	repeating()
default Observable<M>	requestNext()
default Observable<M>	requestUnbounded()
default Observable<M>	retrying()
default Observable<M>	softReference() Derive an observable which automatically disposes of observers at some point after they are no longer softly reachable.
default <O> Observable<OwnedMessage<O,M>>	softReference(O owner) Derive an observable which automatically disposes of observers at some point after the given owner is no longer softly reachable.
default Observable<M>	synchronize()
default Observable<M>	synchronize(java.lang.Object mutex)
default Observable<M>	takeWhile(java.util.function.Predicate<? super M> condition) Derive an observable which completes and disposes itself after receiving a message which matches the given condition.
default Observable<M>	then(Observer<? super M> action) Derive an observable which passes events to the given observer directly before passing them downstream.
default Observable<M>	thenAfter(Observer<? super M> action) Derive an observable which passes events to the given observer directly after passing them downstream.
static <M> java.util.stream.Collector<? extends M,?,Observable<M>>	toObservable() A collector which can be applied to a Stream to derive a cold observable.
default ObservableValue<M>	toValue()
default ObservableValue<M>	toValue(M initial)
default ObservableValue<M>	toValue(java.lang.Throwable initialProblem)
default java.util.Optional<M>	tryGet()
static <T,U extends T> Observable<T>	upcast(Observable<U> from)
static <M> ObservableValue<M>	value(M value)
default Observable<M>	weakReference() Derive an observable which automatically disposes of observers at some point after they are no longer weakly reachable.
default <O> Observable<OwnedMessage<O,M>>	weakReference(O owner) Derive an observable which automatically disposes of observers at some point after the given owner is no longer weakly reachable.

Method Detail

observe

Disposable observe(Observer<? super M> observer)

Observers added will receive messages from this Observable.

Parameters:

observer - an observer to add

Returns:

a disposable over the observation

observe

default Disposable observe()

As observe(Observer) with an empty observer.

Returns:

a disposable over the observation

getNext

default java.util.concurrent.CompletableFuture<M> getNext()

Block until we either receive the next message event of the next failure event. In the case of the former, return it, and in the case of the latter, throw a MissingValueException.

Returns:

the next value

Throws:

MissingValueException - If a failure or completion event is received before the next message event. In the former case the cause will be the failure throwable, in the latter case an instance of AlreadyCompletedException.

get

default M get()

Block until we either receive the next message event of the next failure event. In the case of the former, return it, and in the case of the latter, throw a MissingValueException.

Returns:

the next value

Throws:

MissingValueException - If a failure or completion event is received before the next message event. In the former case the cause will be the failure throwable, in the latter case an instance of AlreadyCompletedException.

tryGet

default java.util.Optional<M> tryGet()

compose

default <T> Observable<T> compose(java.util.function.Function<Observable<? super M>,Observable<? extends T>> transformation)

Derive a new observable by application of the given function. This gives the same result as just applying the function to the observable directly, and exists only to create a more natural fit into the fluent API by making the order of operations clearer in method chains.

Type Parameters:

T - the type of the resulting observable

Parameters:

transformation - the transformation function to apply to the observable

Returns:

the derived observable

upcast

static <T,U extends T> Observable<T> upcast(Observable<U> from)

toObservable

static <M> java.util.stream.Collector<? extends M,?,Observable<M>> toObservable()

A collector which can be applied to a Stream to derive a cold observable.

Type Parameters:

M - the type of the observable

Returns:

an observable over the given stream

then

default Observable<M> then(Observer<? super M> action)

Derive an observable which passes events to the given observer directly before passing them downstream.

Parameters:

action - an observer representing the action to take

Returns:

an observable which performs the injected behavior

thenAfter

default Observable<M> thenAfter(Observer<? super M> action)

Derive an observable which passes events to the given observer directly after passing them downstream.

Parameters:

action - an observer representing the action to take

Returns:

an observable which performs the injected behavior

requestUnbounded

default Observable<M> requestUnbounded()

requestNext

default Observable<M> requestNext()

retrying

default Observable<M> retrying()

repeating

default Observable<M> repeating()

materialize

default Observable<Observable<M>> materialize()

toValue

default ObservableValue<M> toValue()

toValue

default ObservableValue<M> toValue(M initial)

toValue

default ObservableValue<M> toValue(java.lang.Throwable initialProblem)

weakReference

default Observable<M> weakReference()

Derive an observable which automatically disposes of observers at some point after they are no longer weakly reachable.

Returns:

the derived observable

weakReference

default <O> Observable<OwnedMessage<O,M>> weakReference(O owner)

Derive an observable which automatically disposes of observers at some point after the given owner is no longer weakly reachable.

Care should be taken not to refer to the owner directly in any observer logic, as this will create a strong reference to the owner, preventing it from becoming unreachable. For this reason, the message is transformed into an OwnedMessage, which may create references to the owner on demand within observer logic without retainment.

Type Parameters:

O - the type of the owning object

Parameters:

owner - the owning referent object

Returns:

the derived observable

softReference

default Observable<M> softReference()

Derive an observable which automatically disposes of observers at some point after they are no longer softly reachable.

Returns:

the derived observable

softReference

default <O> Observable<OwnedMessage<O,M>> softReference(O owner)

Derive an observable which automatically disposes of observers at some point after the given owner is no longer softly reachable.

Care should be taken not to refer to the owner directly in any observer logic, as this will create a strong reference to the owner, preventing it from becoming unreachable. For this reason, the message is transformed into an OwnedMessage, which may create references to the owner on demand within observer logic without retainment.

Type Parameters:

O - the type of the owning object

Parameters:

owner - the owning referent object

Returns:

the derived observable

executeOn

default Observable<M> executeOn(java.util.concurrent.Executor executor)

Derive an observable which re-emits messages on the given executor.

Parameters:

executor - the target executor

Returns:

the derived observable

map

default <T> Observable<T> map(java.util.function.Function<? super M,? extends T> mapping)

Derive an observable which transforms messages according to the given mapping.

Type Parameters:

T - the type of the derived observable

Parameters:

mapping - the mapping function

Returns:

an observable over the mapped messages

filter

default Observable<M> filter(java.util.function.Predicate<? super M> condition)

Derive an observable which passes along only those messages which match the given condition.

Parameters:

condition - the terminating condition

Returns:

the derived observable

takeWhile

default Observable<M> takeWhile(java.util.function.Predicate<? super M> condition)

Derive an observable which completes and disposes itself after receiving a message which matches the given condition.

Parameters:

condition - the terminating condition

Returns:

the derived observable

dropWhile

default Observable<M> dropWhile(java.util.function.Predicate<? super M> condition)

Derive an observable which completes and disposes itself after receiving a message which matches the given condition.

Parameters:

condition - the terminating condition

Returns:

the derived observable

synchronize

default Observable<M> synchronize()

synchronize

default Observable<M> synchronize(java.lang.Object mutex)

mergeMap

default <T> Observable<T> mergeMap(java.util.function.Function<? super M,? extends Observable<? extends T>> mapping)

A common case of flatMap(Function, RequestAllocator) using balanced request allocation.

An unbounded request is made to the upstream observable, so it is not required to support backpressure.

The intermediate observables are not required to support backpressure, as an unbounded request will be made to them and the downstream observable will forward every message as soon as it is available. Because of this, The downstream observable does not support backpressure.

Type Parameters:

T - the resulting observable message type

Parameters:

mapping - the terminating condition

Returns:

the derived observable

concatMap

default <T> Observable<T> concatMap(java.util.function.Function<? super M,? extends Observable<? extends T>> mapping)

As flatMap(Function, RequestAllocator) using sequential request allocation.

Type Parameters:

T - the resulting observable message type

Parameters:

mapping - the terminating condition

Returns:

the derived observable

flatMap

default <T> Observable<T> flatMap(java.util.function.Function<? super M,? extends Observable<? extends T>> mapping,
                                  RequestAllocator requestAllocator)

Derive an observable which maps each message to an intermediate observable, then combines those intermediate observables into one.

The intermediate observables accept requests from downstream until they are complete. Requests are allocated to the intermediate observables by the given request allocation strategy.

The upstream observable is not required to support backpressure. If a request is made downstream when there are no intermediate observables to fulfill that request, another message is requested from upstream.

The resulting observable supports backpressure if and only if the intermediate observables support backpressure.

Type Parameters:

T - the resulting observable message type

Parameters:

mapping - the terminating condition

requestAllocator - the strategy for allocating downstream requests to upstream observations

Returns:

the derived observable

reduce

default <R> java.util.concurrent.CompletableFuture<R> reduce(java.util.function.Supplier<R> identity,
                                                             java.util.function.BiFunction<R,? super M,R> accumulator)

reduceBackpressure

default <R> Observable<R> reduceBackpressure(java.util.function.Supplier<? extends R> identity,
                                             java.util.function.BiFunction<? super R,? super M,? extends R> accumulator)

Introduce backpressure by reducing messages until a request is made downstream, then forwarding the reduction.

Type Parameters:

R - the resulting reduction type

Parameters:

identity - the identity value for the accumulating function

accumulator - an associative, non-interfering, stateless function for combining two values

Returns:

an observable over the reduced values

reduceBackpressure

default <R> Observable<R> reduceBackpressure(java.util.function.Function<? super M,? extends R> initial,
                                             java.util.function.BiFunction<? super R,? super M,? extends R> accumulator)

Introduce backpressure by reducing messages until a request is made downstream, then forwarding the reduction.

Type Parameters:

R - the resulting reduction type

Parameters:

initial - the initial value for the accumulating function

accumulator - an associative, non-interfering, stateless function for combining two values

Returns:

an observable over the reduced values

reduceBackpressure

default Observable<M> reduceBackpressure(java.util.function.BinaryOperator<M> accumulator)

Introduce backpressure by reducing messages until a request is made downstream, then forwarding the reduction.

Parameters:

accumulator - an associative, non-interfering, stateless function for combining two values

Returns:

an observable over the reduced values

collect

default <R,A> java.util.concurrent.CompletableFuture<R> collect(java.util.stream.Collector<? super M,A,? extends R> collector)

collectBackpressure

default <R,A> Observable<R> collectBackpressure(java.util.stream.Collector<? super M,A,? extends R> collector)

Introduce backpressure by collecting messages until a request is made downstream, then forwarding the collection.

Type Parameters:

R - the resulting collection type

A - the intermediate collection type

Parameters:

collector - the collector to apply to incoming messages

Returns:

an observable over the collected values

aggregateBackpressure

default Observable<java.util.List<M>> aggregateBackpressure()

aggregateBackpressure

default Observable<java.util.List<M>> aggregateBackpressure(long toCapacity)

of

@SafeVarargs
static <M> Observable<M> of(M... messages)

of

static <M> Observable<M> of(java.util.Collection<? extends M> messages)

of

static <M> Observable<M> of(java.util.Optional<? extends M> messages)

merge

@SafeVarargs
static <M> Observable<M> merge(Observable<? extends M>... observables)

merge

static <M> Observable<M> merge(java.util.Collection<? extends Observable<? extends M>> observables)

concat

@SafeVarargs
static <M> Observable<M> concat(Observable<? extends M>... observables)

concat

static <M> Observable<M> concat(java.util.Collection<? extends Observable<? extends M>> observables)

failingValue

static <M> ObservableValue<M> failingValue(java.lang.Throwable failure)

value

static <M> ObservableValue<M> value(M value)

empty

static <T> Observable<T> empty()