Package rx.observables

Class SyncOnSubscribe<S,T>

java.lang.Object

rx.observables.SyncOnSubscribe<S,T>

Type Parameters:

S - the type of the user-define state used in generateState(S) , next(S, Subscriber), and onUnsubscribe(S).

T - the type of Subscribers that will be compatible with this.

All Implemented Interfaces:

Action, Action1<Subscriber<? super T>>, Function, Observable.OnSubscribe<T>

Direct Known Subclasses:

SyncOnSubscribe.SyncOnSubscribeImpl

@Beta
public abstract class SyncOnSubscribe<S,T>
extends java.lang.Object
implements Observable.OnSubscribe<T>

A utility class to create OnSubscribe<T> functions that respond correctly to back pressure requests from subscribers. This is an improvement over Observable.create(OnSubscribe) which does not provide any means of managing back pressure requests out-of-the-box.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
(package private) static class	SyncOnSubscribe.SubscriptionProducer<S,T>	Contains the producer loop that reacts to downstream requests of work.
(package private) static class	SyncOnSubscribe.SyncOnSubscribeImpl<S,T>	An implementation of SyncOnSubscribe that delegates SyncOnSubscribe#next(Object, Subscriber), generateState(), and onUnsubscribe(Object) to provided functions/closures.

Constructor Summary

Constructors

Constructor	Description
SyncOnSubscribe()

Method Summary

Modifier and Type	Method	Description
void	call(Subscriber<? super T> subscriber)
static <S,T> SyncOnSubscribe<S,T>	createSingleState(Func0<? extends S> generator, Action2<? super S,? super Observer<? super T>> next)	Generates a synchronous SyncOnSubscribe that calls the provided next function to generate data to downstream subscribers.
static <S,T> SyncOnSubscribe<S,T>	createSingleState(Func0<? extends S> generator, Action2<? super S,? super Observer<? super T>> next, Action1<? super S> onUnsubscribe)	Generates a synchronous SyncOnSubscribe that calls the provided next function to generate data to downstream subscribers.
static <S,T> SyncOnSubscribe<S,T>	createStateful(Func0<? extends S> generator, Func2<? super S,? super Observer<? super T>,? extends S> next)	Generates a synchronous SyncOnSubscribe that calls the provided next function to generate data to downstream subscribers.
static <S,T> SyncOnSubscribe<S,T>	createStateful(Func0<? extends S> generator, Func2<? super S,? super Observer<? super T>,? extends S> next, Action1<? super S> onUnsubscribe)	Generates a synchronous SyncOnSubscribe that calls the provided next function to generate data to downstream subscribers.
static <T> SyncOnSubscribe<java.lang.Void,T>	createStateless(Action1<? super Observer<? super T>> next)	Generates a synchronous SyncOnSubscribe that calls the provided next function to generate data to downstream subscribers.
static <T> SyncOnSubscribe<java.lang.Void,T>	createStateless(Action1<? super Observer<? super T>> next, Action0 onUnsubscribe)	Generates a synchronous SyncOnSubscribe that calls the provided next function to generate data to downstream subscribers.
protected abstract S	generateState()	Executed once when subscribed to by a subscriber (via call(Subscriber)) to produce a state value.
protected abstract S	next(S state, Observer<? super T> observer)	Called to produce data to the downstream subscribers.
protected void	onUnsubscribe(S state)	Clean up behavior that is executed after the downstream subscriber's subscription is unsubscribed.

Methods inherited from class java.lang.Object

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

Constructor Detail

SyncOnSubscribe

public SyncOnSubscribe()

Method Detail

call

public final void call(Subscriber<? super T> subscriber)

Specified by:

call in interface Action1<S>

generateState

protected abstract S generateState()

Executed once when subscribed to by a subscriber (via call(Subscriber)) to produce a state value. This value is passed into next(S state, Observer observer) on the first iteration. Subsequent iterations of next will receive the state returned by the previous invocation of next.

Returns:

the initial state value

next

protected abstract S next(S state,
                          Observer<? super T> observer)

Called to produce data to the downstream subscribers. To emit data to a downstream subscriber call observer.onNext(t). To signal an error condition call observer.onError(throwable) or throw an Exception. To signal the end of a data stream call observer.onCompleted(). Implementations of this method must follow the following rules.

• Must not call observer.onNext(t) more than 1 time per invocation.
• Must not call observer.onNext(t) concurrently.

The value returned from an invocation of this method will be passed in as the state argument of the next invocation of this method.

Parameters:

state - the state value (from generateState() on the first invocation or the previous invocation of this method.

observer - the observer of data emitted by

Returns:

the next iteration's state value

onUnsubscribe

protected void onUnsubscribe(S state)

Clean up behavior that is executed after the downstream subscriber's subscription is unsubscribed. This method will be invoked exactly once.

Parameters:

state - the last state value prior from generateState() or next(S, Observer<T>) before unsubscribe.

createSingleState

@Beta
public static <S,T> SyncOnSubscribe<S,T> createSingleState(Func0<? extends S> generator,
                                                                       Action2<? super S,? super Observer<? super T>> next)

Generates a synchronous SyncOnSubscribe that calls the provided next function to generate data to downstream subscribers.

Type Parameters:

T - the type of the generated values

S - the type of the associated state with each Subscriber

Parameters:

generator - generates the initial state value (see generateState())

next - produces data to the downstream subscriber (see next(S, Subscriber))

Returns:

a SyncOnSubscribe that emits data in a protocol compatible with back-pressure.

createSingleState

@Beta
public static <S,T> SyncOnSubscribe<S,T> createSingleState(Func0<? extends S> generator,
                                                                       Action2<? super S,? super Observer<? super T>> next,
                                                                       Action1<? super S> onUnsubscribe)

Generates a synchronous SyncOnSubscribe that calls the provided next function to generate data to downstream subscribers. This overload creates a SyncOnSubscribe without an explicit clean up step.

Type Parameters:

T - the type of the generated values

S - the type of the associated state with each Subscriber

Parameters:

generator - generates the initial state value (see generateState())

next - produces data to the downstream subscriber (see next(S, Subscriber))

onUnsubscribe - clean up behavior (see onUnsubscribe(S))

Returns:

a SyncOnSubscribe that emits data downstream in a protocol compatible with back-pressure.

createStateful

@Beta
public static <S,T> SyncOnSubscribe<S,T> createStateful(Func0<? extends S> generator,
                                                                    Func2<? super S,? super Observer<? super T>,? extends S> next,
                                                                    Action1<? super S> onUnsubscribe)

Generates a synchronous SyncOnSubscribe that calls the provided next function to generate data to downstream subscribers.

Type Parameters:

T - the type of the generated values

S - the type of the associated state with each Subscriber

Parameters:

generator - generates the initial state value (see generateState())

next - produces data to the downstream subscriber (see next(S, Subscriber))

onUnsubscribe - clean up behavior (see onUnsubscribe(S))

Returns:

a SyncOnSubscribe that emits data downstream in a protocol compatible with back-pressure.

createStateful

@Beta
public static <S,T> SyncOnSubscribe<S,T> createStateful(Func0<? extends S> generator,
                                                                    Func2<? super S,? super Observer<? super T>,? extends S> next)

Generates a synchronous SyncOnSubscribe that calls the provided next function to generate data to downstream subscribers.

Type Parameters:

T - the type of the generated values

S - the type of the associated state with each Subscriber

Parameters:

generator - generates the initial state value (see generateState())

next - produces data to the downstream subscriber (see next(S, Subscriber))

Returns:

a SyncOnSubscribe that emits data downstream in a protocol compatible with back-pressure.

createStateless

@Beta
public static <T> SyncOnSubscribe<java.lang.Void,T> createStateless(Action1<? super Observer<? super T>> next)

Generates a synchronous SyncOnSubscribe that calls the provided next function to generate data to downstream subscribers. This overload creates a "state-less" SyncOnSubscribe which does not have an explicit state value. This should be used when the next function closes over it's state.

Type Parameters:

T - the type of the generated values

Parameters:

next - produces data to the downstream subscriber (see next(S, Subscriber))

Returns:

a SyncOnSubscribe that emits data downstream in a protocol compatible with back-pressure.

createStateless

@Beta
public static <T> SyncOnSubscribe<java.lang.Void,T> createStateless(Action1<? super Observer<? super T>> next,
                                                                          Action0 onUnsubscribe)

Generates a synchronous SyncOnSubscribe that calls the provided next function to generate data to downstream subscribers. This overload creates a "state-less" SyncOnSubscribe which does not have an explicit state value. This should be used when the next function closes over it's state.

Type Parameters:

T - the type of the generated values

Parameters:

next - produces data to the downstream subscriber (see next(S, Subscriber))

onUnsubscribe - clean up behavior (see onUnsubscribe(S))

Returns:

a SyncOnSubscribe that emits data downstream in a protocol compatible with back-pressure.