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,List<T>>	bufferUntil(rx.functions.Func1<? super T,Boolean> predicate) Buffers the elements into continuous, non-overlapping Lists where the boundary is determined by a predicate receiving each item, after being buffered, and returns true to indicate a new buffer should start.
static <T> rx.Observable.Transformer<T,List<T>>	bufferUntil(rx.functions.Func1<? super T,Boolean> predicate, int capacityHint) Buffers the elements into continuous, non-overlapping Lists where the boundary is determined by a predicate receiving each item, after being buffered, and returns true to indicate a new buffer should start.
static <T> rx.Observable.Transformer<T,List<T>>	bufferWhile(rx.functions.Func1<? super T,Boolean> predicate) Buffers the elements into continuous, non-overlapping Lists where the boundary is determined by a predicate receiving each item, before or after being buffered, and returns true to indicate a new buffer should start.
static <T> rx.Observable.Transformer<T,List<T>>	bufferWhile(rx.functions.Func1<? super T,Boolean> predicate, int capacityHint) Buffers the elements into continuous, non-overlapping Lists where the boundary is determined by a predicate receiving each item, before being buffered, and returns true to indicate a new buffer should start.
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>	delay(rx.functions.Func1<? super T,Long> time, rx.functions.Func0<Double> playRate, long startTime, rx.Scheduler scheduler)
static <T> rx.Observable.Transformer<T,T>	delayFinalUnsubscribe(long duration, TimeUnit unit) If multiple concurrently open subscriptions happen to a source transformed by this method then an additional do-nothing subscription will be maintained to the source and will only be closed after the specified duration has passed from the final unsubscription of the open subscriptions.
static <T> rx.Observable.Transformer<T,T>	delayFinalUnsubscribe(long duration, TimeUnit unit, rx.Scheduler scheduler) If multiple concurrently open subscriptions happen to a source transformed by this method then an additional do-nothing subscription will be maintained to the source and will only be closed after the specified duration has passed from the final unsubscription of the open subscriptions.
static <T> rx.Observable.Transformer<T,T>	doOnEmpty(rx.functions.Action0 onEmpty) Modifies the source Observable so that it invokes an action when it calls onCompleted and no items were emitted.
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 <T,K,R> rx.Observable.Transformer<T,rx.observables.GroupedObservable<K,R>>	groupByEvicting(rx.functions.Func1<? super T,? extends K> keySelector, rx.functions.Func1<? super T,? extends R> elementSelector, rx.functions.Func1<rx.functions.Action1<K>,Map<K,Object>> evictingMapFactory) Groups the items emitted by an Observable according to a specified criterion, and emits these grouped items as GroupedObservables.
static <R,T> rx.Observable.Transformer<T,R>	ignoreElementsThen(rx.Observable<R> next) Returns an observable that subscribes to this and wait for completion but doesn't emit any items and once completes emits the next observable.
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,T>	mapLast(rx.functions.Func1<? super T,? extends T> function)
static <T> rx.Observable.Transformer<T,MapWithIndex.Indexed<T>>	mapWithIndex() Returns a Observable.Transformer that wraps stream emissions with their corresponding zero based index numbers (0,1,2,3,..) in instances of MapWithIndex.Indexed.
static <T> rx.Observable.Transformer<T,T>	onBackpressureBufferRequestLimiting() Rather than requesting Long.MAX_VALUE of upstream as does `Observable.onBackpressureBuffer`, this variant only requests of upstream what is requested of it.
static <T> rx.Observable.Transformer<T,T>	onBackpressureBufferToFile()
static <T> rx.Observable.Transformer<T,T>	onBackpressureBufferToFile(DataSerializer<T> serializer)
static <T> rx.Observable.Transformer<T,T>	onBackpressureBufferToFile(DataSerializer<T> serializer, rx.Scheduler scheduler)
static <T> rx.Observable.Transformer<T,T>	onBackpressureBufferToFile(DataSerializer<T> serializer, rx.Scheduler scheduler, Options options)
static <T> rx.Observable.Transformer<T,T>	onTerminateResume(rx.functions.Func1<Throwable,rx.Observable<T>> onError, rx.Observable<T> onCompleted)
static <T> rx.Observable.Transformer<T,T>	orderedMergeWith(Collection<rx.Observable<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<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>	removePairs(rx.functions.Func1<? super T,Boolean> isCandidateForFirst, rx.functions.Func2<? super T,? super T,Boolean> remove) Removes pairs non-recursively from a stream.
static <T> rx.Observable.Transformer<T,T>	repeatLast()
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(Pattern pattern)
static <T> rx.Observable.Transformer<String,String>	split(String pattern)
static StateMachine.Builder	stateMachine()
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 <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, int initialRequest)
static <T> rx.Observable.Transformer<T,List<T>>	toListUntil(rx.functions.Func1<? super T,Boolean> predicate) Buffers the elements into continuous, non-overlapping Lists where the boundary is determined by a predicate receiving each item, after being buffered, and returns true to indicate a new buffer should start.
static <T> rx.Observable.Transformer<T,List<T>>	toListUntil(rx.functions.Func1<? super T,Boolean> predicate, int capacityHint) Buffers the elements into continuous, non-overlapping Lists where the boundary is determined by a predicate receiving each item, after being buffered, and returns true to indicate a new buffer should start.
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.Func1<? super T,Boolean> predicate) Buffers the elements into continuous, non-overlapping Lists where the boundary is determined by a predicate receiving each item, before or after being buffered, and returns true to indicate a new buffer should start.
static <T> rx.Observable.Transformer<T,List<T>>	toListWhile(rx.functions.Func1<? super T,Boolean> predicate, int capacityHint) Buffers the elements into continuous, non-overlapping Lists where the boundary is determined by a predicate receiving each item, before being buffered, and returns true to indicate a new buffer should start.
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()
static <T extends Comparable<T>> rx.Observable.Transformer<T,T>	windowMax(int windowSize)
static <T> rx.Observable.Transformer<T,T>	windowMax(int windowSize, Comparator<? super T> comparator)
static <T extends Comparable<T>> rx.Observable.Transformer<T,T>	windowMin(int windowSize)
static <T> rx.Observable.Transformer<T,T>	windowMin(int windowSize, Comparator<? super T> comparator)

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()

