114 lines
3.6 KiB
Java
114 lines
3.6 KiB
Java
package io.reactivex.internal.subscribers;
|
|
|
|
import io.reactivex.FlowableSubscriber;
|
|
import io.reactivex.internal.fuseable.QueueSubscription;
|
|
import io.reactivex.internal.fuseable.SimpleQueue;
|
|
import io.reactivex.internal.subscriptions.SubscriptionHelper;
|
|
import io.reactivex.internal.util.QueueDrainHelper;
|
|
import java.util.concurrent.atomic.AtomicReference;
|
|
import org.reactivestreams.Subscription;
|
|
|
|
/* loaded from: classes2.dex */
|
|
public final class InnerQueuedSubscriber<T> extends AtomicReference<Subscription> implements FlowableSubscriber<T>, Subscription {
|
|
private static final long serialVersionUID = 22876611072430776L;
|
|
volatile boolean done;
|
|
int fusionMode;
|
|
final int limit;
|
|
final InnerQueuedSubscriberSupport<T> parent;
|
|
final int prefetch;
|
|
long produced;
|
|
volatile SimpleQueue<T> queue;
|
|
|
|
public InnerQueuedSubscriber(InnerQueuedSubscriberSupport<T> innerQueuedSubscriberSupport, int i) {
|
|
this.parent = innerQueuedSubscriberSupport;
|
|
this.prefetch = i;
|
|
this.limit = i - (i >> 2);
|
|
}
|
|
|
|
@Override // org.reactivestreams.Subscription
|
|
public void cancel() {
|
|
SubscriptionHelper.cancel(this);
|
|
}
|
|
|
|
public boolean isDone() {
|
|
return this.done;
|
|
}
|
|
|
|
@Override // org.reactivestreams.Subscriber
|
|
public void onComplete() {
|
|
this.parent.a(this);
|
|
}
|
|
|
|
@Override // org.reactivestreams.Subscriber
|
|
public void onError(Throwable th) {
|
|
this.parent.a((InnerQueuedSubscriber) this, th);
|
|
}
|
|
|
|
@Override // org.reactivestreams.Subscriber
|
|
public void onNext(T t) {
|
|
if (this.fusionMode == 0) {
|
|
this.parent.a((InnerQueuedSubscriber<InnerQueuedSubscriber<T>>) this, (InnerQueuedSubscriber<T>) t);
|
|
} else {
|
|
this.parent.a();
|
|
}
|
|
}
|
|
|
|
@Override // org.reactivestreams.Subscriber
|
|
public void onSubscribe(Subscription subscription) {
|
|
if (SubscriptionHelper.setOnce(this, subscription)) {
|
|
if (subscription instanceof QueueSubscription) {
|
|
QueueSubscription queueSubscription = (QueueSubscription) subscription;
|
|
int requestFusion = queueSubscription.requestFusion(3);
|
|
if (requestFusion == 1) {
|
|
this.fusionMode = requestFusion;
|
|
this.queue = queueSubscription;
|
|
this.done = true;
|
|
this.parent.a(this);
|
|
return;
|
|
}
|
|
if (requestFusion == 2) {
|
|
this.fusionMode = requestFusion;
|
|
this.queue = queueSubscription;
|
|
QueueDrainHelper.a(subscription, this.prefetch);
|
|
return;
|
|
}
|
|
}
|
|
this.queue = QueueDrainHelper.a(this.prefetch);
|
|
QueueDrainHelper.a(subscription, this.prefetch);
|
|
}
|
|
}
|
|
|
|
public SimpleQueue<T> queue() {
|
|
return this.queue;
|
|
}
|
|
|
|
@Override // org.reactivestreams.Subscription
|
|
public void request(long j) {
|
|
if (this.fusionMode != 1) {
|
|
long j2 = this.produced + j;
|
|
if (j2 < this.limit) {
|
|
this.produced = j2;
|
|
} else {
|
|
this.produced = 0L;
|
|
get().request(j2);
|
|
}
|
|
}
|
|
}
|
|
|
|
public void requestOne() {
|
|
if (this.fusionMode != 1) {
|
|
long j = this.produced + 1;
|
|
if (j != this.limit) {
|
|
this.produced = j;
|
|
} else {
|
|
this.produced = 0L;
|
|
get().request(j);
|
|
}
|
|
}
|
|
}
|
|
|
|
public void setDone() {
|
|
this.done = true;
|
|
}
|
|
}
|