729 lines
25 KiB
Java
729 lines
25 KiB
Java
package io.reactivex.internal.operators.observable;
|
|
|
|
import io.reactivex.Observable;
|
|
import io.reactivex.ObservableSource;
|
|
import io.reactivex.Observer;
|
|
import io.reactivex.Scheduler;
|
|
import io.reactivex.disposables.Disposable;
|
|
import io.reactivex.exceptions.Exceptions;
|
|
import io.reactivex.functions.Consumer;
|
|
import io.reactivex.functions.Function;
|
|
import io.reactivex.internal.disposables.DisposableHelper;
|
|
import io.reactivex.internal.disposables.EmptyDisposable;
|
|
import io.reactivex.internal.disposables.ResettableConnectable;
|
|
import io.reactivex.internal.functions.ObjectHelper;
|
|
import io.reactivex.internal.fuseable.HasUpstreamObservableSource;
|
|
import io.reactivex.internal.util.ExceptionHelper;
|
|
import io.reactivex.internal.util.NotificationLite;
|
|
import io.reactivex.observables.ConnectableObservable;
|
|
import io.reactivex.plugins.RxJavaPlugins;
|
|
import io.reactivex.schedulers.Timed;
|
|
import java.util.ArrayList;
|
|
import java.util.concurrent.Callable;
|
|
import java.util.concurrent.TimeUnit;
|
|
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 ObservableReplay<T> extends ConnectableObservable<T> implements HasUpstreamObservableSource<T>, ResettableConnectable {
|
|
static final BufferSupplier e = new UnBoundedFactory();
|
|
final ObservableSource<T> a;
|
|
final AtomicReference<ReplayObserver<T>> b;
|
|
final BufferSupplier<T> c;
|
|
final ObservableSource<T> d;
|
|
|
|
static abstract class BoundedReplayBuffer<T> extends AtomicReference<Node> implements ReplayBuffer<T> {
|
|
Node a;
|
|
int b;
|
|
|
|
BoundedReplayBuffer() {
|
|
Node node = new Node(null);
|
|
this.a = node;
|
|
set(node);
|
|
}
|
|
|
|
Object a(Object obj) {
|
|
return obj;
|
|
}
|
|
|
|
final void a(Node node) {
|
|
this.a.set(node);
|
|
this.a = node;
|
|
this.b++;
|
|
}
|
|
|
|
Object b(Object obj) {
|
|
return obj;
|
|
}
|
|
|
|
final void b(Node node) {
|
|
set(node);
|
|
}
|
|
|
|
final void c() {
|
|
this.b--;
|
|
b(get().get());
|
|
}
|
|
|
|
final void d() {
|
|
Node node = get();
|
|
if (node.a != null) {
|
|
Node node2 = new Node(null);
|
|
node2.lazySet(node.get());
|
|
set(node2);
|
|
}
|
|
}
|
|
|
|
abstract void e();
|
|
|
|
void f() {
|
|
d();
|
|
}
|
|
|
|
@Override // io.reactivex.internal.operators.observable.ObservableReplay.ReplayBuffer
|
|
public final void next(T t) {
|
|
a(new Node(a(NotificationLite.next(t))));
|
|
e();
|
|
}
|
|
|
|
Node b() {
|
|
return get();
|
|
}
|
|
|
|
@Override // io.reactivex.internal.operators.observable.ObservableReplay.ReplayBuffer
|
|
public final void a(Throwable th) {
|
|
a(new Node(a(NotificationLite.error(th))));
|
|
f();
|
|
}
|
|
|
|
@Override // io.reactivex.internal.operators.observable.ObservableReplay.ReplayBuffer
|
|
public final void a() {
|
|
a(new Node(a(NotificationLite.complete())));
|
|
f();
|
|
}
|
|
|
|
@Override // io.reactivex.internal.operators.observable.ObservableReplay.ReplayBuffer
|
|
public final void a(InnerDisposable<T> innerDisposable) {
|
|
if (innerDisposable.getAndIncrement() != 0) {
|
|
return;
|
|
}
|
|
int i = 1;
|
|
do {
|
|
Node node = (Node) innerDisposable.a();
|
|
if (node == null) {
|
|
node = b();
|
|
innerDisposable.c = node;
|
|
}
|
|
while (!innerDisposable.b()) {
|
|
Node node2 = node.get();
|
|
if (node2 != null) {
|
|
if (NotificationLite.accept(b(node2.a), innerDisposable.b)) {
|
|
innerDisposable.c = null;
|
|
return;
|
|
}
|
|
node = node2;
|
|
} else {
|
|
innerDisposable.c = node;
|
|
i = innerDisposable.addAndGet(-i);
|
|
}
|
|
}
|
|
return;
|
|
} while (i != 0);
|
|
}
|
|
}
|
|
|
|
interface BufferSupplier<T> {
|
|
ReplayBuffer<T> call();
|
|
}
|
|
|
|
static final class DisposeConsumer<R> implements Consumer<Disposable> {
|
|
private final ObserverResourceWrapper<R> a;
|
|
|
|
DisposeConsumer(ObserverResourceWrapper<R> observerResourceWrapper) {
|
|
this.a = observerResourceWrapper;
|
|
}
|
|
|
|
@Override // io.reactivex.functions.Consumer
|
|
/* renamed from: a, reason: merged with bridge method [inline-methods] */
|
|
public void accept(Disposable disposable) {
|
|
this.a.setResource(disposable);
|
|
}
|
|
}
|
|
|
|
static final class InnerDisposable<T> extends AtomicInteger implements Disposable {
|
|
final ReplayObserver<T> a;
|
|
final Observer<? super T> b;
|
|
Object c;
|
|
volatile boolean d;
|
|
|
|
InnerDisposable(ReplayObserver<T> replayObserver, Observer<? super T> observer) {
|
|
this.a = replayObserver;
|
|
this.b = observer;
|
|
}
|
|
|
|
<U> U a() {
|
|
return (U) this.c;
|
|
}
|
|
|
|
public boolean b() {
|
|
return this.d;
|
|
}
|
|
|
|
@Override // io.reactivex.disposables.Disposable
|
|
public void dispose() {
|
|
if (this.d) {
|
|
return;
|
|
}
|
|
this.d = true;
|
|
this.a.b(this);
|
|
}
|
|
}
|
|
|
|
static final class MulticastReplay<R, U> extends Observable<R> {
|
|
private final Callable<? extends ConnectableObservable<U>> a;
|
|
private final Function<? super Observable<U>, ? extends ObservableSource<R>> b;
|
|
|
|
MulticastReplay(Callable<? extends ConnectableObservable<U>> callable, Function<? super Observable<U>, ? extends ObservableSource<R>> function) {
|
|
this.a = callable;
|
|
this.b = function;
|
|
}
|
|
|
|
@Override // io.reactivex.Observable
|
|
protected void subscribeActual(Observer<? super R> observer) {
|
|
try {
|
|
ConnectableObservable<U> call = this.a.call();
|
|
ObjectHelper.a(call, "The connectableFactory returned a null ConnectableObservable");
|
|
ConnectableObservable<U> connectableObservable = call;
|
|
ObservableSource<R> apply = this.b.apply(connectableObservable);
|
|
ObjectHelper.a(apply, "The selector returned a null ObservableSource");
|
|
ObservableSource<R> observableSource = apply;
|
|
ObserverResourceWrapper observerResourceWrapper = new ObserverResourceWrapper(observer);
|
|
observableSource.subscribe(observerResourceWrapper);
|
|
connectableObservable.a(new DisposeConsumer(observerResourceWrapper));
|
|
} catch (Throwable th) {
|
|
Exceptions.b(th);
|
|
EmptyDisposable.error(th, observer);
|
|
}
|
|
}
|
|
}
|
|
|
|
static final class Node extends AtomicReference<Node> {
|
|
final Object a;
|
|
|
|
Node(Object obj) {
|
|
this.a = obj;
|
|
}
|
|
}
|
|
|
|
static final class Replay<T> extends ConnectableObservable<T> {
|
|
private final ConnectableObservable<T> a;
|
|
private final Observable<T> b;
|
|
|
|
Replay(ConnectableObservable<T> connectableObservable, Observable<T> observable) {
|
|
this.a = connectableObservable;
|
|
this.b = observable;
|
|
}
|
|
|
|
@Override // io.reactivex.observables.ConnectableObservable
|
|
public void a(Consumer<? super Disposable> consumer) {
|
|
this.a.a(consumer);
|
|
}
|
|
|
|
@Override // io.reactivex.Observable
|
|
protected void subscribeActual(Observer<? super T> observer) {
|
|
this.b.subscribe(observer);
|
|
}
|
|
}
|
|
|
|
interface ReplayBuffer<T> {
|
|
void a();
|
|
|
|
void a(InnerDisposable<T> innerDisposable);
|
|
|
|
void a(Throwable th);
|
|
|
|
void next(T t);
|
|
}
|
|
|
|
static final class ReplayBufferSupplier<T> implements BufferSupplier<T> {
|
|
private final int a;
|
|
|
|
ReplayBufferSupplier(int i) {
|
|
this.a = i;
|
|
}
|
|
|
|
@Override // io.reactivex.internal.operators.observable.ObservableReplay.BufferSupplier
|
|
public ReplayBuffer<T> call() {
|
|
return new SizeBoundReplayBuffer(this.a);
|
|
}
|
|
}
|
|
|
|
static final class ReplayObserver<T> extends AtomicReference<Disposable> implements Observer<T>, Disposable {
|
|
static final InnerDisposable[] e = new InnerDisposable[0];
|
|
static final InnerDisposable[] f = new InnerDisposable[0];
|
|
final ReplayBuffer<T> a;
|
|
boolean b;
|
|
final AtomicReference<InnerDisposable[]> c = new AtomicReference<>(e);
|
|
final AtomicBoolean d = new AtomicBoolean();
|
|
|
|
ReplayObserver(ReplayBuffer<T> replayBuffer) {
|
|
this.a = replayBuffer;
|
|
}
|
|
|
|
public boolean a() {
|
|
return this.c.get() == f;
|
|
}
|
|
|
|
void b(InnerDisposable<T> innerDisposable) {
|
|
InnerDisposable[] innerDisposableArr;
|
|
InnerDisposable[] innerDisposableArr2;
|
|
do {
|
|
innerDisposableArr = this.c.get();
|
|
int length = innerDisposableArr.length;
|
|
if (length == 0) {
|
|
return;
|
|
}
|
|
int i = -1;
|
|
int i2 = 0;
|
|
while (true) {
|
|
if (i2 >= length) {
|
|
break;
|
|
}
|
|
if (innerDisposableArr[i2].equals(innerDisposable)) {
|
|
i = i2;
|
|
break;
|
|
}
|
|
i2++;
|
|
}
|
|
if (i < 0) {
|
|
return;
|
|
}
|
|
if (length == 1) {
|
|
innerDisposableArr2 = e;
|
|
} else {
|
|
InnerDisposable[] innerDisposableArr3 = new InnerDisposable[length - 1];
|
|
System.arraycopy(innerDisposableArr, 0, innerDisposableArr3, 0, i);
|
|
System.arraycopy(innerDisposableArr, i + 1, innerDisposableArr3, i, (length - i) - 1);
|
|
innerDisposableArr2 = innerDisposableArr3;
|
|
}
|
|
} while (!this.c.compareAndSet(innerDisposableArr, innerDisposableArr2));
|
|
}
|
|
|
|
void c() {
|
|
for (InnerDisposable<T> innerDisposable : this.c.getAndSet(f)) {
|
|
this.a.a(innerDisposable);
|
|
}
|
|
}
|
|
|
|
@Override // io.reactivex.disposables.Disposable
|
|
public void dispose() {
|
|
this.c.set(f);
|
|
DisposableHelper.dispose(this);
|
|
}
|
|
|
|
@Override // io.reactivex.Observer
|
|
public void onComplete() {
|
|
if (this.b) {
|
|
return;
|
|
}
|
|
this.b = true;
|
|
this.a.a();
|
|
c();
|
|
}
|
|
|
|
@Override // io.reactivex.Observer
|
|
public void onError(Throwable th) {
|
|
if (this.b) {
|
|
RxJavaPlugins.b(th);
|
|
return;
|
|
}
|
|
this.b = true;
|
|
this.a.a(th);
|
|
c();
|
|
}
|
|
|
|
@Override // io.reactivex.Observer
|
|
public void onNext(T t) {
|
|
if (this.b) {
|
|
return;
|
|
}
|
|
this.a.next(t);
|
|
b();
|
|
}
|
|
|
|
@Override // io.reactivex.Observer
|
|
public void onSubscribe(Disposable disposable) {
|
|
if (DisposableHelper.setOnce(this, disposable)) {
|
|
b();
|
|
}
|
|
}
|
|
|
|
boolean a(InnerDisposable<T> innerDisposable) {
|
|
InnerDisposable[] innerDisposableArr;
|
|
InnerDisposable[] innerDisposableArr2;
|
|
do {
|
|
innerDisposableArr = this.c.get();
|
|
if (innerDisposableArr == f) {
|
|
return false;
|
|
}
|
|
int length = innerDisposableArr.length;
|
|
innerDisposableArr2 = new InnerDisposable[length + 1];
|
|
System.arraycopy(innerDisposableArr, 0, innerDisposableArr2, 0, length);
|
|
innerDisposableArr2[length] = innerDisposable;
|
|
} while (!this.c.compareAndSet(innerDisposableArr, innerDisposableArr2));
|
|
return true;
|
|
}
|
|
|
|
void b() {
|
|
for (InnerDisposable<T> innerDisposable : this.c.get()) {
|
|
this.a.a(innerDisposable);
|
|
}
|
|
}
|
|
}
|
|
|
|
static final class ReplaySource<T> implements ObservableSource<T> {
|
|
private final AtomicReference<ReplayObserver<T>> a;
|
|
private final BufferSupplier<T> b;
|
|
|
|
ReplaySource(AtomicReference<ReplayObserver<T>> atomicReference, BufferSupplier<T> bufferSupplier) {
|
|
this.a = atomicReference;
|
|
this.b = bufferSupplier;
|
|
}
|
|
|
|
@Override // io.reactivex.ObservableSource
|
|
public void subscribe(Observer<? super T> observer) {
|
|
ReplayObserver<T> replayObserver;
|
|
while (true) {
|
|
replayObserver = this.a.get();
|
|
if (replayObserver != null) {
|
|
break;
|
|
}
|
|
ReplayObserver<T> replayObserver2 = new ReplayObserver<>(this.b.call());
|
|
if (this.a.compareAndSet(null, replayObserver2)) {
|
|
replayObserver = replayObserver2;
|
|
break;
|
|
}
|
|
}
|
|
InnerDisposable<T> innerDisposable = new InnerDisposable<>(replayObserver, observer);
|
|
observer.onSubscribe(innerDisposable);
|
|
replayObserver.a(innerDisposable);
|
|
if (innerDisposable.b()) {
|
|
replayObserver.b(innerDisposable);
|
|
} else {
|
|
replayObserver.a.a(innerDisposable);
|
|
}
|
|
}
|
|
}
|
|
|
|
static final class ScheduledReplaySupplier<T> implements BufferSupplier<T> {
|
|
private final int a;
|
|
private final long b;
|
|
private final TimeUnit c;
|
|
private final Scheduler d;
|
|
|
|
ScheduledReplaySupplier(int i, long j, TimeUnit timeUnit, Scheduler scheduler) {
|
|
this.a = i;
|
|
this.b = j;
|
|
this.c = timeUnit;
|
|
this.d = scheduler;
|
|
}
|
|
|
|
@Override // io.reactivex.internal.operators.observable.ObservableReplay.BufferSupplier
|
|
public ReplayBuffer<T> call() {
|
|
return new SizeAndTimeBoundReplayBuffer(this.a, this.b, this.c, this.d);
|
|
}
|
|
}
|
|
|
|
static final class SizeAndTimeBoundReplayBuffer<T> extends BoundedReplayBuffer<T> {
|
|
final Scheduler c;
|
|
final long d;
|
|
final TimeUnit e;
|
|
final int f;
|
|
|
|
SizeAndTimeBoundReplayBuffer(int i, long j, TimeUnit timeUnit, Scheduler scheduler) {
|
|
this.c = scheduler;
|
|
this.f = i;
|
|
this.d = j;
|
|
this.e = timeUnit;
|
|
}
|
|
|
|
@Override // io.reactivex.internal.operators.observable.ObservableReplay.BoundedReplayBuffer
|
|
Object a(Object obj) {
|
|
return new Timed(obj, this.c.a(this.e), this.e);
|
|
}
|
|
|
|
@Override // io.reactivex.internal.operators.observable.ObservableReplay.BoundedReplayBuffer
|
|
Object b(Object obj) {
|
|
return ((Timed) obj).b();
|
|
}
|
|
|
|
@Override // io.reactivex.internal.operators.observable.ObservableReplay.BoundedReplayBuffer
|
|
void e() {
|
|
Node node;
|
|
long a = this.c.a(this.e) - this.d;
|
|
Node node2 = get();
|
|
Node node3 = node2.get();
|
|
int i = 0;
|
|
while (true) {
|
|
Node node4 = node3;
|
|
node = node2;
|
|
node2 = node4;
|
|
if (node2 != null) {
|
|
int i2 = this.b;
|
|
if (i2 <= this.f) {
|
|
if (((Timed) node2.a).a() > a) {
|
|
break;
|
|
}
|
|
i++;
|
|
this.b--;
|
|
node3 = node2.get();
|
|
} else {
|
|
i++;
|
|
this.b = i2 - 1;
|
|
node3 = node2.get();
|
|
}
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
if (i != 0) {
|
|
b(node);
|
|
}
|
|
}
|
|
|
|
/* JADX WARN: Code restructure failed: missing block: B:11:0x003e, code lost:
|
|
|
|
b(r3);
|
|
*/
|
|
/* JADX WARN: Code restructure failed: missing block: B:12:0x0041, code lost:
|
|
|
|
return;
|
|
*/
|
|
@Override // io.reactivex.internal.operators.observable.ObservableReplay.BoundedReplayBuffer
|
|
/*
|
|
Code decompiled incorrectly, please refer to instructions dump.
|
|
To view partially-correct code enable 'Show inconsistent code' option in preferences
|
|
*/
|
|
void f() {
|
|
/*
|
|
r10 = this;
|
|
io.reactivex.Scheduler r0 = r10.c
|
|
java.util.concurrent.TimeUnit r1 = r10.e
|
|
long r0 = r0.a(r1)
|
|
long r2 = r10.d
|
|
long r0 = r0 - r2
|
|
java.lang.Object r2 = r10.get()
|
|
io.reactivex.internal.operators.observable.ObservableReplay$Node r2 = (io.reactivex.internal.operators.observable.ObservableReplay.Node) r2
|
|
java.lang.Object r3 = r2.get()
|
|
io.reactivex.internal.operators.observable.ObservableReplay$Node r3 = (io.reactivex.internal.operators.observable.ObservableReplay.Node) r3
|
|
r4 = 0
|
|
L18:
|
|
r9 = r3
|
|
r3 = r2
|
|
r2 = r9
|
|
if (r2 == 0) goto L3c
|
|
int r5 = r10.b
|
|
r6 = 1
|
|
if (r5 <= r6) goto L3c
|
|
java.lang.Object r5 = r2.a
|
|
io.reactivex.schedulers.Timed r5 = (io.reactivex.schedulers.Timed) r5
|
|
long r7 = r5.a()
|
|
int r5 = (r7 > r0 ? 1 : (r7 == r0 ? 0 : -1))
|
|
if (r5 > 0) goto L3c
|
|
int r4 = r4 + 1
|
|
int r3 = r10.b
|
|
int r3 = r3 - r6
|
|
r10.b = r3
|
|
java.lang.Object r3 = r2.get()
|
|
io.reactivex.internal.operators.observable.ObservableReplay$Node r3 = (io.reactivex.internal.operators.observable.ObservableReplay.Node) r3
|
|
goto L18
|
|
L3c:
|
|
if (r4 == 0) goto L41
|
|
r10.b(r3)
|
|
L41:
|
|
return
|
|
*/
|
|
throw new UnsupportedOperationException("Method not decompiled: io.reactivex.internal.operators.observable.ObservableReplay.SizeAndTimeBoundReplayBuffer.f():void");
|
|
}
|
|
|
|
@Override // io.reactivex.internal.operators.observable.ObservableReplay.BoundedReplayBuffer
|
|
Node b() {
|
|
Node node;
|
|
long a = this.c.a(this.e) - this.d;
|
|
Node node2 = get();
|
|
Node node3 = node2.get();
|
|
while (true) {
|
|
Node node4 = node3;
|
|
node = node2;
|
|
node2 = node4;
|
|
if (node2 != null) {
|
|
Timed timed = (Timed) node2.a;
|
|
if (NotificationLite.isComplete(timed.b()) || NotificationLite.isError(timed.b()) || timed.a() > a) {
|
|
break;
|
|
}
|
|
node3 = node2.get();
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
return node;
|
|
}
|
|
}
|
|
|
|
static final class SizeBoundReplayBuffer<T> extends BoundedReplayBuffer<T> {
|
|
final int c;
|
|
|
|
SizeBoundReplayBuffer(int i) {
|
|
this.c = i;
|
|
}
|
|
|
|
@Override // io.reactivex.internal.operators.observable.ObservableReplay.BoundedReplayBuffer
|
|
void e() {
|
|
if (this.b > this.c) {
|
|
c();
|
|
}
|
|
}
|
|
}
|
|
|
|
static final class UnBoundedFactory implements BufferSupplier<Object> {
|
|
UnBoundedFactory() {
|
|
}
|
|
|
|
@Override // io.reactivex.internal.operators.observable.ObservableReplay.BufferSupplier
|
|
public ReplayBuffer<Object> call() {
|
|
return new UnboundedReplayBuffer(16);
|
|
}
|
|
}
|
|
|
|
private ObservableReplay(ObservableSource<T> observableSource, ObservableSource<T> observableSource2, AtomicReference<ReplayObserver<T>> atomicReference, BufferSupplier<T> bufferSupplier) {
|
|
this.d = observableSource;
|
|
this.a = observableSource2;
|
|
this.b = atomicReference;
|
|
this.c = bufferSupplier;
|
|
}
|
|
|
|
public static <U, R> Observable<R> a(Callable<? extends ConnectableObservable<U>> callable, Function<? super Observable<U>, ? extends ObservableSource<R>> function) {
|
|
return RxJavaPlugins.a(new MulticastReplay(callable, function));
|
|
}
|
|
|
|
@Override // io.reactivex.Observable
|
|
protected void subscribeActual(Observer<? super T> observer) {
|
|
this.d.subscribe(observer);
|
|
}
|
|
|
|
static final class UnboundedReplayBuffer<T> extends ArrayList<Object> implements ReplayBuffer<T> {
|
|
volatile int a;
|
|
|
|
UnboundedReplayBuffer(int i) {
|
|
super(i);
|
|
}
|
|
|
|
@Override // io.reactivex.internal.operators.observable.ObservableReplay.ReplayBuffer
|
|
public void a(Throwable th) {
|
|
add(NotificationLite.error(th));
|
|
this.a++;
|
|
}
|
|
|
|
@Override // io.reactivex.internal.operators.observable.ObservableReplay.ReplayBuffer
|
|
public void next(T t) {
|
|
add(NotificationLite.next(t));
|
|
this.a++;
|
|
}
|
|
|
|
@Override // io.reactivex.internal.operators.observable.ObservableReplay.ReplayBuffer
|
|
public void a() {
|
|
add(NotificationLite.complete());
|
|
this.a++;
|
|
}
|
|
|
|
@Override // io.reactivex.internal.operators.observable.ObservableReplay.ReplayBuffer
|
|
public void a(InnerDisposable<T> innerDisposable) {
|
|
if (innerDisposable.getAndIncrement() != 0) {
|
|
return;
|
|
}
|
|
Observer<? super T> observer = innerDisposable.b;
|
|
int i = 1;
|
|
while (!innerDisposable.b()) {
|
|
int i2 = this.a;
|
|
Integer num = (Integer) innerDisposable.a();
|
|
int intValue = num != null ? num.intValue() : 0;
|
|
while (intValue < i2) {
|
|
if (NotificationLite.accept(get(intValue), observer) || innerDisposable.b()) {
|
|
return;
|
|
} else {
|
|
intValue++;
|
|
}
|
|
}
|
|
innerDisposable.c = Integer.valueOf(intValue);
|
|
i = innerDisposable.addAndGet(-i);
|
|
if (i == 0) {
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
public static <T> ConnectableObservable<T> a(ConnectableObservable<T> connectableObservable, Scheduler scheduler) {
|
|
return RxJavaPlugins.a((ConnectableObservable) new Replay(connectableObservable, connectableObservable.observeOn(scheduler)));
|
|
}
|
|
|
|
public static <T> ConnectableObservable<T> a(ObservableSource<? extends T> observableSource) {
|
|
return a(observableSource, e);
|
|
}
|
|
|
|
public static <T> ConnectableObservable<T> a(ObservableSource<T> observableSource, int i) {
|
|
if (i == Integer.MAX_VALUE) {
|
|
return a(observableSource);
|
|
}
|
|
return a(observableSource, new ReplayBufferSupplier(i));
|
|
}
|
|
|
|
public static <T> ConnectableObservable<T> a(ObservableSource<T> observableSource, long j, TimeUnit timeUnit, Scheduler scheduler) {
|
|
return a(observableSource, j, timeUnit, scheduler, Integer.MAX_VALUE);
|
|
}
|
|
|
|
public static <T> ConnectableObservable<T> a(ObservableSource<T> observableSource, long j, TimeUnit timeUnit, Scheduler scheduler, int i) {
|
|
return a(observableSource, new ScheduledReplaySupplier(i, j, timeUnit, scheduler));
|
|
}
|
|
|
|
static <T> ConnectableObservable<T> a(ObservableSource<T> observableSource, BufferSupplier<T> bufferSupplier) {
|
|
AtomicReference atomicReference = new AtomicReference();
|
|
return RxJavaPlugins.a((ConnectableObservable) new ObservableReplay(new ReplaySource(atomicReference, bufferSupplier), observableSource, atomicReference, bufferSupplier));
|
|
}
|
|
|
|
@Override // io.reactivex.internal.disposables.ResettableConnectable
|
|
public void a(Disposable disposable) {
|
|
this.b.compareAndSet((ReplayObserver) disposable, null);
|
|
}
|
|
|
|
@Override // io.reactivex.observables.ConnectableObservable
|
|
public void a(Consumer<? super Disposable> consumer) {
|
|
ReplayObserver<T> replayObserver;
|
|
while (true) {
|
|
replayObserver = this.b.get();
|
|
if (replayObserver != null && !replayObserver.a()) {
|
|
break;
|
|
}
|
|
ReplayObserver<T> replayObserver2 = new ReplayObserver<>(this.c.call());
|
|
if (this.b.compareAndSet(replayObserver, replayObserver2)) {
|
|
replayObserver = replayObserver2;
|
|
break;
|
|
}
|
|
}
|
|
boolean z = !replayObserver.d.get() && replayObserver.d.compareAndSet(false, true);
|
|
try {
|
|
consumer.accept(replayObserver);
|
|
if (z) {
|
|
this.a.subscribe(replayObserver);
|
|
}
|
|
} catch (Throwable th) {
|
|
if (z) {
|
|
replayObserver.d.compareAndSet(true, false);
|
|
}
|
|
Exceptions.b(th);
|
|
throw ExceptionHelper.a(th);
|
|
}
|
|
}
|
|
}
|