Returns a Observable.Transformer that wraps stream emissions with their corresponding zero based index numbers (0,1,2,3,..) in instances of MapWithIndex.Indexed.

Example usage:

 
  
    Observable
      .just("a","b","c)
      .mapWithIndex(Transformers.mapWithIndex())
      .map(x -> x.index() + "->" + x.value())
      .forEach(System.out::println);
  
 

Type Parameters:

T - generic type of the stream being supplemented with an index

Returns:

transformer that supplements each stream emission with its index (zero-based position) in the stream.

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<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,
                                                                            int initialRequest)

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

stateMachine

public static StateMachine.Builder stateMachine()

orderedMergeWith

public static final <T> rx.Observable.Transformer<T,T> orderedMergeWith(rx.Observable<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<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)

Returns an observable that subscribes to this and wait for completion but doesn't emit any items and once completes emits the next observable.

Type Parameters:

R - input observable type

T - output observable type

Parameters:

next - observable to be emitted after ignoring elements of this

Returns:

Transformer that applied to a source Observable ignores the elements of the source and emits the elements of a second observable

split

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

split

public static <T> rx.Observable.Transformer<String,String> split(Pattern 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)

onBackpressureBufferToFile

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

onBackpressureBufferToFile

public static <T> rx.Observable.Transformer<T,T> onBackpressureBufferToFile(DataSerializer<T> serializer)

onBackpressureBufferToFile

public static <T> rx.Observable.Transformer<T,T> onBackpressureBufferToFile(DataSerializer<T> serializer,
                                                                            rx.Scheduler scheduler)

onBackpressureBufferToFile

