com.github.davidmoten.rx

Class Transformers

java.lang.Object

com.github.davidmoten.rx.Transformers

public final class Transformers
extends Object

Constructor Summary

Constructors

Constructor and Description
Transformers()

Method Summary

Modifier and Type	Method and Description
static <T> rx.Observable.Transformer<T,T>	bufferEmissions()
static <T> rx.Observable.Transformer<T,T>	cache(long duration, TimeUnit unit, rx.Scheduler.Worker worker)
static <T extends Number> rx.Observable.Transformer<T,Statistics>	collectStats()
static <T,R extends Number> rx.Observable.Transformer<T,Pair<T,Statistics>>	collectStats(rx.functions.Func1<? super T,? extends R> function)
static <T,R extends Collection<T>> rx.Observable.Transformer<T,R>	collectWhile(rx.functions.Func0<R> factory, rx.functions.Action2<? super R,? super T> collect) Returns a Observable.Transformer that returns an Observable that is collected into Collection instances created by factory that are emitted when items are not equal or on completion.
static <T,R extends Iterable<?>> rx.Observable.Transformer<T,R>	collectWhile(rx.functions.Func0<R> factory, rx.functions.Action2<? super R,? super T> collect, rx.functions.Func2<? super R,? super T,Boolean> condition)
static <T,R> rx.Observable.Transformer<T,R>	collectWhile(rx.functions.Func0<R> factory, rx.functions.Action2<? super R,? super T> collect, rx.functions.Func2<? super R,? super T,Boolean> condition, rx.functions.Func1<? super R,Boolean> isEmpty) Returns a Observable.Transformer that returns an Observable that is collected into Collection instances created by factory that are emitted when the collection and latest emission do not satisfy condition or on completion.
static rx.Observable.Transformer<byte[],String>	decode(CharsetDecoder charsetDecoder) Decodes a stream of multibyte chunks into a stream of strings that works on infinite streams and handles when a multibyte character spans two chunks.
static <T> rx.Observable.Transformer<T,T>	doOnFirst(rx.functions.Action1<? super T> action) Returns a Observable.Transformer that applied to a source Observable calls the given action on the first onNext emission.
static <T> rx.Observable.Transformer<T,T>	doOnNext(int n, rx.functions.Action1<? super T> action) Returns a Observable.Transformer that applied to a source Observable calls the given action on the nth onNext emission.
static <R,T> rx.Observable.Transformer<T,R>	ignoreElementsThen(rx.Observable<R> next)
static <T> rx.Observable.Transformer<T,T>	limitSubscribers(AtomicInteger subscriberCount, int maxSubscribers)
static <T> rx.Observable.Transformer<T,T>	limitSubscribers(int maxSubscribers)
static <T> rx.Observable.Transformer<T,MapWithIndex.Indexed<T>>	mapWithIndex()
static <T> rx.Observable.Transformer<T,T>	orderedMergeWith(Collection<rx.Observable<? extends T>> others, Comparator<? super T> comparator) Returns the source Observable merged with all of the other observables using the given Comparator for order.
static <T> rx.Observable.Transformer<T,T>	orderedMergeWith(rx.Observable<? extends T> other, Comparator<? super T> comparator) Returns the source Observable merged with the other observable using the given Comparator for order.
static <T> rx.Observable.Transformer<T,T>	sampleFirst(long duration, TimeUnit unit)
static <T> rx.Observable.Transformer<T,T>	sampleFirst(long duration, TimeUnit unit, rx.Scheduler scheduler)
static <T extends Comparable<? super T>> rx.Observable.Transformer<T,T>	sort()
static <T> rx.Observable.Transformer<T,T>	sort(Comparator<? super T> comparator)
static <T> rx.Observable.Transformer<String,String>	split(String pattern)
static <State,In,Out> rx.Observable.Transformer<In,Out>	stateMachine(rx.functions.Func0<? extends State> initialStateFactory, rx.functions.Func3<? super State,? super In,? super rx.Subscriber<Out>,? extends State> transition, rx.functions.Func2<? super State,? super rx.Subscriber<Out>,Boolean> completion) Returns a Observable.Transformer that allows processing of the source stream to be defined in a state machine where transitions of the state machine may also emit items to downstream that are buffered if necessary when backpressure is requested.
static <State,In,Out> rx.Observable.Transformer<In,Out>	stateMachine(rx.functions.Func0<State> initialStateFactory, rx.functions.Func3<? super State,? super In,? super rx.Subscriber<Out>,? extends State> transition, rx.functions.Func2<? super State,? super rx.Subscriber<Out>,Boolean> completion, BackpressureStrategy backpressureStrategy) Returns a Observable.Transformer that allows processing of the source stream to be defined in a state machine where transitions of the state machine may also emit items to downstream that are buffered if necessary when backpressure is requested.
static <T> rx.Observable.Transformer<T,List<T>>	toListUntilChanged() Returns a Observable.Transformer that returns an Observable that is a buffering of the source Observable into lists of sequential items that are equal.
static <T> rx.Observable.Transformer<T,List<T>>	toListWhile(rx.functions.Func2<? super List<T>,? super T,Boolean> condition) Returns a Observable.Transformer that returns an Observable that is a buffering of the source Observable into lists of sequential items that satisfy the condition condition.
static <T,R> rx.Observable.Operator<R,T>	toOperator(rx.functions.Func1<? super rx.Observable<T>,? extends rx.Observable<R>> function)
static <T> rx.Observable.Transformer<T,Set<T>>	toSet()

