305 lines
10 KiB
Java
305 lines
10 KiB
Java
package io.reactivex.internal.operators.observable;
|
|
|
|
import io.reactivex.ObservableSource;
|
|
import io.reactivex.Observer;
|
|
import io.reactivex.disposables.Disposable;
|
|
import io.reactivex.exceptions.Exceptions;
|
|
import io.reactivex.functions.Function;
|
|
import io.reactivex.internal.disposables.DisposableHelper;
|
|
import io.reactivex.internal.disposables.EmptyDisposable;
|
|
import io.reactivex.internal.functions.ObjectHelper;
|
|
import io.reactivex.internal.queue.SpscLinkedArrayQueue;
|
|
import io.reactivex.observables.GroupedObservable;
|
|
import java.util.ArrayList;
|
|
import java.util.Iterator;
|
|
import java.util.Map;
|
|
import java.util.concurrent.ConcurrentHashMap;
|
|
import java.util.concurrent.atomic.AtomicBoolean;
|
|
import java.util.concurrent.atomic.AtomicInteger;
|
|
import java.util.concurrent.atomic.AtomicReference;
|
|
|
|
/* loaded from: classes2.dex */
|
|
public final class ObservableGroupBy<T, K, V> extends AbstractObservableWithUpstream<T, GroupedObservable<K, V>> {
|
|
final Function<? super T, ? extends K> b;
|
|
final Function<? super T, ? extends V> c;
|
|
final int d;
|
|
final boolean e;
|
|
|
|
public static final class GroupByObserver<T, K, V> extends AtomicInteger implements Observer<T>, Disposable {
|
|
static final Object NULL_KEY = new Object();
|
|
private static final long serialVersionUID = -3688291656102519502L;
|
|
final int bufferSize;
|
|
final boolean delayError;
|
|
final Observer<? super GroupedObservable<K, V>> downstream;
|
|
final Function<? super T, ? extends K> keySelector;
|
|
Disposable upstream;
|
|
final Function<? super T, ? extends V> valueSelector;
|
|
final AtomicBoolean cancelled = new AtomicBoolean();
|
|
final Map<Object, GroupedUnicast<K, V>> groups = new ConcurrentHashMap();
|
|
|
|
public GroupByObserver(Observer<? super GroupedObservable<K, V>> observer, Function<? super T, ? extends K> function, Function<? super T, ? extends V> function2, int i, boolean z) {
|
|
this.downstream = observer;
|
|
this.keySelector = function;
|
|
this.valueSelector = function2;
|
|
this.bufferSize = i;
|
|
this.delayError = z;
|
|
lazySet(1);
|
|
}
|
|
|
|
public void cancel(K k) {
|
|
if (k == null) {
|
|
k = (K) NULL_KEY;
|
|
}
|
|
this.groups.remove(k);
|
|
if (decrementAndGet() == 0) {
|
|
this.upstream.dispose();
|
|
}
|
|
}
|
|
|
|
@Override // io.reactivex.disposables.Disposable
|
|
public void dispose() {
|
|
if (this.cancelled.compareAndSet(false, true) && decrementAndGet() == 0) {
|
|
this.upstream.dispose();
|
|
}
|
|
}
|
|
|
|
public boolean isDisposed() {
|
|
return this.cancelled.get();
|
|
}
|
|
|
|
@Override // io.reactivex.Observer
|
|
public void onComplete() {
|
|
ArrayList arrayList = new ArrayList(this.groups.values());
|
|
this.groups.clear();
|
|
Iterator it = arrayList.iterator();
|
|
while (it.hasNext()) {
|
|
((GroupedUnicast) it.next()).onComplete();
|
|
}
|
|
this.downstream.onComplete();
|
|
}
|
|
|
|
@Override // io.reactivex.Observer
|
|
public void onError(Throwable th) {
|
|
ArrayList arrayList = new ArrayList(this.groups.values());
|
|
this.groups.clear();
|
|
Iterator it = arrayList.iterator();
|
|
while (it.hasNext()) {
|
|
((GroupedUnicast) it.next()).onError(th);
|
|
}
|
|
this.downstream.onError(th);
|
|
}
|
|
|
|
@Override // io.reactivex.Observer
|
|
public void onNext(T t) {
|
|
try {
|
|
K apply = this.keySelector.apply(t);
|
|
Object obj = apply != null ? apply : NULL_KEY;
|
|
GroupedUnicast<K, V> groupedUnicast = this.groups.get(obj);
|
|
if (groupedUnicast == null) {
|
|
if (this.cancelled.get()) {
|
|
return;
|
|
}
|
|
groupedUnicast = GroupedUnicast.a(apply, this.bufferSize, this, this.delayError);
|
|
this.groups.put(obj, groupedUnicast);
|
|
getAndIncrement();
|
|
this.downstream.onNext(groupedUnicast);
|
|
}
|
|
try {
|
|
V apply2 = this.valueSelector.apply(t);
|
|
ObjectHelper.a(apply2, "The value supplied is null");
|
|
groupedUnicast.onNext(apply2);
|
|
} catch (Throwable th) {
|
|
Exceptions.b(th);
|
|
this.upstream.dispose();
|
|
onError(th);
|
|
}
|
|
} catch (Throwable th2) {
|
|
Exceptions.b(th2);
|
|
this.upstream.dispose();
|
|
onError(th2);
|
|
}
|
|
}
|
|
|
|
@Override // io.reactivex.Observer
|
|
public void onSubscribe(Disposable disposable) {
|
|
if (DisposableHelper.validate(this.upstream, disposable)) {
|
|
this.upstream = disposable;
|
|
this.downstream.onSubscribe(this);
|
|
}
|
|
}
|
|
}
|
|
|
|
static final class GroupedUnicast<K, T> extends GroupedObservable<K, T> {
|
|
final State<T, K> a;
|
|
|
|
protected GroupedUnicast(K k, State<T, K> state) {
|
|
super(k);
|
|
this.a = state;
|
|
}
|
|
|
|
public static <T, K> GroupedUnicast<K, T> a(K k, int i, GroupByObserver<?, K, T> groupByObserver, boolean z) {
|
|
return new GroupedUnicast<>(k, new State(i, groupByObserver, k, z));
|
|
}
|
|
|
|
public void onComplete() {
|
|
this.a.b();
|
|
}
|
|
|
|
public void onError(Throwable th) {
|
|
this.a.a(th);
|
|
}
|
|
|
|
public void onNext(T t) {
|
|
this.a.a((State<T, K>) t);
|
|
}
|
|
|
|
@Override // io.reactivex.Observable
|
|
protected void subscribeActual(Observer<? super T> observer) {
|
|
this.a.subscribe(observer);
|
|
}
|
|
}
|
|
|
|
public ObservableGroupBy(ObservableSource<T> observableSource, Function<? super T, ? extends K> function, Function<? super T, ? extends V> function2, int i, boolean z) {
|
|
super(observableSource);
|
|
this.b = function;
|
|
this.c = function2;
|
|
this.d = i;
|
|
this.e = z;
|
|
}
|
|
|
|
@Override // io.reactivex.Observable
|
|
public void subscribeActual(Observer<? super GroupedObservable<K, V>> observer) {
|
|
this.a.subscribe(new GroupByObserver(observer, this.b, this.c, this.d, this.e));
|
|
}
|
|
|
|
static final class State<T, K> extends AtomicInteger implements Disposable, ObservableSource<T> {
|
|
final K a;
|
|
final SpscLinkedArrayQueue<T> b;
|
|
final GroupByObserver<?, K, T> c;
|
|
final boolean d;
|
|
volatile boolean e;
|
|
Throwable f;
|
|
final AtomicBoolean g = new AtomicBoolean();
|
|
final AtomicBoolean h = new AtomicBoolean();
|
|
final AtomicReference<Observer<? super T>> i = new AtomicReference<>();
|
|
|
|
State(int i, GroupByObserver<?, K, T> groupByObserver, K k, boolean z) {
|
|
this.b = new SpscLinkedArrayQueue<>(i);
|
|
this.c = groupByObserver;
|
|
this.a = k;
|
|
this.d = z;
|
|
}
|
|
|
|
public void a(T t) {
|
|
this.b.offer(t);
|
|
a();
|
|
}
|
|
|
|
public void b() {
|
|
this.e = true;
|
|
a();
|
|
}
|
|
|
|
@Override // io.reactivex.disposables.Disposable
|
|
public void dispose() {
|
|
if (this.g.compareAndSet(false, true) && getAndIncrement() == 0) {
|
|
this.i.lazySet(null);
|
|
this.c.cancel(this.a);
|
|
}
|
|
}
|
|
|
|
@Override // io.reactivex.ObservableSource
|
|
public void subscribe(Observer<? super T> observer) {
|
|
if (!this.h.compareAndSet(false, true)) {
|
|
EmptyDisposable.error(new IllegalStateException("Only one Observer allowed!"), observer);
|
|
return;
|
|
}
|
|
observer.onSubscribe(this);
|
|
this.i.lazySet(observer);
|
|
if (this.g.get()) {
|
|
this.i.lazySet(null);
|
|
} else {
|
|
a();
|
|
}
|
|
}
|
|
|
|
public void a(Throwable th) {
|
|
this.f = th;
|
|
this.e = true;
|
|
a();
|
|
}
|
|
|
|
void a() {
|
|
if (getAndIncrement() != 0) {
|
|
return;
|
|
}
|
|
SpscLinkedArrayQueue<T> spscLinkedArrayQueue = this.b;
|
|
boolean z = this.d;
|
|
Observer<? super T> observer = this.i.get();
|
|
int i = 1;
|
|
while (true) {
|
|
if (observer != null) {
|
|
while (true) {
|
|
boolean z2 = this.e;
|
|
T poll = spscLinkedArrayQueue.poll();
|
|
boolean z3 = poll == null;
|
|
if (a(z2, z3, observer, z)) {
|
|
return;
|
|
}
|
|
if (z3) {
|
|
break;
|
|
} else {
|
|
observer.onNext(poll);
|
|
}
|
|
}
|
|
}
|
|
i = addAndGet(-i);
|
|
if (i == 0) {
|
|
return;
|
|
}
|
|
if (observer == null) {
|
|
observer = this.i.get();
|
|
}
|
|
}
|
|
}
|
|
|
|
boolean a(boolean z, boolean z2, Observer<? super T> observer, boolean z3) {
|
|
if (this.g.get()) {
|
|
this.b.clear();
|
|
this.c.cancel(this.a);
|
|
this.i.lazySet(null);
|
|
return true;
|
|
}
|
|
if (!z) {
|
|
return false;
|
|
}
|
|
if (z3) {
|
|
if (!z2) {
|
|
return false;
|
|
}
|
|
Throwable th = this.f;
|
|
this.i.lazySet(null);
|
|
if (th != null) {
|
|
observer.onError(th);
|
|
} else {
|
|
observer.onComplete();
|
|
}
|
|
return true;
|
|
}
|
|
Throwable th2 = this.f;
|
|
if (th2 != null) {
|
|
this.b.clear();
|
|
this.i.lazySet(null);
|
|
observer.onError(th2);
|
|
return true;
|
|
}
|
|
if (!z2) {
|
|
return false;
|
|
}
|
|
this.i.lazySet(null);
|
|
observer.onComplete();
|
|
return true;
|
|
}
|
|
}
|
|
}
|