public static <T> rx.Observable.Transformer<T,T> onBackpressureBufferToFile(DataSerializer<T> serializer,
                                                                            rx.Scheduler scheduler,
                                                                            Options options)

windowMin

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

windowMax

public static <T extends Comparable<T>> rx.Observable.Transformer<T,T> windowMax(int windowSize)

windowMax

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

windowMin

public static <T extends Comparable<T>> rx.Observable.Transformer<T,T> windowMin(int windowSize)

groupByEvicting

public static <T,K,R> rx.Observable.Transformer<T,rx.observables.GroupedObservable<K,R>> groupByEvicting(rx.functions.Func1<? super T,? extends K> keySelector,
                                                                                                         rx.functions.Func1<? super T,? extends R> elementSelector,
                                                                                                         rx.functions.Func1<rx.functions.Action1<K>,Map<K,Object>> evictingMapFactory)

Groups the items emitted by an Observable according to a specified criterion, and emits these grouped items as GroupedObservables. The emitted GroupedObservable allows only a single Subscriber during its lifetime and if this Subscriber unsubscribes before the source terminates, the next emission by the source having the same key will trigger a new GroupedObservable emission.

Note: A GroupedObservable will cache the items it is to emit until such time as it is subscribed to. For this reason, in order to avoid memory leaks, you should not simply ignore those GroupedObservables that do not concern you. Instead, you can signal to them that they may discard their buffers by applying an operator like .ignoreElements() to them.

Scheduler:

groupBy does not operate by default on a particular Scheduler.

Type Parameters:

T - the type of the input observable

K - the key type

R - the element type

Parameters:

keySelector - a function that extracts the key for each item

elementSelector - a function that extracts the return element for each item

evictingMapFactory - a function that given an eviction action returns a Map instance that will be used to assign items to the appropriate GroupedObservables. The Map instance must be thread-safe and any eviction must trigger a call to the supplied action (synchronously or asynchronously). This can be used to limit the size of the map by evicting keys by maximum size or access time for instance. If evictingMapFactory is null then no eviction strategy will be applied (and a suitable default thread-safe implementation of Map will be supplied). Here's an example using Guava's CacheBuilder from v19.0:

            
            Func1<Action1<K>, Map<K, Object>> mapFactory 
              = action -> CacheBuilder.newBuilder()
                  .maximumSize(1000)
                  .expireAfterAccess(12, TimeUnit.HOUR)
                  .removalListener(key -> action.call(key))
                  .<K, Object> build().asMap();
            
            

Returns:

an Observable that emits GroupedObservables, each of which corresponds to a unique key value and each of which emits those items from the source Observable that share that key value

See Also:

ReactiveX operators documentation: GroupBy

delayFinalUnsubscribe

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

If multiple concurrently open subscriptions happen to a source transformed by this method then an additional do-nothing subscription will be maintained to the source and will only be closed after the specified duration has passed from the final unsubscription of the open subscriptions. If another subscription happens during this wait period then the scheduled unsubscription will be cancelled.

Type Parameters:

T - generic type of stream

Parameters:

duration - duration of period to leave at least one source subscription open

unit - units for duration

Returns:

transformer

delayFinalUnsubscribe

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

If multiple concurrently open subscriptions happen to a source transformed by this method then an additional do-nothing subscription will be maintained to the source and will only be closed after the specified duration has passed from the final unsubscription of the open subscriptions. If another subscription happens during this wait period then the scheduled unsubscription will be cancelled.

Type Parameters:

T - generic type of stream

Parameters:

duration - duration of period to leave at least one source subscription open

unit - units for duration

scheduler - scheduler to use to schedule wait for unsubscribe

Returns:

transformer

removePairs

public static <T> rx.Observable.Transformer<T,T> removePairs(rx.functions.Func1<? super T,Boolean> isCandidateForFirst,
                                                             rx.functions.Func2<? super T,? super T,Boolean> remove)