Methods inherited from class java.lang.Object

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

Constructor Detail

Transformers

public Transformers()

Method Detail

toOperator

public static <T,R> rx.Observable.Operator<R,T> toOperator(rx.functions.Func1<? super rx.Observable<T>,? extends rx.Observable<R>> function)

collectStats

public static <T extends Number> rx.Observable.Transformer<T,Statistics> collectStats()

collectStats

public static <T,R extends Number> rx.Observable.Transformer<T,Pair<T,Statistics>> collectStats(rx.functions.Func1<? super T,? extends R> function)

sort

public static <T extends Comparable<? super T>> rx.Observable.Transformer<T,T> sort()

sort

public static <T> rx.Observable.Transformer<T,T> sort(Comparator<? super T> comparator)

toSet

public static <T> rx.Observable.Transformer<T,Set<T>> toSet()

mapWithIndex

public static <T> rx.Observable.Transformer<T,MapWithIndex.Indexed<T>> mapWithIndex()

stateMachine

public static <State,In,Out> rx.Observable.Transformer<In,Out> stateMachine(rx.functions.Func0<State> initialStateFactory,
                                                                            rx.functions.Func3<? super State,? super In,? super rx.Subscriber<Out>,? extends State> transition,
                                                                            rx.functions.Func2<? super State,? super rx.Subscriber<Out>,Boolean> completion,
                                                                            BackpressureStrategy backpressureStrategy)

Returns a Observable.Transformer that allows processing of the source stream to be defined in a state machine where transitions of the state machine may also emit items to downstream that are buffered if necessary when backpressure is requested. flatMap is part of the processing chain so the source may experience requests for more items than are strictly required by the endpoint subscriber.

Type Parameters:

State - the class representing the state of the state machine

In - the input observable type

Out - the output observable type

Parameters:

initialStateFactory - the factory to create the initial state of the state machine.

