Package io.reactivex.rxjava3.internal.subscribers

Class BasicFuseableSubscriber<T,R>

java.lang.Object

io.reactivex.rxjava3.internal.subscribers.BasicFuseableSubscriber<T,R>

Type Parameters:

T - the upstream value type

R - the downstream value type

All Implemented Interfaces:

FlowableSubscriber<T>, QueueFuseable<R>, QueueSubscription<R>, SimpleQueue<R>, org.reactivestreams.Subscriber<T>, org.reactivestreams.Subscription

Direct Known Subclasses:

FlowableDistinct.DistinctSubscriber, FlowableDistinctUntilChanged.DistinctUntilChangedSubscriber, FlowableDoAfterNext.DoAfterSubscriber, FlowableDoOnEach.DoOnEachSubscriber, FlowableFilter.FilterSubscriber, FlowableMap.MapSubscriber, FlowableMapOptional.MapOptionalSubscriber

public abstract class BasicFuseableSubscriber<T,R>
extends java.lang.Object
implements FlowableSubscriber<T>, QueueSubscription<R>

Base class for a fuseable intermediate subscriber.

Field Summary

Fields

Modifier and Type	Field	Description
protected boolean	done	Flag indicating no further onXXX event should be accepted.
protected org.reactivestreams.Subscriber<? super R>	downstream	The downstream subscriber.
protected QueueSubscription<T>	qs	The upstream's QueueSubscription if not null.
protected int	sourceMode	Holds the established fusion mode of the upstream.
protected org.reactivestreams.Subscription	upstream	The upstream subscription.

Fields inherited from interface io.reactivex.rxjava3.operators.QueueFuseable

ANY, ASYNC, BOUNDARY, NONE, SYNC

Constructor Summary

Constructors

Constructor	Description
BasicFuseableSubscriber(org.reactivestreams.Subscriber<? super R> downstream)	Construct a BasicFuseableSubscriber by wrapping the given subscriber.

Method Summary

Modifier and Type	Method	Description
protected void	afterDownstream()	Override this to perform actions after the call to actual.onSubscribe(this) happened.
protected boolean	beforeDownstream()	Override this to perform actions before the call actual.onSubscribe(this) happens.
void	cancel()
void	clear()	Removes all enqueued items from this queue.
protected void	fail(java.lang.Throwable t)	Rethrows the throwable if it is a fatal exception or calls onError(Throwable).
boolean	isEmpty()	Returns true if the queue is empty.
boolean	offer(R e)	Atomically enqueue a single value.
boolean	offer(R v1, R v2)	Atomically enqueue two values.
void	onComplete()
void	onError(java.lang.Throwable t)
void	onSubscribe(org.reactivestreams.Subscription s)	Implementors of this method should make sure everything that needs to be visible in Subscriber.onNext(Object) is established before calling Subscription.request(long).
void	request(long n)
protected int	transitiveBoundaryFusion(int mode)	Calls the upstream's QueueSubscription.requestFusion with the mode and saves the established mode in sourceMode if that mode doesn't have the QueueFuseable.BOUNDARY flag set.

Methods inherited from class java.lang.Object

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

Methods inherited from interface io.reactivex.rxjava3.operators.QueueFuseable

requestFusion

Methods inherited from interface io.reactivex.rxjava3.operators.SimpleQueue

poll

Methods inherited from interface org.reactivestreams.Subscriber

onNext

Field Detail

downstream

protected final org.reactivestreams.Subscriber<? super R> downstream

The downstream subscriber.

upstream

protected org.reactivestreams.Subscription upstream

The upstream subscription.

qs

protected QueueSubscription<T> qs

The upstream's QueueSubscription if not null.

done

protected boolean done

Flag indicating no further onXXX event should be accepted.

sourceMode

protected int sourceMode

Holds the established fusion mode of the upstream.

Constructor Detail

BasicFuseableSubscriber

public BasicFuseableSubscriber(org.reactivestreams.Subscriber<? super R> downstream)

Construct a BasicFuseableSubscriber by wrapping the given subscriber.

Parameters:

downstream - the subscriber, not null (not verified)

Method Detail

onSubscribe

public final void onSubscribe(org.reactivestreams.Subscription s)

Description copied from interface: FlowableSubscriber

Implementors of this method should make sure everything that needs to be visible in Subscriber.onNext(Object) is established before calling Subscription.request(long). In practice this means no initialization should happen after the request() call and additional behavior is thread safe in respect to onNext.

Specified by:

onSubscribe in interface FlowableSubscriber<T>

Specified by:

onSubscribe in interface org.reactivestreams.Subscriber<T>

beforeDownstream

protected boolean beforeDownstream()

Override this to perform actions before the call actual.onSubscribe(this) happens.

Returns:

true if onSubscribe should continue with the call

afterDownstream

protected void afterDownstream()

Override this to perform actions after the call to actual.onSubscribe(this) happened.

onError

public void onError(java.lang.Throwable t)

Specified by:

onError in interface org.reactivestreams.Subscriber<T>

fail

protected final void fail(java.lang.Throwable t)

Rethrows the throwable if it is a fatal exception or calls onError(Throwable).

Parameters:

t - the throwable to rethrow or signal to the actual subscriber

onComplete

public void onComplete()

Specified by:

onComplete in interface org.reactivestreams.Subscriber<T>

transitiveBoundaryFusion

protected final int transitiveBoundaryFusion(int mode)

Calls the upstream's QueueSubscription.requestFusion with the mode and saves the established mode in sourceMode if that mode doesn't have the QueueFuseable.BOUNDARY flag set.

If the upstream doesn't support fusion (qs is null), the method returns QueueFuseable.NONE.

Parameters:

mode - the fusion mode requested

Returns:

the established fusion mode

request

public void request(long n)

Specified by:

request in interface org.reactivestreams.Subscription

cancel

public void cancel()

Specified by:

cancel in interface org.reactivestreams.Subscription

isEmpty

public boolean isEmpty()

Description copied from interface: SimpleQueue

Returns true if the queue is empty.

Note however that due to potential fused functions in SimpleQueue.poll() it is possible this method returns false but then poll() returns null because the fused function swallowed the available item(s).

Specified by:

isEmpty in interface SimpleQueue<T>

Returns:

true if the queue is empty

clear

public void clear()

Description copied from interface: SimpleQueue

Removes all enqueued items from this queue.

Specified by:

clear in interface SimpleQueue<T>

offer

public final boolean offer(R e)

Description copied from interface: SimpleQueue

Atomically enqueue a single value.

Specified by:

offer in interface SimpleQueue<T>

Parameters:

e - the value to enqueue, not null

Returns:

true if successful, false if the value was not enqueued likely due to reaching the queue capacity)

offer

public final boolean offer(R v1,
                           R v2)

Description copied from interface: SimpleQueue

Atomically enqueue two values.

Specified by:

offer in interface SimpleQueue<T>

Parameters:

v1 - the first value to enqueue, not null

v2 - the second value to enqueue, not null

Returns:

true if successful, false if the value was not enqueued likely due to reaching the queue capacity)