Removes pairs non-recursively from a stream. Uses Transformers.stateMachine() under the covers to ensure items are emitted as soon as possible (if an item can't be in a pair then it is emitted straight away).

Type Parameters:

T - generic type of stream being transformed

Parameters:

isCandidateForFirst - returns true if item is potentially the first of a pair that we want to remove

remove - returns true if a pair should be removed

Returns:

transformed stream

onBackpressureBufferRequestLimiting

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

Rather than requesting Long.MAX_VALUE of upstream as does `Observable.onBackpressureBuffer`, this variant only requests of upstream what is requested of it. Thus an operator can be written that overproduces.

Type Parameters:

T - the value type

Returns:

transformer that buffers on backpressure but only requests of upstream what is requested of it

bufferUntil

public static final <T> rx.Observable.Transformer<T,List<T>> bufferUntil(rx.functions.Func1<? super T,Boolean> predicate)

Buffers the elements into continuous, non-overlapping Lists where the boundary is determined by a predicate receiving each item, after being buffered, and returns true to indicate a new buffer should start.

The operator won't return an empty first or last buffer.

Backpressure Support:

This operator supports backpressure.

Scheduler:

This operator does not operate by default on a particular Scheduler.

Type Parameters:

T - the input value type

Parameters:

predicate - the Func1 that receives each item, after being buffered, and should return true to indicate a new buffer has to start.

Returns:

the new Observable instance

Since:

(if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)

See Also:

bufferWhile(Func1)

toListUntil

public static final <T> rx.Observable.Transformer<T,List<T>> toListUntil(rx.functions.Func1<? super T,Boolean> predicate)

Buffers the elements into continuous, non-overlapping Lists where the boundary is determined by a predicate receiving each item, after being buffered, and returns true to indicate a new buffer should start.

The operator won't return an empty first or last buffer.

Backpressure Support:

This operator supports backpressure.

Scheduler:

This operator does not operate by default on a particular Scheduler.

Type Parameters:

T - the input value type

Parameters:

predicate - the Func1 that receives each item, after being buffered, and should return true to indicate a new buffer has to start.

Returns:

the new Observable instance

Since:

(if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)

See Also:

bufferWhile(Func1)

bufferUntil

public static final <T> rx.Observable.Transformer<T,List<T>> bufferUntil(rx.functions.Func1<? super T,Boolean> predicate,
                                                                         int capacityHint)

Buffers the elements into continuous, non-overlapping Lists where the boundary is determined by a predicate receiving each item, after being buffered, and returns true to indicate a new buffer should start.

The operator won't return an empty first or last buffer.

Backpressure Support:

This operator supports backpressure.

Scheduler:

This operator does not operate by default on a particular Scheduler.

Type Parameters:

T - the input value type

Parameters:

predicate - the Func1 that receives each item, after being buffered, and should return true to indicate a new buffer has to start.

capacityHint - the expected number of items in each buffer

Returns:

the new Observable instance

Since:

(if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)

See Also:

bufferWhile(Func1)

toListUntil

public static final <T> rx.Observable.Transformer<T,List<T>> toListUntil(rx.functions.Func1<? super T,Boolean> predicate,
                                                                         int capacityHint)

Buffers the elements into continuous, non-overlapping Lists where the boundary is determined by a predicate receiving each item, after being buffered, and returns true to indicate a new buffer should start.

The operator won't return an empty first or last buffer.

Backpressure Support:

This operator supports backpressure.

Scheduler:

This operator does not operate by default on a particular Scheduler.

Type Parameters:

T - the input value type

Parameters:

predicate - the Func1 that receives each item, after being buffered, and should return true to indicate a new buffer has to start.

capacityHint - the expected number of items in each buffer

Returns:

the new Observable instance

Since:

(if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)

See Also:

bufferWhile(Func1)

bufferWhile

public static final <T> rx.Observable.Transformer<T,List<T>> bufferWhile(rx.functions.Func1<? super T,Boolean> predicate)

Buffers the elements into continuous, non-overlapping Lists where the boundary is determined by a predicate receiving each item, before or after being buffered, and returns true to indicate a new buffer should start.

The operator won't return an empty first or last buffer.

Backpressure Support:

This operator supports backpressure.

Scheduler:

This operator does not operate by default on a particular Scheduler.

Type Parameters:

T - the input value type

Parameters:

predicate - the Func1 that receives each item, before being buffered, and should return true to indicate a new buffer has to start.

Returns:

the new Observable instance

Since:

(if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)

See Also:

bufferWhile(Func1)

toListWhile

public static final <T> rx.Observable.Transformer<T,List<T>> toListWhile(rx.functions.Func1<? super T,Boolean> predicate)

Buffers the elements into continuous, non-overlapping Lists where the boundary is determined by a predicate receiving each item, before or after being buffered, and returns true to indicate a new buffer should start.

The operator won't return an empty first or last buffer.

Backpressure Support:

This operator supports backpressure.

Scheduler:

This operator does not operate by default on a particular Scheduler.

Type Parameters:

T - the input value type

Parameters:

predicate - the Func1 that receives each item, before being buffered, and should return true to indicate a new buffer has to start.

Returns:

the new Observable instance

Since:

(if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)

See Also:

bufferWhile(Func1)

bufferWhile

public static final <T> rx.Observable.Transformer<T,List<T>> bufferWhile(rx.functions.Func1<? super T,Boolean> predicate,
                                                                         int capacityHint)

Buffers the elements into continuous, non-overlapping Lists where the boundary is determined by a predicate receiving each item, before being buffered, and returns true to indicate a new buffer should start.

The operator won't return an empty first or last buffer.

Backpressure Support:

This operator supports backpressure.

Scheduler:

This operator does not operate by default on a particular Scheduler.

Type Parameters:

T - the input value type

Parameters:

predicate - the Func1 that receives each item, before being buffered, and should return true to indicate a new buffer has to start.

capacityHint - the expected number of items in each buffer

Returns:

the new Observable instance

Since:

(if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)

See Also:

bufferWhile(Func1)

toListWhile

public static final <T> rx.Observable.Transformer<T,List<T>> toListWhile(rx.functions.Func1<? super T,Boolean> predicate,
                                                                         int capacityHint)

Buffers the elements into continuous, non-overlapping Lists where the boundary is determined by a predicate receiving each item, before being buffered, and returns true to indicate a new buffer should start.

The operator won't return an empty first or last buffer.

Backpressure Support:

This operator supports backpressure.

Scheduler:

This operator does not operate by default on a particular Scheduler.

Type Parameters:

T - the input value type

Parameters:

predicate - the Func1 that receives each item, before being buffered, and should return true to indicate a new buffer has to start.

capacityHint - the expected number of items in each buffer

Returns:

the new Observable instance

Since:

(if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)

See Also:

bufferWhile(Func1)

delay

public static final <T> rx.Observable.Transformer<T,T> delay(rx.functions.Func1<? super T,Long> time,
                                                             rx.functions.Func0<Double> playRate,
                                                             long startTime,
                                                             rx.Scheduler scheduler)

doOnEmpty

public static final <T> rx.Observable.Transformer<T,T> doOnEmpty(rx.functions.Action0 onEmpty)

Modifies the source Observable so that it invokes an action when it calls onCompleted and no items were emitted.

Scheduler:

doOnEmpty does not operate by default on a particular Scheduler.

Type Parameters:

T - generic type of observable being transformed

Parameters:

onEmpty - the action to invoke when the source Observable calls onCompleted, contingent on no items were emitted

Returns:

the source Observable with the side-effecting behavior applied

onTerminateResume

public static final <T> rx.Observable.Transformer<T,T> onTerminateResume(rx.functions.Func1<Throwable,rx.Observable<T>> onError,
                                                                         rx.Observable<T> onCompleted)

repeatLast

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

mapLast

public static <T> rx.Observable.Transformer<T,T> mapLast(rx.functions.Func1<? super T,? extends T> function)