transition - defines state transitions and consequent emissions to downstream when an item arrives from upstream. The Subscriber is called with the emissions to downstream. You can optionally call Subscriber.isUnsubscribed() to check if you can stop emitting from the transition. If you do wish to terminate the Observable then call Subscriber.unsubscribe() and return anything (say null from the transition (as the next state which will not be used). You can also complete the Observable by calling Observer.onCompleted() or Observer.onError(java.lang.Throwable) from within the transition and return anything from the transition (will not be used). The transition should run synchronously so that completion of a call to the transition should also signify all emissions from that transition have been made.

completion - defines activity that should happen based on the final state just before downstream onCompleted() is called. For example any buffered emissions in state could be emitted at this point. Don't call observer.onCompleted() as it is called for you after the action completes if and only if you return true from this function.

backpressureStrategy - is applied to the emissions from one call of transition and should enforce backpressure.

Returns:

a backpressure supporting transformer that implements the state machine specified by the parameters

Throws:

NullPointerException - if initialStateFactory or transition,or completionAction is null

stateMachine

public static <State,In,Out> rx.Observable.Transformer<In,Out> stateMachine(rx.functions.Func0<? extends State> initialStateFactory,
                                                                            rx.functions.Func3<? super State,? super In,? super rx.Subscriber<Out>,? extends State> transition,
                                                                            rx.functions.Func2<? super State,? super rx.Subscriber<Out>,Boolean> completion)

Returns a Observable.Transformer that allows processing of the source stream to be defined in a state machine where transitions of the state machine may also emit items to downstream that are buffered if necessary when backpressure is requested. flatMap is part of the processing chain so the source may experience requests for more items than are strictly required by the endpoint subscriber. The backpressure strategy used for emissions from the transition into the flatMap is BackpressureStrategy.BUFFER which corresponds to Observable.onBackpressureBuffer().

Type Parameters:

State - the class representing the state of the state machine

In - the input observable type

Out - the output observable type

Parameters:

initialStateFactory - the factory to create the initial state of the state machine.

transition - defines state transitions and consequent emissions to downstream when an item arrives from upstream. The Subscriber is called with the emissions to downstream. You can optionally call Subscriber.isUnsubscribed() to check if you can stop emitting from the transition. If you do wish to terminate the Observable then call Subscriber.unsubscribe() and return anything (say null from the transition (as the next state which will not be used). You can also complete the Observable by calling Observer.onCompleted() or Observer.onError(java.lang.Throwable) from within the transition and return anything from the transition (will not be used). The transition should run synchronously so that completion of a call to the transition should also signify all emissions from that transition have been made.

completion - defines activity that should happen based on the final state just before downstream onCompleted() is called. For example any buffered emissions in state could be emitted at this point. Don't call observer.onCompleted() as it is called for you after the action completes if and only if you return true from this function.

Returns:

a backpressure supporting transformer that implements the state machine specified by the parameters

Throws:

NullPointerException - if initialStateFactory or transition,or completion is null

bufferEmissions

public static <T> rx.Observable.Transformer<T,T> bufferEmissions()

orderedMergeWith

public static final <T> rx.Observable.Transformer<T,T> orderedMergeWith(rx.Observable<? extends T> other,
                                                                        Comparator<? super T> comparator)

Returns the source Observable merged with the other observable using the given Comparator for order. A precondition is that the source and other are already ordered. This transformer supports backpressure and its inputs must also support backpressure.

Type Parameters:

T - the generic type of the objects being compared

Parameters:

other - the other already ordered observable

comparator - the ordering to use

Returns:

merged and ordered observable

orderedMergeWith

public static final <T> rx.Observable.Transformer<T,T> orderedMergeWith(Collection<rx.Observable<? extends T>> others,
                                                                        Comparator<? super T> comparator)

Returns the source Observable merged with all of the other observables using the given Comparator for order. A precondition is that the source and other are already ordered. This transformer supports backpressure and its inputs must also support backpressure.

Type Parameters:

T - the generic type of the objects being compared

Parameters:

others - a collection of already ordered observables to merge with

comparator - the ordering to use

Returns:

merged and ordered observable

toListUntilChanged

public static <T> rx.Observable.Transformer<T,List<T>> toListUntilChanged()

Returns a Observable.Transformer that returns an Observable that is a buffering of the source Observable into lists of sequential items that are equal.

For example, the stream Observable.just(1, 1, 2, 2, 1).compose(toListUntilChanged()) would emit [1,1], [2], [1].

Type Parameters:

T - the generic type of the source Observable

Returns:

transformer as above

toListWhile

public static <T> rx.Observable.Transformer<T,List<T>> toListWhile(rx.functions.Func2<? super List<T>,? super T,Boolean> condition)

Returns a Observable.Transformer that returns an Observable that is a buffering of the source Observable into lists of sequential items that satisfy the condition condition.

Type Parameters:

T - the generic type of the source Observable

Parameters:

condition - condition function that must return true if an item is to be part of the list being prepared for emission

Returns:

transformer as above

collectWhile

public static <T,R> rx.Observable.Transformer<T,R> collectWhile(rx.functions.Func0<R> factory,
                                                                rx.functions.Action2<? super R,? super T> collect,
                                                                rx.functions.Func2<? super R,? super T,Boolean> condition,
                                                                rx.functions.Func1<? super R,Boolean> isEmpty)

Returns a Observable.Transformer that returns an Observable that is collected into Collection instances created by factory that are emitted when the collection and latest emission do not satisfy condition or on completion.

Type Parameters:

T - generic type of source observable

R - collection type emitted by transformed Observable

Parameters:

factory - collection instance creator

collect - collection action

condition - returns true if and only if emission should be collected in current collection being prepared for emission

isEmpty - indicates that the collection is empty

Returns:

transformer as above

collectWhile

public static <T,R extends Collection<T>> rx.Observable.Transformer<T,R> collectWhile(rx.functions.Func0<R> factory,
                                                                                      rx.functions.Action2<? super R,? super T> collect)

Returns a Observable.Transformer that returns an Observable that is collected into Collection instances created by factory that are emitted when items are not equal or on completion.

Type Parameters:

T - generic type of source observable

R - collection type emitted by transformed Observable

Parameters:

factory - collection instance creator

collect - collection action

Returns:

transformer as above

collectWhile

public static <T,R extends Iterable<?>> rx.Observable.Transformer<T,R> collectWhile(rx.functions.Func0<R> factory,
                                                                                    rx.functions.Action2<? super R,? super T> collect,
                                                                                    rx.functions.Func2<? super R,? super T,Boolean> condition)

doOnNext

public static <T> rx.Observable.Transformer<T,T> doOnNext(int n,
                                                          rx.functions.Action1<? super T> action)

Returns a Observable.Transformer that applied to a source Observable calls the given action on the nth onNext emission.

Type Parameters:

T - the generic type of the Observable being transformed

Parameters:

n - the 1-based count of onNext to do the action on

action - is performed on nth onNext.

Returns:

Transformer that applied to a source Observable calls the given action on the nth onNext emission.

doOnFirst

public static <T> rx.Observable.Transformer<T,T> doOnFirst(rx.functions.Action1<? super T> action)

Returns a Observable.Transformer that applied to a source Observable calls the given action on the first onNext emission.

Type Parameters:

T - the generic type of the Observable being transformed

Parameters:

action - is performed on first onNext

Returns:

Transformer that applied to a source Observable calls the given action on the first onNext emission.

ignoreElementsThen

public static <R,T> rx.Observable.Transformer<T,R> ignoreElementsThen(rx.Observable<R> next)

split

public static <T> rx.Observable.Transformer<String,String> split(String pattern)

decode

public static rx.Observable.Transformer<byte[],String> decode(CharsetDecoder charsetDecoder)

Decodes a stream of multibyte chunks into a stream of strings that works on infinite streams and handles when a multibyte character spans two chunks. This method allows for more control over how malformed and unmappable characters are handled.

Parameters:

charsetDecoder - decodes the bytes into strings

Returns:

the Observable returning a stream of decoded strings

limitSubscribers

public static <T> rx.Observable.Transformer<T,T> limitSubscribers(AtomicInteger subscriberCount,
                                                                  int maxSubscribers)

limitSubscribers

public static <T> rx.Observable.Transformer<T,T> limitSubscribers(int maxSubscribers)

cache

public static <T> rx.Observable.Transformer<T,T> cache(long duration,
                                                       TimeUnit unit,
                                                       rx.Scheduler.Worker worker)

sampleFirst

public static <T> rx.Observable.Transformer<T,T> sampleFirst(long duration,
                                                             TimeUnit unit)

sampleFirst

public static <T> rx.Observable.Transformer<T,T> sampleFirst(long duration,
                                                             TimeUnit unit,
                                                             rx.Scheduler scheduler)