diff --git a/src/main/java/io/reactivex/flowables/ConnectableFlowable.java b/src/main/java/io/reactivex/flowables/ConnectableFlowable.java index 21636e67da..ef5b667bac 100644 --- a/src/main/java/io/reactivex/flowables/ConnectableFlowable.java +++ b/src/main/java/io/reactivex/flowables/ConnectableFlowable.java @@ -68,6 +68,24 @@ public final Disposable connect() { return cc.disposable; } + /** + * Apply a workaround for a race condition with the regular publish().refCount() + * so that racing subscribers and refCount won't hang. + * + * @return the ConnectableFlowable to work with + * @since 2.2.10 + */ + private ConnectableFlowable onRefCount() { + if (this instanceof FlowablePublishClassic) { + @SuppressWarnings("unchecked") + FlowablePublishClassic fp = (FlowablePublishClassic) this; + return RxJavaPlugins.onAssembly( + new FlowablePublishAlt(fp.publishSource(), fp.publishBufferSize()) + ); + } + return this; + } + /** * Returns a {@code Flowable} that stays connected to this {@code ConnectableFlowable} as long as there * is at least one subscription to this {@code ConnectableFlowable}. @@ -89,7 +107,7 @@ public final Disposable connect() { @SchedulerSupport(SchedulerSupport.NONE) @BackpressureSupport(BackpressureKind.PASS_THROUGH) public Flowable refCount() { - return RxJavaPlugins.onAssembly(new FlowableRefCount(this)); + return RxJavaPlugins.onAssembly(new FlowableRefCount(onRefCount())); } /** @@ -216,7 +234,7 @@ public final Flowable refCount(int subscriberCount, long timeout, TimeUnit un ObjectHelper.verifyPositive(subscriberCount, "subscriberCount"); ObjectHelper.requireNonNull(unit, "unit is null"); ObjectHelper.requireNonNull(scheduler, "scheduler is null"); - return RxJavaPlugins.onAssembly(new FlowableRefCount(this, subscriberCount, timeout, unit, scheduler)); + return RxJavaPlugins.onAssembly(new FlowableRefCount(onRefCount(), subscriberCount, timeout, unit, scheduler)); } /** diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublish.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublish.java index 7ab84a568b..e573f3daf3 100644 --- a/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublish.java +++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublish.java @@ -33,7 +33,8 @@ * manner. * @param the value type */ -public final class FlowablePublish extends ConnectableFlowable implements HasUpstreamPublisher { +public final class FlowablePublish extends ConnectableFlowable +implements HasUpstreamPublisher, FlowablePublishClassic { /** * Indicates this child has been cancelled: the state is swapped in atomically and * will prevent the dispatch() to emit (too many) values to a terminated child subscriber. @@ -77,6 +78,19 @@ public Publisher source() { return source; } + /** + * @return The internal buffer size of this FloawblePublish operator. + */ + @Override + public int publishBufferSize() { + return bufferSize; + } + + @Override + public Publisher publishSource() { + return source; + } + @Override protected void subscribeActual(Subscriber s) { onSubscribe.subscribe(s); diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublishAlt.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublishAlt.java new file mode 100644 index 0000000000..d58ba84503 --- /dev/null +++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublishAlt.java @@ -0,0 +1,483 @@ +/** + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.internal.operators.flowable; + +import java.util.concurrent.atomic.*; + +import org.reactivestreams.*; + +import io.reactivex.FlowableSubscriber; +import io.reactivex.disposables.Disposable; +import io.reactivex.exceptions.*; +import io.reactivex.flowables.ConnectableFlowable; +import io.reactivex.functions.Consumer; +import io.reactivex.internal.disposables.ResettableConnectable; +import io.reactivex.internal.fuseable.*; +import io.reactivex.internal.queue.SpscArrayQueue; +import io.reactivex.internal.subscriptions.SubscriptionHelper; +import io.reactivex.internal.util.*; +import io.reactivex.plugins.RxJavaPlugins; + +/** + * Shares a single underlying connection to the upstream Publisher + * and multicasts events to all subscribed subscribers until the upstream + * completes or the connection is disposed. + *

+ * The difference to FlowablePublish is that when the upstream terminates, + * late subscriberss will receive that terminal event until the connection is + * disposed and the ConnectableFlowable is reset to its fresh state. + * + * @param the element type + * @since 2.2.10 + */ +public final class FlowablePublishAlt extends ConnectableFlowable +implements HasUpstreamPublisher, ResettableConnectable { + + final Publisher source; + + final int bufferSize; + + final AtomicReference> current; + + public FlowablePublishAlt(Publisher source, int bufferSize) { + this.source = source; + this.bufferSize = bufferSize; + this.current = new AtomicReference>(); + } + + @Override + public Publisher source() { + return source; + } + + /** + * @return The internal buffer size of this FloawblePublishAlt operator. + */ + public int publishBufferSize() { + return bufferSize; + } + + @Override + public void connect(Consumer connection) { + PublishConnection conn; + boolean doConnect = false; + + for (;;) { + conn = current.get(); + + if (conn == null || conn.isDisposed()) { + PublishConnection fresh = new PublishConnection(current, bufferSize); + if (!current.compareAndSet(conn, fresh)) { + continue; + } + conn = fresh; + } + + doConnect = !conn.connect.get() && conn.connect.compareAndSet(false, true); + break; + } + + try { + connection.accept(conn); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + throw ExceptionHelper.wrapOrThrow(ex); + } + + if (doConnect) { + source.subscribe(conn); + } + } + + @Override + protected void subscribeActual(Subscriber s) { + PublishConnection conn; + + for (;;) { + conn = current.get(); + + // don't create a fresh connection if the current is disposed + if (conn == null) { + PublishConnection fresh = new PublishConnection(current, bufferSize); + if (!current.compareAndSet(conn, fresh)) { + continue; + } + conn = fresh; + } + + break; + } + + InnerSubscription inner = new InnerSubscription(s, conn); + s.onSubscribe(inner); + + if (conn.add(inner)) { + if (inner.isCancelled()) { + conn.remove(inner); + } else { + conn.drain(); + } + return; + } + + Throwable ex = conn.error; + if (ex != null) { + s.onError(ex); + } else { + s.onComplete(); + } + } + + @SuppressWarnings("unchecked") + @Override + public void resetIf(Disposable connection) { + current.compareAndSet((PublishConnection)connection, null); + } + + static final class PublishConnection + extends AtomicInteger + implements FlowableSubscriber, Disposable { + + private static final long serialVersionUID = -1672047311619175801L; + + final AtomicReference> current; + + final AtomicReference upstream; + + final AtomicBoolean connect; + + final AtomicReference[]> subscribers; + + final int bufferSize; + + volatile SimpleQueue queue; + + int sourceMode; + + volatile boolean done; + Throwable error; + + int consumed; + + @SuppressWarnings("rawtypes") + static final InnerSubscription[] EMPTY = new InnerSubscription[0]; + @SuppressWarnings("rawtypes") + static final InnerSubscription[] TERMINATED = new InnerSubscription[0]; + + @SuppressWarnings("unchecked") + PublishConnection(AtomicReference> current, int bufferSize) { + this.current = current; + this.upstream = new AtomicReference(); + this.connect = new AtomicBoolean(); + this.bufferSize = bufferSize; + this.subscribers = new AtomicReference[]>(EMPTY); + } + + @SuppressWarnings("unchecked") + @Override + public void dispose() { + subscribers.getAndSet(TERMINATED); + current.compareAndSet(this, null); + SubscriptionHelper.cancel(upstream); + } + + @Override + public boolean isDisposed() { + return subscribers.get() == TERMINATED; + } + + @Override + public void onSubscribe(Subscription s) { + if (SubscriptionHelper.setOnce(this.upstream, s)) { + if (s instanceof QueueSubscription) { + @SuppressWarnings("unchecked") + QueueSubscription qs = (QueueSubscription) s; + + int m = qs.requestFusion(QueueSubscription.ANY | QueueSubscription.BOUNDARY); + if (m == QueueSubscription.SYNC) { + sourceMode = m; + queue = qs; + done = true; + drain(); + return; + } + if (m == QueueSubscription.ASYNC) { + sourceMode = m; + queue = qs; + s.request(bufferSize); + return; + } + } + + queue = new SpscArrayQueue(bufferSize); + + s.request(bufferSize); + } + } + + @Override + public void onNext(T t) { + // we expect upstream to honor backpressure requests + if (sourceMode == QueueSubscription.NONE && !queue.offer(t)) { + onError(new MissingBackpressureException("Prefetch queue is full?!")); + return; + } + // since many things can happen concurrently, we have a common dispatch + // loop to act on the current state serially + drain(); + } + + @Override + public void onError(Throwable t) { + if (done) { + RxJavaPlugins.onError(t); + } else { + error = t; + done = true; + drain(); + } + } + + @Override + public void onComplete() { + done = true; + drain(); + } + + void drain() { + if (getAndIncrement() != 0) { + return; + } + + int missed = 1; + SimpleQueue queue = this.queue; + int consumed = this.consumed; + int limit = this.bufferSize - (this.bufferSize >> 2); + boolean async = this.sourceMode != QueueSubscription.SYNC; + + outer: + for (;;) { + if (queue != null) { + long minDemand = Long.MAX_VALUE; + boolean hasDemand = false; + + InnerSubscription[] innerSubscriptions = subscribers.get(); + + for (InnerSubscription inner : innerSubscriptions) { + long request = inner.get(); + if (request != Long.MIN_VALUE) { + hasDemand = true; + minDemand = Math.min(request - inner.emitted, minDemand); + } + } + + if (!hasDemand) { + minDemand = 0L; + } + + while (minDemand != 0L) { + boolean d = done; + T v; + + try { + v = queue.poll(); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + upstream.get().cancel(); + queue.clear(); + done = true; + signalError(ex); + return; + } + + boolean empty = v == null; + + if (checkTerminated(d, empty)) { + return; + } + + if (empty) { + break; + } + + for (InnerSubscription inner : innerSubscriptions) { + if (!inner.isCancelled()) { + inner.downstream.onNext(v); + inner.emitted++; + } + } + + if (async && ++consumed == limit) { + consumed = 0; + upstream.get().request(limit); + } + minDemand--; + + if (innerSubscriptions != subscribers.get()) { + continue outer; + } + } + + if (checkTerminated(done, queue.isEmpty())) { + return; + } + } + + this.consumed = consumed; + missed = addAndGet(-missed); + if (missed == 0) { + break; + } + if (queue == null) { + queue = this.queue; + } + } + } + + @SuppressWarnings("unchecked") + boolean checkTerminated(boolean isDone, boolean isEmpty) { + if (isDone && isEmpty) { + Throwable ex = error; + + if (ex != null) { + signalError(ex); + } else { + for (InnerSubscription inner : subscribers.getAndSet(TERMINATED)) { + if (!inner.isCancelled()) { + inner.downstream.onComplete(); + } + } + } + return true; + } + return false; + } + + @SuppressWarnings("unchecked") + void signalError(Throwable ex) { + for (InnerSubscription inner : subscribers.getAndSet(TERMINATED)) { + if (!inner.isCancelled()) { + inner.downstream.onError(ex); + } + } + } + + boolean add(InnerSubscription inner) { + // the state can change so we do a CAS loop to achieve atomicity + for (;;) { + // get the current producer array + InnerSubscription[] c = subscribers.get(); + // if this subscriber-to-source reached a terminal state by receiving + // an onError or onComplete, just refuse to add the new producer + if (c == TERMINATED) { + return false; + } + // we perform a copy-on-write logic + int len = c.length; + @SuppressWarnings("unchecked") + InnerSubscription[] u = new InnerSubscription[len + 1]; + System.arraycopy(c, 0, u, 0, len); + u[len] = inner; + // try setting the subscribers array + if (subscribers.compareAndSet(c, u)) { + return true; + } + // if failed, some other operation succeeded (another add, remove or termination) + // so retry + } + } + + @SuppressWarnings("unchecked") + void remove(InnerSubscription inner) { + // the state can change so we do a CAS loop to achieve atomicity + for (;;) { + // let's read the current subscribers array + InnerSubscription[] c = subscribers.get(); + int len = c.length; + // if it is either empty or terminated, there is nothing to remove so we quit + if (len == 0) { + break; + } + // let's find the supplied producer in the array + // although this is O(n), we don't expect too many child subscribers in general + int j = -1; + for (int i = 0; i < len; i++) { + if (c[i] == inner) { + j = i; + break; + } + } + // we didn't find it so just quit + if (j < 0) { + return; + } + // we do copy-on-write logic here + InnerSubscription[] u; + // we don't create a new empty array if producer was the single inhabitant + // but rather reuse an empty array + if (len == 1) { + u = EMPTY; + } else { + // otherwise, create a new array one less in size + u = new InnerSubscription[len - 1]; + // copy elements being before the given producer + System.arraycopy(c, 0, u, 0, j); + // copy elements being after the given producer + System.arraycopy(c, j + 1, u, j, len - j - 1); + } + // try setting this new array as + if (subscribers.compareAndSet(c, u)) { + break; + } + // if we failed, it means something else happened + // (a concurrent add/remove or termination), we need to retry + } + } + } + + static final class InnerSubscription extends AtomicLong + implements Subscription { + + private static final long serialVersionUID = 2845000326761540265L; + + final Subscriber downstream; + + final PublishConnection parent; + + long emitted; + + InnerSubscription(Subscriber downstream, PublishConnection parent) { + this.downstream = downstream; + this.parent = parent; + } + + @Override + public void request(long n) { + BackpressureHelper.addCancel(this, n); + parent.drain(); + } + + @Override + public void cancel() { + if (getAndSet(Long.MIN_VALUE) != Long.MIN_VALUE) { + parent.remove(this); + parent.drain(); + } + } + + public boolean isCancelled() { + return get() == Long.MIN_VALUE; + } + } +} diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublishClassic.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublishClassic.java new file mode 100644 index 0000000000..0cbf70efad --- /dev/null +++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublishClassic.java @@ -0,0 +1,41 @@ +/** + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.internal.operators.flowable; + +import org.reactivestreams.Publisher; + +/** + * Interface to mark classic publish() operators to + * indicate refCount() should replace them with the Alt + * implementation. + *

+ * Without this, hooking the connectables with an intercept + * implementation would result in the unintended lack + * or presense of the replacement by refCount(). + * + * @param the element type of the sequence + * @since 2.2.10 + */ +public interface FlowablePublishClassic { + + /** + * @return the upstream source of this publish operator + */ + Publisher publishSource(); + + /** + * @return the internal buffer size of this publish operator + */ + int publishBufferSize(); +} diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservablePublish.java b/src/main/java/io/reactivex/internal/operators/observable/ObservablePublish.java index debc37875b..04b90506c3 100644 --- a/src/main/java/io/reactivex/internal/operators/observable/ObservablePublish.java +++ b/src/main/java/io/reactivex/internal/operators/observable/ObservablePublish.java @@ -30,7 +30,8 @@ * manner. * @param the value type */ -public final class ObservablePublish extends ConnectableObservable implements HasUpstreamObservableSource { +public final class ObservablePublish extends ConnectableObservable +implements HasUpstreamObservableSource, ObservablePublishClassic { /** The source observable. */ final ObservableSource source; /** Holds the current subscriber that is, will be or just was subscribed to the source observable. */ @@ -63,6 +64,11 @@ public ObservableSource source() { return source; } + @Override + public ObservableSource publishSource() { + return source; + } + @Override protected void subscribeActual(Observer observer) { onSubscribe.subscribe(observer); diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservablePublishAlt.java b/src/main/java/io/reactivex/internal/operators/observable/ObservablePublishAlt.java new file mode 100644 index 0000000000..771e58dda8 --- /dev/null +++ b/src/main/java/io/reactivex/internal/operators/observable/ObservablePublishAlt.java @@ -0,0 +1,282 @@ +/** + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.internal.operators.observable; + +import java.util.concurrent.atomic.*; + +import io.reactivex.*; +import io.reactivex.disposables.Disposable; +import io.reactivex.exceptions.Exceptions; +import io.reactivex.functions.Consumer; +import io.reactivex.internal.disposables.*; +import io.reactivex.internal.fuseable.HasUpstreamObservableSource; +import io.reactivex.internal.util.ExceptionHelper; +import io.reactivex.observables.ConnectableObservable; + +/** + * Shares a single underlying connection to the upstream ObservableSource + * and multicasts events to all subscribed observers until the upstream + * completes or the connection is disposed. + *

+ * The difference to ObservablePublish is that when the upstream terminates, + * late observers will receive that terminal event until the connection is + * disposed and the ConnectableObservable is reset to its fresh state. + * + * @param the element type + * @since 2.2.10 + */ +public final class ObservablePublishAlt extends ConnectableObservable +implements HasUpstreamObservableSource, ResettableConnectable { + + final ObservableSource source; + + final AtomicReference> current; + + public ObservablePublishAlt(ObservableSource source) { + this.source = source; + this.current = new AtomicReference>(); + } + + @Override + public void connect(Consumer connection) { + boolean doConnect = false; + PublishConnection conn; + + for (;;) { + conn = current.get(); + + if (conn == null || conn.isDisposed()) { + PublishConnection fresh = new PublishConnection(current); + if (!current.compareAndSet(conn, fresh)) { + continue; + } + conn = fresh; + } + + doConnect = !conn.connect.get() && conn.connect.compareAndSet(false, true); + break; + } + + try { + connection.accept(conn); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + throw ExceptionHelper.wrapOrThrow(ex); + } + + if (doConnect) { + source.subscribe(conn); + } + } + + @Override + protected void subscribeActual(Observer observer) { + PublishConnection conn; + + for (;;) { + conn = current.get(); + // we don't create a fresh connection if the current is terminated + if (conn == null) { + PublishConnection fresh = new PublishConnection(current); + if (!current.compareAndSet(conn, fresh)) { + continue; + } + conn = fresh; + } + break; + } + + InnerDisposable inner = new InnerDisposable(observer, conn); + observer.onSubscribe(inner); + if (conn.add(inner)) { + if (inner.isDisposed()) { + conn.remove(inner); + } + return; + } + // Late observers will be simply terminated + Throwable error = conn.error; + if (error != null) { + observer.onError(error); + } else { + observer.onComplete(); + } + } + + @Override + @SuppressWarnings("unchecked") + public void resetIf(Disposable connection) { + current.compareAndSet((PublishConnection)connection, null); + } + + @Override + public ObservableSource source() { + return source; + } + + static final class PublishConnection + extends AtomicReference[]> + implements Observer, Disposable { + + private static final long serialVersionUID = -3251430252873581268L; + + final AtomicBoolean connect; + + final AtomicReference> current; + + final AtomicReference upstream; + + @SuppressWarnings("rawtypes") + static final InnerDisposable[] EMPTY = new InnerDisposable[0]; + + @SuppressWarnings("rawtypes") + static final InnerDisposable[] TERMINATED = new InnerDisposable[0]; + + Throwable error; + + @SuppressWarnings("unchecked") + public PublishConnection(AtomicReference> current) { + this.connect = new AtomicBoolean(); + this.current = current; + this.upstream = new AtomicReference(); + lazySet(EMPTY); + } + + @SuppressWarnings("unchecked") + @Override + public void dispose() { + getAndSet(TERMINATED); + current.compareAndSet(this, null); + DisposableHelper.dispose(upstream); + } + + @Override + public boolean isDisposed() { + return get() == TERMINATED; + } + + @Override + public void onSubscribe(Disposable d) { + DisposableHelper.setOnce(upstream, d); + } + + @Override + public void onNext(T t) { + for (InnerDisposable inner : get()) { + inner.downstream.onNext(t); + } + } + + @Override + @SuppressWarnings("unchecked") + public void onError(Throwable e) { + error = e; + upstream.lazySet(DisposableHelper.DISPOSED); + for (InnerDisposable inner : getAndSet(TERMINATED)) { + inner.downstream.onError(e); + } + } + + @Override + @SuppressWarnings("unchecked") + public void onComplete() { + upstream.lazySet(DisposableHelper.DISPOSED); + for (InnerDisposable inner : getAndSet(TERMINATED)) { + inner.downstream.onComplete(); + } + } + + public boolean add(InnerDisposable inner) { + for (;;) { + InnerDisposable[] a = get(); + if (a == TERMINATED) { + return false; + } + int n = a.length; + @SuppressWarnings("unchecked") + InnerDisposable[] b = new InnerDisposable[n + 1]; + System.arraycopy(a, 0, b, 0, n); + b[n] = inner; + if (compareAndSet(a, b)) { + return true; + } + } + } + + @SuppressWarnings("unchecked") + public void remove(InnerDisposable inner) { + for (;;) { + InnerDisposable[] a = get(); + int n = a.length; + if (n == 0) { + return; + } + + int j = -1; + for (int i = 0; i < n; i++) { + if (a[i] == inner) { + j = i; + break; + } + } + + if (j < 0) { + return; + } + InnerDisposable[] b = EMPTY; + if (n != 1) { + b = new InnerDisposable[n - 1]; + System.arraycopy(a, 0, b, 0, j); + System.arraycopy(a, j + 1, b, j, n - j - 1); + } + if (compareAndSet(a, b)) { + return; + } + } + } + } + + /** + * Intercepts the dispose signal from the downstream and + * removes itself from the connection's observers array + * at most once. + * @param the element type + */ + static final class InnerDisposable + extends AtomicReference> + implements Disposable { + + private static final long serialVersionUID = 7463222674719692880L; + + final Observer downstream; + + public InnerDisposable(Observer downstream, PublishConnection parent) { + this.downstream = downstream; + lazySet(parent); + } + + @Override + public void dispose() { + PublishConnection p = getAndSet(null); + if (p != null) { + p.remove(this); + } + } + + @Override + public boolean isDisposed() { + return get() == null; + } + } +} diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservablePublishClassic.java b/src/main/java/io/reactivex/internal/operators/observable/ObservablePublishClassic.java new file mode 100644 index 0000000000..f072779930 --- /dev/null +++ b/src/main/java/io/reactivex/internal/operators/observable/ObservablePublishClassic.java @@ -0,0 +1,36 @@ +/** + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.internal.operators.observable; + +import io.reactivex.ObservableSource; + +/** + * Interface to mark classic publish() operators to + * indicate refCount() should replace them with the Alt + * implementation. + *

+ * Without this, hooking the connectables with an intercept + * implementation would result in the unintended lack + * or presense of the replacement by refCount(). + * + * @param the element type of the sequence + * @since 2.2.10 + */ +public interface ObservablePublishClassic { + + /** + * @return the upstream source of this publish operator + */ + ObservableSource publishSource(); +} diff --git a/src/main/java/io/reactivex/observables/ConnectableObservable.java b/src/main/java/io/reactivex/observables/ConnectableObservable.java index b5e54054b1..09fa70899e 100644 --- a/src/main/java/io/reactivex/observables/ConnectableObservable.java +++ b/src/main/java/io/reactivex/observables/ConnectableObservable.java @@ -66,6 +66,23 @@ public final Disposable connect() { return cc.disposable; } + /** + * Apply a workaround for a race condition with the regular publish().refCount() + * so that racing observers and refCount won't hang. + * + * @return the ConnectableObservable to work with + * @since 2.2.10 + */ + @SuppressWarnings("unchecked") + private ConnectableObservable onRefCount() { + if (this instanceof ObservablePublishClassic) { + return RxJavaPlugins.onAssembly( + new ObservablePublishAlt(((ObservablePublishClassic)this).publishSource()) + ); + } + return this; + } + /** * Returns an {@code Observable} that stays connected to this {@code ConnectableObservable} as long as there * is at least one subscription to this {@code ConnectableObservable}. @@ -83,7 +100,7 @@ public final Disposable connect() { @CheckReturnValue @SchedulerSupport(SchedulerSupport.NONE) public Observable refCount() { - return RxJavaPlugins.onAssembly(new ObservableRefCount(this)); + return RxJavaPlugins.onAssembly(new ObservableRefCount(onRefCount())); } /** @@ -190,7 +207,7 @@ public final Observable refCount(int subscriberCount, long timeout, TimeUnit ObjectHelper.verifyPositive(subscriberCount, "subscriberCount"); ObjectHelper.requireNonNull(unit, "unit is null"); ObjectHelper.requireNonNull(scheduler, "scheduler is null"); - return RxJavaPlugins.onAssembly(new ObservableRefCount(this, subscriberCount, timeout, unit, scheduler)); + return RxJavaPlugins.onAssembly(new ObservableRefCount(onRefCount(), subscriberCount, timeout, unit, scheduler)); } /** diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowablePublishAltTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowablePublishAltTest.java new file mode 100644 index 0000000000..414aa79c07 --- /dev/null +++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowablePublishAltTest.java @@ -0,0 +1,1629 @@ +/** + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.internal.operators.flowable; + +import static org.junit.Assert.*; + +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; + +import org.junit.*; +import org.reactivestreams.*; + +import io.reactivex.*; +import io.reactivex.disposables.Disposable; +import io.reactivex.exceptions.*; +import io.reactivex.flowables.ConnectableFlowable; +import io.reactivex.functions.*; +import io.reactivex.internal.functions.Functions; +import io.reactivex.internal.fuseable.HasUpstreamPublisher; +import io.reactivex.internal.operators.flowable.FlowablePublish.*; +import io.reactivex.internal.schedulers.ImmediateThinScheduler; +import io.reactivex.internal.subscriptions.BooleanSubscription; +import io.reactivex.plugins.RxJavaPlugins; +import io.reactivex.processors.PublishProcessor; +import io.reactivex.schedulers.*; +import io.reactivex.subscribers.TestSubscriber; + +public class FlowablePublishAltTest { + + @Test + public void testPublish() throws InterruptedException { + final AtomicInteger counter = new AtomicInteger(); + ConnectableFlowable f = Flowable.unsafeCreate(new Publisher() { + + @Override + public void subscribe(final Subscriber subscriber) { + subscriber.onSubscribe(new BooleanSubscription()); + new Thread(new Runnable() { + + @Override + public void run() { + counter.incrementAndGet(); + subscriber.onNext("one"); + subscriber.onComplete(); + } + }).start(); + } + }).publish(); + + final CountDownLatch latch = new CountDownLatch(2); + + // subscribe once + f.subscribe(new Consumer() { + + @Override + public void accept(String v) { + assertEquals("one", v); + latch.countDown(); + } + }); + + // subscribe again + f.subscribe(new Consumer() { + + @Override + public void accept(String v) { + assertEquals("one", v); + latch.countDown(); + } + }); + + Disposable connection = f.connect(); + try { + if (!latch.await(1000, TimeUnit.MILLISECONDS)) { + fail("subscriptions did not receive values"); + } + assertEquals(1, counter.get()); + } finally { + connection.dispose(); + } + } + + @Test + public void testBackpressureFastSlow() { + ConnectableFlowable is = Flowable.range(1, Flowable.bufferSize() * 2).publish(); + Flowable fast = is.observeOn(Schedulers.computation()) + .doOnComplete(new Action() { + @Override + public void run() { + System.out.println("^^^^^^^^^^^^^ completed FAST"); + } + }); + + Flowable slow = is.observeOn(Schedulers.computation()).map(new Function() { + int c; + + @Override + public Integer apply(Integer i) { + if (c == 0) { + try { + Thread.sleep(500); + } catch (InterruptedException e) { + } + } + c++; + return i; + } + + }).doOnComplete(new Action() { + + @Override + public void run() { + System.out.println("^^^^^^^^^^^^^ completed SLOW"); + } + + }); + + TestSubscriber ts = new TestSubscriber(); + Flowable.merge(fast, slow).subscribe(ts); + is.connect(); + ts.awaitTerminalEvent(); + ts.assertNoErrors(); + assertEquals(Flowable.bufferSize() * 4, ts.valueCount()); + } + + // use case from https://github.com/ReactiveX/RxJava/issues/1732 + @Test + public void testTakeUntilWithPublishedStreamUsingSelector() { + final AtomicInteger emitted = new AtomicInteger(); + Flowable xs = Flowable.range(0, Flowable.bufferSize() * 2).doOnNext(new Consumer() { + + @Override + public void accept(Integer t1) { + emitted.incrementAndGet(); + } + + }); + TestSubscriber ts = new TestSubscriber(); + xs.publish(new Function, Flowable>() { + + @Override + public Flowable apply(Flowable xs) { + return xs.takeUntil(xs.skipWhile(new Predicate() { + + @Override + public boolean test(Integer i) { + return i <= 3; + } + + })); + } + + }).subscribe(ts); + ts.awaitTerminalEvent(); + ts.assertNoErrors(); + ts.assertValues(0, 1, 2, 3); + assertEquals(5, emitted.get()); + System.out.println(ts.values()); + } + + // use case from https://github.com/ReactiveX/RxJava/issues/1732 + @Test + public void testTakeUntilWithPublishedStream() { + Flowable xs = Flowable.range(0, Flowable.bufferSize() * 2); + TestSubscriber ts = new TestSubscriber(); + ConnectableFlowable xsp = xs.publish(); + xsp.takeUntil(xsp.skipWhile(new Predicate() { + + @Override + public boolean test(Integer i) { + return i <= 3; + } + + })).subscribe(ts); + xsp.connect(); + System.out.println(ts.values()); + } + + @Test(timeout = 10000) + public void testBackpressureTwoConsumers() { + final AtomicInteger sourceEmission = new AtomicInteger(); + final AtomicBoolean sourceUnsubscribed = new AtomicBoolean(); + final Flowable source = Flowable.range(1, 100) + .doOnNext(new Consumer() { + @Override + public void accept(Integer t1) { + sourceEmission.incrementAndGet(); + } + }) + .doOnCancel(new Action() { + @Override + public void run() { + sourceUnsubscribed.set(true); + } + }).share(); + ; + + final AtomicBoolean child1Unsubscribed = new AtomicBoolean(); + final AtomicBoolean child2Unsubscribed = new AtomicBoolean(); + + final TestSubscriber ts2 = new TestSubscriber(); + + final TestSubscriber ts1 = new TestSubscriber() { + @Override + public void onNext(Integer t) { + if (valueCount() == 2) { + source.doOnCancel(new Action() { + @Override + public void run() { + child2Unsubscribed.set(true); + } + }).take(5).subscribe(ts2); + } + super.onNext(t); + } + }; + + source.doOnCancel(new Action() { + @Override + public void run() { + child1Unsubscribed.set(true); + } + }).take(5) + .subscribe(ts1); + + ts1.awaitTerminalEvent(); + ts2.awaitTerminalEvent(); + + ts1.assertNoErrors(); + ts2.assertNoErrors(); + + assertTrue(sourceUnsubscribed.get()); + assertTrue(child1Unsubscribed.get()); + assertTrue(child2Unsubscribed.get()); + + ts1.assertValues(1, 2, 3, 4, 5); + ts2.assertValues(4, 5, 6, 7, 8); + + assertEquals(8, sourceEmission.get()); + } + + @Test + public void testConnectWithNoSubscriber() { + TestScheduler scheduler = new TestScheduler(); + ConnectableFlowable cf = Flowable.interval(10, 10, TimeUnit.MILLISECONDS, scheduler).take(3).publish(); + cf.connect(); + // Emit 0 + scheduler.advanceTimeBy(15, TimeUnit.MILLISECONDS); + TestSubscriber subscriber = new TestSubscriber(); + cf.subscribe(subscriber); + // Emit 1 and 2 + scheduler.advanceTimeBy(50, TimeUnit.MILLISECONDS); + subscriber.assertValues(1L, 2L); + subscriber.assertNoErrors(); + subscriber.assertTerminated(); + } + + @Test + public void testSubscribeAfterDisconnectThenConnect() { + ConnectableFlowable source = Flowable.just(1).publish(); + + TestSubscriber ts1 = new TestSubscriber(); + + source.subscribe(ts1); + + Disposable connection = source.connect(); + + ts1.assertValue(1); + ts1.assertNoErrors(); + ts1.assertTerminated(); + + TestSubscriber ts2 = new TestSubscriber(); + + source.subscribe(ts2); + + Disposable connection2 = source.connect(); + + ts2.assertValue(1); + ts2.assertNoErrors(); + ts2.assertTerminated(); + + System.out.println(connection); + System.out.println(connection2); + } + + @Test + public void testNoSubscriberRetentionOnCompleted() { + FlowablePublish source = (FlowablePublish)Flowable.just(1).publish(); + + TestSubscriber ts1 = new TestSubscriber(); + + source.subscribe(ts1); + + ts1.assertNoValues(); + ts1.assertNoErrors(); + ts1.assertNotComplete(); + + source.connect(); + + ts1.assertValue(1); + ts1.assertNoErrors(); + ts1.assertTerminated(); + + assertNull(source.current.get()); + } + + @Test + public void testNonNullConnection() { + ConnectableFlowable source = Flowable.never().publish(); + + assertNotNull(source.connect()); + assertNotNull(source.connect()); + } + + @Test + public void testNoDisconnectSomeoneElse() { + ConnectableFlowable source = Flowable.never().publish(); + + Disposable connection1 = source.connect(); + Disposable connection2 = source.connect(); + + connection1.dispose(); + + Disposable connection3 = source.connect(); + + connection2.dispose(); + + assertTrue(checkPublishDisposed(connection1)); + assertTrue(checkPublishDisposed(connection2)); + assertFalse(checkPublishDisposed(connection3)); + } + + @SuppressWarnings("unchecked") + static boolean checkPublishDisposed(Disposable d) { + return ((FlowablePublish.PublishSubscriber)d).isDisposed(); + } + + @Test + public void testZeroRequested() { + ConnectableFlowable source = Flowable.just(1).publish(); + + TestSubscriber ts = new TestSubscriber(0L); + + source.subscribe(ts); + + ts.assertNoValues(); + ts.assertNoErrors(); + ts.assertNotComplete(); + + source.connect(); + + ts.assertNoValues(); + ts.assertNoErrors(); + ts.assertNotComplete(); + + ts.request(5); + + ts.assertValue(1); + ts.assertNoErrors(); + ts.assertTerminated(); + } + + @Test + public void testConnectIsIdempotent() { + final AtomicInteger calls = new AtomicInteger(); + Flowable source = Flowable.unsafeCreate(new Publisher() { + @Override + public void subscribe(Subscriber t) { + t.onSubscribe(new BooleanSubscription()); + calls.getAndIncrement(); + } + }); + + ConnectableFlowable conn = source.publish(); + + assertEquals(0, calls.get()); + + conn.connect(); + conn.connect(); + + assertEquals(1, calls.get()); + + conn.connect().dispose(); + + conn.connect(); + conn.connect(); + + assertEquals(2, calls.get()); + } + + @Test + public void syncFusedObserveOn() { + ConnectableFlowable cf = Flowable.range(0, 1000).publish(); + Flowable obs = cf.observeOn(Schedulers.computation()); + for (int i = 0; i < 1000; i++) { + for (int j = 1; j < 6; j++) { + List> tss = new ArrayList>(); + for (int k = 1; k < j; k++) { + TestSubscriber ts = new TestSubscriber(); + tss.add(ts); + obs.subscribe(ts); + } + + Disposable connection = cf.connect(); + + for (TestSubscriber ts : tss) { + ts.awaitDone(5, TimeUnit.SECONDS) + .assertSubscribed() + .assertValueCount(1000) + .assertNoErrors() + .assertComplete(); + } + connection.dispose(); + } + } + } + + @Test + public void syncFusedObserveOn2() { + ConnectableFlowable cf = Flowable.range(0, 1000).publish(); + Flowable obs = cf.observeOn(ImmediateThinScheduler.INSTANCE); + for (int i = 0; i < 1000; i++) { + for (int j = 1; j < 6; j++) { + List> tss = new ArrayList>(); + for (int k = 1; k < j; k++) { + TestSubscriber ts = new TestSubscriber(); + tss.add(ts); + obs.subscribe(ts); + } + + Disposable connection = cf.connect(); + + for (TestSubscriber ts : tss) { + ts.awaitDone(5, TimeUnit.SECONDS) + .assertSubscribed() + .assertValueCount(1000) + .assertNoErrors() + .assertComplete(); + } + connection.dispose(); + } + } + } + + @Test + public void asyncFusedObserveOn() { + ConnectableFlowable cf = Flowable.range(0, 1000).observeOn(ImmediateThinScheduler.INSTANCE).publish(); + for (int i = 0; i < 1000; i++) { + for (int j = 1; j < 6; j++) { + List> tss = new ArrayList>(); + for (int k = 1; k < j; k++) { + TestSubscriber ts = new TestSubscriber(); + tss.add(ts); + cf.subscribe(ts); + } + + Disposable connection = cf.connect(); + + for (TestSubscriber ts : tss) { + ts.awaitDone(5, TimeUnit.SECONDS) + .assertSubscribed() + .assertValueCount(1000) + .assertNoErrors() + .assertComplete(); + } + connection.dispose(); + } + } + } + + @Test + public void testObserveOn() { + ConnectableFlowable cf = Flowable.range(0, 1000).hide().publish(); + Flowable obs = cf.observeOn(Schedulers.computation()); + for (int i = 0; i < 1000; i++) { + for (int j = 1; j < 6; j++) { + List> tss = new ArrayList>(); + for (int k = 1; k < j; k++) { + TestSubscriber ts = new TestSubscriber(); + tss.add(ts); + obs.subscribe(ts); + } + + Disposable connection = cf.connect(); + + for (TestSubscriber ts : tss) { + ts.awaitDone(5, TimeUnit.SECONDS) + .assertSubscribed() + .assertValueCount(1000) + .assertNoErrors() + .assertComplete(); + } + connection.dispose(); + } + } + } + + @Test + public void source() { + Flowable f = Flowable.never(); + + assertSame(f, (((HasUpstreamPublisher)f.publish()).source())); + } + + @Test + public void connectThrows() { + ConnectableFlowable cf = Flowable.empty().publish(); + try { + cf.connect(new Consumer() { + @Override + public void accept(Disposable d) throws Exception { + throw new TestException(); + } + }); + } catch (TestException ex) { + // expected + } + } + + @Test + public void addRemoveRace() { + for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { + + final ConnectableFlowable cf = Flowable.empty().publish(); + + final TestSubscriber ts = cf.test(); + + final TestSubscriber ts2 = new TestSubscriber(); + + Runnable r1 = new Runnable() { + @Override + public void run() { + cf.subscribe(ts2); + } + }; + + Runnable r2 = new Runnable() { + @Override + public void run() { + ts.cancel(); + } + }; + + TestHelper.race(r1, r2); + } + } + + @Test + public void disposeOnArrival() { + ConnectableFlowable cf = Flowable.empty().publish(); + + cf.test(Long.MAX_VALUE, true).assertEmpty(); + } + + @Test + public void disposeOnArrival2() { + Flowable co = Flowable.never().publish().autoConnect(); + + co.test(Long.MAX_VALUE, true).assertEmpty(); + } + + @Test + public void dispose() { + TestHelper.checkDisposed(Flowable.never().publish()); + + TestHelper.checkDisposed(Flowable.never().publish(Functions.>identity())); + } + + @Test + public void empty() { + ConnectableFlowable cf = Flowable.empty().publish(); + + cf.connect(); + } + + @Test + public void take() { + ConnectableFlowable cf = Flowable.range(1, 2).publish(); + + TestSubscriber ts = cf.take(1).test(); + + cf.connect(); + + ts.assertResult(1); + } + + @Test + public void just() { + final PublishProcessor pp = PublishProcessor.create(); + + ConnectableFlowable cf = pp.publish(); + + TestSubscriber ts = new TestSubscriber() { + @Override + public void onNext(Integer t) { + super.onNext(t); + pp.onComplete(); + } + }; + + cf.subscribe(ts); + cf.connect(); + + pp.onNext(1); + + ts.assertResult(1); + } + + @Test + public void nextCancelRace() { + for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { + + final PublishProcessor pp = PublishProcessor.create(); + + final ConnectableFlowable cf = pp.publish(); + + final TestSubscriber ts = cf.test(); + + Runnable r1 = new Runnable() { + @Override + public void run() { + pp.onNext(1); + } + }; + + Runnable r2 = new Runnable() { + @Override + public void run() { + ts.cancel(); + } + }; + + TestHelper.race(r1, r2); + } + } + + @Test + public void badSource() { + List errors = TestHelper.trackPluginErrors(); + try { + new Flowable() { + @Override + protected void subscribeActual(Subscriber subscriber) { + subscriber.onSubscribe(new BooleanSubscription()); + subscriber.onNext(1); + subscriber.onComplete(); + subscriber.onNext(2); + subscriber.onError(new TestException()); + subscriber.onComplete(); + } + } + .publish() + .autoConnect() + .test() + .assertResult(1); + + TestHelper.assertUndeliverable(errors, 0, TestException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void noErrorLoss() { + List errors = TestHelper.trackPluginErrors(); + try { + ConnectableFlowable cf = Flowable.error(new TestException()).publish(); + + cf.connect(); + + TestHelper.assertUndeliverable(errors, 0, TestException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void subscribeDisconnectRace() { + for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { + + final PublishProcessor pp = PublishProcessor.create(); + + final ConnectableFlowable cf = pp.publish(); + + final Disposable d = cf.connect(); + final TestSubscriber ts = new TestSubscriber(); + + Runnable r1 = new Runnable() { + @Override + public void run() { + d.dispose(); + } + }; + + Runnable r2 = new Runnable() { + @Override + public void run() { + cf.subscribe(ts); + } + }; + + TestHelper.race(r1, r2); + } + } + + @Test + public void selectorDisconnectsIndependentSource() { + PublishProcessor pp = PublishProcessor.create(); + + pp.publish(new Function, Flowable>() { + @Override + public Flowable apply(Flowable v) throws Exception { + return Flowable.range(1, 2); + } + }) + .test() + .assertResult(1, 2); + + assertFalse(pp.hasSubscribers()); + } + + @Test(timeout = 5000) + public void selectorLatecommer() { + Flowable.range(1, 5) + .publish(new Function, Flowable>() { + @Override + public Flowable apply(Flowable v) throws Exception { + return v.concatWith(v); + } + }) + .test() + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void mainError() { + Flowable.error(new TestException()) + .publish(Functions.>identity()) + .test() + .assertFailure(TestException.class); + } + + @Test + public void selectorInnerError() { + PublishProcessor pp = PublishProcessor.create(); + + pp.publish(new Function, Flowable>() { + @Override + public Flowable apply(Flowable v) throws Exception { + return Flowable.error(new TestException()); + } + }) + .test() + .assertFailure(TestException.class); + + assertFalse(pp.hasSubscribers()); + } + + @Test + public void preNextConnect() { + for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { + + final ConnectableFlowable cf = Flowable.empty().publish(); + + cf.connect(); + + Runnable r1 = new Runnable() { + @Override + public void run() { + cf.test(); + } + }; + + TestHelper.race(r1, r1); + } + } + + @Test + public void connectRace() { + for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { + + final ConnectableFlowable cf = Flowable.empty().publish(); + + Runnable r1 = new Runnable() { + @Override + public void run() { + cf.connect(); + } + }; + + TestHelper.race(r1, r1); + } + } + + @Test + public void selectorCrash() { + Flowable.just(1).publish(new Function, Flowable>() { + @Override + public Flowable apply(Flowable v) throws Exception { + throw new TestException(); + } + }) + .test() + .assertFailure(TestException.class); + } + + @Test + public void pollThrows() { + Flowable.just(1) + .map(new Function() { + @Override + public Object apply(Integer v) throws Exception { + throw new TestException(); + } + }) + .compose(TestHelper.flowableStripBoundary()) + .publish() + .autoConnect() + .test() + .assertFailure(TestException.class); + } + + @Test + public void pollThrowsNoSubscribers() { + ConnectableFlowable cf = Flowable.just(1, 2) + .map(new Function() { + @Override + public Integer apply(Integer v) throws Exception { + if (v == 2) { + throw new TestException(); + } + return v; + } + }) + .compose(TestHelper.flowableStripBoundary()) + .publish(); + + TestSubscriber ts = cf.take(1) + .test(); + + cf.connect(); + + ts.assertResult(1); + } + + @Test + public void dryRunCrash() { + List errors = TestHelper.trackPluginErrors(); + try { + final TestSubscriber ts = new TestSubscriber(1L) { + @Override + public void onNext(Object t) { + super.onNext(t); + onComplete(); + cancel(); + } + }; + + Flowable.range(1, 10) + .map(new Function() { + @Override + public Object apply(Integer v) throws Exception { + if (v == 2) { + throw new TestException(); + } + return v; + } + }) + .publish() + .autoConnect() + .subscribe(ts); + + ts + .assertResult(1); + + TestHelper.assertUndeliverable(errors, 0, TestException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void overflowQueue() { + List errors = TestHelper.trackPluginErrors(); + try { + Flowable.create(new FlowableOnSubscribe() { + @Override + public void subscribe(FlowableEmitter s) throws Exception { + for (int i = 0; i < 10; i++) { + s.onNext(i); + } + } + }, BackpressureStrategy.MISSING) + .publish(8) + .autoConnect() + .test(0L) + .assertFailure(MissingBackpressureException.class); + + TestHelper.assertError(errors, 0, MissingBackpressureException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void delayedUpstreamOnSubscribe() { + final Subscriber[] sub = { null }; + + new Flowable() { + @Override + protected void subscribeActual(Subscriber s) { + sub[0] = s; + } + } + .publish() + .connect() + .dispose(); + + BooleanSubscription bs = new BooleanSubscription(); + + sub[0].onSubscribe(bs); + + assertTrue(bs.isCancelled()); + } + + @Test + public void disposeRace() { + for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { + + final AtomicReference ref = new AtomicReference(); + + final ConnectableFlowable cf = new Flowable() { + @Override + protected void subscribeActual(Subscriber s) { + s.onSubscribe(new BooleanSubscription()); + ref.set((Disposable)s); + } + }.publish(); + + cf.connect(); + + Runnable r1 = new Runnable() { + @Override + public void run() { + ref.get().dispose(); + } + }; + + TestHelper.race(r1, r1); + } + } + + @Test + public void removeNotPresent() { + final AtomicReference> ref = new AtomicReference>(); + + final ConnectableFlowable cf = new Flowable() { + @Override + @SuppressWarnings("unchecked") + protected void subscribeActual(Subscriber s) { + s.onSubscribe(new BooleanSubscription()); + ref.set((PublishSubscriber)s); + } + }.publish(); + + cf.connect(); + + ref.get().add(new InnerSubscriber(new TestSubscriber())); + ref.get().remove(null); + } + + @Test + @Ignore("publish() keeps consuming the upstream if there are no subscribers, 3.x should change this") + public void subscriberSwap() { + final ConnectableFlowable cf = Flowable.range(1, 5).publish(); + + cf.connect(); + + TestSubscriber ts1 = new TestSubscriber() { + @Override + public void onNext(Integer t) { + super.onNext(t); + cancel(); + onComplete(); + } + }; + + cf.subscribe(ts1); + + ts1.assertResult(1); + + TestSubscriber ts2 = new TestSubscriber(0); + cf.subscribe(ts2); + + ts2 + .assertEmpty() + .requestMore(4) + .assertResult(2, 3, 4, 5); + } + + @Test + public void subscriberLiveSwap() { + final ConnectableFlowable cf = new FlowablePublishAlt(Flowable.range(1, 5), 128); + + final TestSubscriber ts2 = new TestSubscriber(0); + + TestSubscriber ts1 = new TestSubscriber() { + @Override + public void onNext(Integer t) { + super.onNext(t); + cancel(); + onComplete(); + cf.subscribe(ts2); + } + }; + + cf.subscribe(ts1); + + cf.connect(); + + ts1.assertResult(1); + + ts2 + .assertEmpty() + .requestMore(4) + .assertResult(2, 3, 4, 5); + } + + @Test + public void selectorSubscriberSwap() { + final AtomicReference> ref = new AtomicReference>(); + + Flowable.range(1, 5).publish(new Function, Publisher>() { + @Override + public Publisher apply(Flowable f) throws Exception { + ref.set(f); + return Flowable.never(); + } + }).test(); + + ref.get().take(2).test().assertResult(1, 2); + + ref.get() + .test(0) + .assertEmpty() + .requestMore(2) + .assertValuesOnly(3, 4) + .requestMore(1) + .assertResult(3, 4, 5); + } + + @Test + public void leavingSubscriberOverrequests() { + final AtomicReference> ref = new AtomicReference>(); + + PublishProcessor pp = PublishProcessor.create(); + + pp.publish(new Function, Publisher>() { + @Override + public Publisher apply(Flowable f) throws Exception { + ref.set(f); + return Flowable.never(); + } + }).test(); + + TestSubscriber ts1 = ref.get().take(2).test(); + + pp.onNext(1); + pp.onNext(2); + + ts1.assertResult(1, 2); + + pp.onNext(3); + pp.onNext(4); + + TestSubscriber ts2 = ref.get().test(0L); + + ts2.assertEmpty(); + + ts2.requestMore(2); + + ts2.assertValuesOnly(3, 4); + } + + // call a transformer only if the input is non-empty + @Test + public void composeIfNotEmpty() { + final FlowableTransformer transformer = new FlowableTransformer() { + @Override + public Publisher apply(Flowable g) { + return g.map(new Function() { + @Override + public Integer apply(Integer v) throws Exception { + return v + 1; + } + }); + } + }; + + final AtomicInteger calls = new AtomicInteger(); + Flowable.range(1, 5) + .publish(new Function, Publisher>() { + @Override + public Publisher apply(final Flowable shared) + throws Exception { + return shared.take(1).concatMap(new Function>() { + @Override + public Publisher apply(Integer first) + throws Exception { + calls.incrementAndGet(); + return transformer.apply(Flowable.just(first).concatWith(shared)); + } + }); + } + }) + .test() + .assertResult(2, 3, 4, 5, 6); + + assertEquals(1, calls.get()); + } + + // call a transformer only if the input is non-empty + @Test + public void composeIfNotEmptyNotFused() { + final FlowableTransformer transformer = new FlowableTransformer() { + @Override + public Publisher apply(Flowable g) { + return g.map(new Function() { + @Override + public Integer apply(Integer v) throws Exception { + return v + 1; + } + }); + } + }; + + final AtomicInteger calls = new AtomicInteger(); + Flowable.range(1, 5).hide() + .publish(new Function, Publisher>() { + @Override + public Publisher apply(final Flowable shared) + throws Exception { + return shared.take(1).concatMap(new Function>() { + @Override + public Publisher apply(Integer first) + throws Exception { + calls.incrementAndGet(); + return transformer.apply(Flowable.just(first).concatWith(shared)); + } + }); + } + }) + .test() + .assertResult(2, 3, 4, 5, 6); + + assertEquals(1, calls.get()); + } + + // call a transformer only if the input is non-empty + @Test + public void composeIfNotEmptyIsEmpty() { + final FlowableTransformer transformer = new FlowableTransformer() { + @Override + public Publisher apply(Flowable g) { + return g.map(new Function() { + @Override + public Integer apply(Integer v) throws Exception { + return v + 1; + } + }); + } + }; + + final AtomicInteger calls = new AtomicInteger(); + Flowable.empty().hide() + .publish(new Function, Publisher>() { + @Override + public Publisher apply(final Flowable shared) + throws Exception { + return shared.take(1).concatMap(new Function>() { + @Override + public Publisher apply(Integer first) + throws Exception { + calls.incrementAndGet(); + return transformer.apply(Flowable.just(first).concatWith(shared)); + } + }); + } + }) + .test() + .assertResult(); + + assertEquals(0, calls.get()); + } + + @Test + public void publishFunctionCancelOuterAfterOneInner() { + final AtomicReference> ref = new AtomicReference>(); + + PublishProcessor pp = PublishProcessor.create(); + + final TestSubscriber ts = pp.publish(new Function, Publisher>() { + @Override + public Publisher apply(Flowable f) throws Exception { + ref.set(f); + return Flowable.never(); + } + }).test(); + + ref.get().subscribe(new TestSubscriber() { + @Override + public void onNext(Integer t) { + super.onNext(t); + onComplete(); + ts.cancel(); + } + }); + + pp.onNext(1); + } + + @Test + public void publishFunctionCancelOuterAfterOneInnerBackpressured() { + final AtomicReference> ref = new AtomicReference>(); + + PublishProcessor pp = PublishProcessor.create(); + + final TestSubscriber ts = pp.publish(new Function, Publisher>() { + @Override + public Publisher apply(Flowable f) throws Exception { + ref.set(f); + return Flowable.never(); + } + }).test(); + + ref.get().subscribe(new TestSubscriber(1L) { + @Override + public void onNext(Integer t) { + super.onNext(t); + onComplete(); + ts.cancel(); + } + }); + + pp.onNext(1); + } + + @Test + public void publishCancelOneAsync() { + for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) { + + final PublishProcessor pp = PublishProcessor.create(); + + final AtomicReference> ref = new AtomicReference>(); + + pp.publish(new Function, Publisher>() { + @Override + public Publisher apply(Flowable f) throws Exception { + ref.set(f); + return Flowable.never(); + } + }).test(); + + final TestSubscriber ts1 = ref.get().test(); + TestSubscriber ts2 = ref.get().test(); + + Runnable r1 = new Runnable() { + @Override + public void run() { + pp.onNext(1); + } + }; + + Runnable r2 = new Runnable() { + @Override + public void run() { + ts1.cancel(); + } + }; + + TestHelper.race(r1, r2); + + ts2.assertValuesOnly(1); + } + } + + @Test + public void publishCancelOneAsync2() { + final PublishProcessor pp = PublishProcessor.create(); + + ConnectableFlowable cf = pp.publish(); + + final TestSubscriber ts1 = new TestSubscriber(); + + final AtomicReference> ref = new AtomicReference>(); + + cf.subscribe(new FlowableSubscriber() { + @SuppressWarnings("unchecked") + @Override + public void onSubscribe(Subscription s) { + ts1.onSubscribe(new BooleanSubscription()); + // pretend to be cancelled without removing it from the subscriber list + ref.set((InnerSubscriber)s); + } + + @Override + public void onNext(Integer t) { + ts1.onNext(t); + } + + @Override + public void onError(Throwable t) { + ts1.onError(t); + } + + @Override + public void onComplete() { + ts1.onComplete(); + } + }); + TestSubscriber ts2 = cf.test(); + + cf.connect(); + + ref.get().set(Long.MIN_VALUE); + + pp.onNext(1); + + ts1.assertEmpty(); + ts2.assertValuesOnly(1); + } + + @Test + public void boundaryFusion() { + Flowable.range(1, 10000) + .observeOn(Schedulers.single()) + .map(new Function() { + @Override + public String apply(Integer t) throws Exception { + String name = Thread.currentThread().getName(); + if (name.contains("RxSingleScheduler")) { + return "RxSingleScheduler"; + } + return name; + } + }) + .share() + .observeOn(Schedulers.computation()) + .distinct() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult("RxSingleScheduler"); + } + + @Test + public void badRequest() { + TestHelper.assertBadRequestReported(Flowable.range(1, 5).publish()); + } + + @Test + @SuppressWarnings("unchecked") + public void splitCombineSubscriberChangeAfterOnNext() { + Flowable source = Flowable.range(0, 20) + .doOnSubscribe(new Consumer() { + @Override + public void accept(Subscription v) throws Exception { + System.out.println("Subscribed"); + } + }) + .publish(10) + .refCount() + ; + + Flowable evenNumbers = source.filter(new Predicate() { + @Override + public boolean test(Integer v) throws Exception { + return v % 2 == 0; + } + }); + + Flowable oddNumbers = source.filter(new Predicate() { + @Override + public boolean test(Integer v) throws Exception { + return v % 2 != 0; + } + }); + + final Single getNextOdd = oddNumbers.first(0); + + TestSubscriber> ts = evenNumbers.concatMap(new Function>>() { + @Override + public Publisher> apply(Integer v) throws Exception { + return Single.zip( + Single.just(v), getNextOdd, + new BiFunction>() { + @Override + public List apply(Integer a, Integer b) throws Exception { + return Arrays.asList( a, b ); + } + } + ) + .toFlowable(); + } + }) + .takeWhile(new Predicate>() { + @Override + public boolean test(List v) throws Exception { + return v.get(0) < 20; + } + }) + .test(); + + ts + .assertResult( + Arrays.asList(0, 1), + Arrays.asList(2, 3), + Arrays.asList(4, 5), + Arrays.asList(6, 7), + Arrays.asList(8, 9), + Arrays.asList(10, 11), + Arrays.asList(12, 13), + Arrays.asList(14, 15), + Arrays.asList(16, 17), + Arrays.asList(18, 19) + ); + } + + @Test + @SuppressWarnings("unchecked") + public void splitCombineSubscriberChangeAfterOnNextFused() { + Flowable source = Flowable.range(0, 20) + .publish(10) + .refCount() + ; + + Flowable evenNumbers = source.filter(new Predicate() { + @Override + public boolean test(Integer v) throws Exception { + return v % 2 == 0; + } + }); + + Flowable oddNumbers = source.filter(new Predicate() { + @Override + public boolean test(Integer v) throws Exception { + return v % 2 != 0; + } + }); + + final Single getNextOdd = oddNumbers.first(0); + + TestSubscriber> ts = evenNumbers.concatMap(new Function>>() { + @Override + public Publisher> apply(Integer v) throws Exception { + return Single.zip( + Single.just(v), getNextOdd, + new BiFunction>() { + @Override + public List apply(Integer a, Integer b) throws Exception { + return Arrays.asList( a, b ); + } + } + ) + .toFlowable(); + } + }) + .takeWhile(new Predicate>() { + @Override + public boolean test(List v) throws Exception { + return v.get(0) < 20; + } + }) + .test(); + + ts + .assertResult( + Arrays.asList(0, 1), + Arrays.asList(2, 3), + Arrays.asList(4, 5), + Arrays.asList(6, 7), + Arrays.asList(8, 9), + Arrays.asList(10, 11), + Arrays.asList(12, 13), + Arrays.asList(14, 15), + Arrays.asList(16, 17), + Arrays.asList(18, 19) + ); + } + + @Test + public void altConnectCrash() { + try { + new FlowablePublishAlt(Flowable.empty(), 128) + .connect(new Consumer() { + @Override + public void accept(Disposable t) throws Exception { + throw new TestException(); + } + }); + fail("Should have thrown"); + } catch (TestException expected) { + // expected + } + } + + @Test + public void altConnectRace() { + for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) { + final ConnectableFlowable cf = + new FlowablePublishAlt(Flowable.never(), 128); + + Runnable r = new Runnable() { + @Override + public void run() { + cf.connect(); + } + }; + + TestHelper.race(r, r); + } + } + + @Test + public void fusedPollCrash() { + Flowable.range(1, 5) + .map(new Function() { + @Override + public Object apply(Integer v) throws Exception { + throw new TestException(); + } + }) + .compose(TestHelper.flowableStripBoundary()) + .publish() + .refCount() + .test() + .assertFailure(TestException.class); + } + + @Test + public void syncFusedNoRequest() { + Flowable.range(1, 5) + .publish(1) + .refCount() + .test() + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void normalBackpressuredPolls() { + Flowable.range(1, 5) + .hide() + .publish(1) + .refCount() + .test() + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void emptyHidden() { + Flowable.empty() + .hide() + .publish(1) + .refCount() + .test() + .assertResult(); + } + + @Test + public void emptyFused() { + Flowable.empty() + .publish(1) + .refCount() + .test() + .assertResult(); + } + + @Test + public void overflowQueueRefCount() { + new Flowable() { + @Override + protected void subscribeActual(Subscriber s) { + s.onSubscribe(new BooleanSubscription()); + s.onNext(1); + s.onNext(2); + } + } + .publish(1) + .refCount() + .test(0) + .requestMore(1) + .assertFailure(MissingBackpressureException.class, 1); + } + + @Test + public void doubleErrorRefCount() { + List errors = TestHelper.trackPluginErrors(); + try { + new Flowable() { + @Override + protected void subscribeActual(Subscriber s) { + s.onSubscribe(new BooleanSubscription()); + s.onError(new TestException("one")); + s.onError(new TestException("two")); + } + } + .publish(1) + .refCount() + .test(0) + .assertFailureAndMessage(TestException.class, "one"); + + TestHelper.assertUndeliverable(errors, 0, TestException.class, "two"); + assertEquals(1, errors.size()); + } finally { + RxJavaPlugins.reset(); + } + } +} diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowablePublishTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowablePublishTest.java index eac12749f5..80af00c66f 100644 --- a/src/test/java/io/reactivex/internal/operators/flowable/FlowablePublishTest.java +++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowablePublishTest.java @@ -39,6 +39,28 @@ public class FlowablePublishTest { + // This will undo the workaround so that the plain ObservablePublish is still + // tested. + @Before + @SuppressWarnings({ "rawtypes", "unchecked" }) + public void before() { + RxJavaPlugins.setOnConnectableFlowableAssembly(new Function() { + @Override + public ConnectableFlowable apply(ConnectableFlowable co) throws Exception { + if (co instanceof FlowablePublishAlt) { + FlowablePublishAlt fpa = (FlowablePublishAlt) co; + return FlowablePublish.create(Flowable.fromPublisher(fpa.source()), fpa.publishBufferSize()); + } + return co; + } + }); + } + + @After + public void after() { + RxJavaPlugins.setOnConnectableFlowableAssembly(null); + } + @Test public void testPublish() throws InterruptedException { final AtomicInteger counter = new AtomicInteger(); diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableRefCountAltTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableRefCountAltTest.java new file mode 100644 index 0000000000..e048d47650 --- /dev/null +++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowableRefCountAltTest.java @@ -0,0 +1,1447 @@ +/** + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.internal.operators.flowable; + +import static org.junit.Assert.*; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.*; + +import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; + +import org.junit.Test; +import org.mockito.InOrder; +import org.reactivestreams.*; + +import io.reactivex.*; +import io.reactivex.disposables.*; +import io.reactivex.exceptions.*; +import io.reactivex.flowables.ConnectableFlowable; +import io.reactivex.functions.*; +import io.reactivex.internal.functions.Functions; +import io.reactivex.internal.operators.flowable.FlowableRefCount.RefConnection; +import io.reactivex.internal.subscriptions.BooleanSubscription; +import io.reactivex.internal.util.ExceptionHelper; +import io.reactivex.plugins.RxJavaPlugins; +import io.reactivex.processors.*; +import io.reactivex.schedulers.*; +import io.reactivex.subscribers.TestSubscriber; + +public class FlowableRefCountAltTest { + + @Test + public void testRefCountAsync() { + final AtomicInteger subscribeCount = new AtomicInteger(); + final AtomicInteger nextCount = new AtomicInteger(); + Flowable r = Flowable.interval(0, 20, TimeUnit.MILLISECONDS) + .doOnSubscribe(new Consumer() { + @Override + public void accept(Subscription s) { + subscribeCount.incrementAndGet(); + } + }) + .doOnNext(new Consumer() { + @Override + public void accept(Long l) { + nextCount.incrementAndGet(); + } + }) + .publish().refCount(); + + final AtomicInteger receivedCount = new AtomicInteger(); + Disposable d1 = r.subscribe(new Consumer() { + @Override + public void accept(Long l) { + receivedCount.incrementAndGet(); + } + }); + + Disposable d2 = r.subscribe(); + + try { + Thread.sleep(10); + } catch (InterruptedException e) { + } + + for (;;) { + int a = nextCount.get(); + int b = receivedCount.get(); + if (a > 10 && a < 20 && a == b) { + break; + } + if (a >= 20) { + break; + } + try { + Thread.sleep(20); + } catch (InterruptedException e) { + } + } + // give time to emit + + // now unsubscribe + d2.dispose(); // unsubscribe s2 first as we're counting in 1 and there can be a race between unsubscribe and one subscriber getting a value but not the other + d1.dispose(); + + System.out.println("onNext: " + nextCount.get()); + + // should emit once for both subscribers + assertEquals(nextCount.get(), receivedCount.get()); + // only 1 subscribe + assertEquals(1, subscribeCount.get()); + } + + @Test + public void testRefCountSynchronous() { + final AtomicInteger subscribeCount = new AtomicInteger(); + final AtomicInteger nextCount = new AtomicInteger(); + Flowable r = Flowable.just(1, 2, 3, 4, 5, 6, 7, 8, 9) + .doOnSubscribe(new Consumer() { + @Override + public void accept(Subscription s) { + subscribeCount.incrementAndGet(); + } + }) + .doOnNext(new Consumer() { + @Override + public void accept(Integer l) { + nextCount.incrementAndGet(); + } + }) + .publish().refCount(); + + final AtomicInteger receivedCount = new AtomicInteger(); + Disposable d1 = r.subscribe(new Consumer() { + @Override + public void accept(Integer l) { + receivedCount.incrementAndGet(); + } + }); + + Disposable d2 = r.subscribe(); + + // give time to emit + try { + Thread.sleep(50); + } catch (InterruptedException e) { + } + + // now unsubscribe + d2.dispose(); // unsubscribe s2 first as we're counting in 1 and there can be a race between unsubscribe and one subscriber getting a value but not the other + d1.dispose(); + + System.out.println("onNext Count: " + nextCount.get()); + + // it will emit twice because it is synchronous + assertEquals(nextCount.get(), receivedCount.get() * 2); + // it will subscribe twice because it is synchronous + assertEquals(2, subscribeCount.get()); + } + + @Test + public void testRefCountSynchronousTake() { + final AtomicInteger nextCount = new AtomicInteger(); + Flowable r = Flowable.just(1, 2, 3, 4, 5, 6, 7, 8, 9) + .doOnNext(new Consumer() { + @Override + public void accept(Integer l) { + System.out.println("onNext --------> " + l); + nextCount.incrementAndGet(); + } + }) + .take(4) + .publish().refCount(); + + final AtomicInteger receivedCount = new AtomicInteger(); + r.subscribe(new Consumer() { + @Override + public void accept(Integer l) { + receivedCount.incrementAndGet(); + } + }); + + System.out.println("onNext: " + nextCount.get()); + + assertEquals(4, receivedCount.get()); + assertEquals(4, receivedCount.get()); + } + + @Test + public void testRepeat() { + final AtomicInteger subscribeCount = new AtomicInteger(); + final AtomicInteger unsubscribeCount = new AtomicInteger(); + Flowable r = Flowable.interval(0, 1, TimeUnit.MILLISECONDS) + .doOnSubscribe(new Consumer() { + @Override + public void accept(Subscription s) { + System.out.println("******************************* Subscribe received"); + // when we are subscribed + subscribeCount.incrementAndGet(); + } + }) + .doOnCancel(new Action() { + @Override + public void run() { + System.out.println("******************************* Unsubscribe received"); + // when we are unsubscribed + unsubscribeCount.incrementAndGet(); + } + }) + .publish().refCount(); + + for (int i = 0; i < 10; i++) { + TestSubscriber ts1 = new TestSubscriber(); + TestSubscriber ts2 = new TestSubscriber(); + r.subscribe(ts1); + r.subscribe(ts2); + try { + Thread.sleep(50); + } catch (InterruptedException e) { + } + ts1.dispose(); + ts2.dispose(); + ts1.assertNoErrors(); + ts2.assertNoErrors(); + assertTrue(ts1.valueCount() > 0); + assertTrue(ts2.valueCount() > 0); + } + + assertEquals(10, subscribeCount.get()); + assertEquals(10, unsubscribeCount.get()); + } + + @Test + public void testConnectUnsubscribe() throws InterruptedException { + final CountDownLatch unsubscribeLatch = new CountDownLatch(1); + final CountDownLatch subscribeLatch = new CountDownLatch(1); + + Flowable f = synchronousInterval() + .doOnSubscribe(new Consumer() { + @Override + public void accept(Subscription s) { + System.out.println("******************************* Subscribe received"); + // when we are subscribed + subscribeLatch.countDown(); + } + }) + .doOnCancel(new Action() { + @Override + public void run() { + System.out.println("******************************* Unsubscribe received"); + // when we are unsubscribed + unsubscribeLatch.countDown(); + } + }); + + TestSubscriber s = new TestSubscriber(); + f.publish().refCount().subscribeOn(Schedulers.newThread()).subscribe(s); + System.out.println("send unsubscribe"); + // wait until connected + subscribeLatch.await(); + // now unsubscribe + s.dispose(); + System.out.println("DONE sending unsubscribe ... now waiting"); + if (!unsubscribeLatch.await(3000, TimeUnit.MILLISECONDS)) { + System.out.println("Errors: " + s.errors()); + if (s.errors().size() > 0) { + s.errors().get(0).printStackTrace(); + } + fail("timed out waiting for unsubscribe"); + } + s.assertNoErrors(); + } + + @Test + public void testConnectUnsubscribeRaceConditionLoop() throws InterruptedException { + for (int i = 0; i < 100; i++) { + testConnectUnsubscribeRaceCondition(); + } + } + + @Test + public void testConnectUnsubscribeRaceCondition() throws InterruptedException { + final AtomicInteger subUnsubCount = new AtomicInteger(); + Flowable f = synchronousInterval() + .doOnCancel(new Action() { + @Override + public void run() { + System.out.println("******************************* Unsubscribe received"); + // when we are unsubscribed + subUnsubCount.decrementAndGet(); + } + }) + .doOnSubscribe(new Consumer() { + @Override + public void accept(Subscription s) { + System.out.println("******************************* SUBSCRIBE received"); + subUnsubCount.incrementAndGet(); + } + }); + + TestSubscriber s = new TestSubscriber(); + + f.publish().refCount().subscribeOn(Schedulers.computation()).subscribe(s); + System.out.println("send unsubscribe"); + // now immediately unsubscribe while subscribeOn is racing to subscribe + s.dispose(); + // this generally will mean it won't even subscribe as it is already unsubscribed by the time connect() gets scheduled + // give time to the counter to update + Thread.sleep(10); + // either we subscribed and then unsubscribed, or we didn't ever even subscribe + assertEquals(0, subUnsubCount.get()); + + System.out.println("DONE sending unsubscribe ... now waiting"); + System.out.println("Errors: " + s.errors()); + if (s.errors().size() > 0) { + s.errors().get(0).printStackTrace(); + } + s.assertNoErrors(); + } + + private Flowable synchronousInterval() { + return Flowable.unsafeCreate(new Publisher() { + @Override + public void subscribe(Subscriber subscriber) { + final AtomicBoolean cancel = new AtomicBoolean(); + subscriber.onSubscribe(new Subscription() { + @Override + public void request(long n) { + + } + + @Override + public void cancel() { + cancel.set(true); + } + + }); + for (;;) { + if (cancel.get()) { + break; + } + try { + Thread.sleep(100); + } catch (InterruptedException e) { + } + subscriber.onNext(1L); + } + } + }); + } + + @Test + public void onlyFirstShouldSubscribeAndLastUnsubscribe() { + final AtomicInteger subscriptionCount = new AtomicInteger(); + final AtomicInteger unsubscriptionCount = new AtomicInteger(); + Flowable flowable = Flowable.unsafeCreate(new Publisher() { + @Override + public void subscribe(Subscriber subscriber) { + subscriptionCount.incrementAndGet(); + subscriber.onSubscribe(new Subscription() { + @Override + public void request(long n) { + + } + + @Override + public void cancel() { + unsubscriptionCount.incrementAndGet(); + } + }); + } + }); + Flowable refCounted = flowable.publish().refCount(); + + Disposable first = refCounted.subscribe(); + assertEquals(1, subscriptionCount.get()); + + Disposable second = refCounted.subscribe(); + assertEquals(1, subscriptionCount.get()); + + first.dispose(); + assertEquals(0, unsubscriptionCount.get()); + + second.dispose(); + assertEquals(1, unsubscriptionCount.get()); + } + + @Test + public void testRefCount() { + TestScheduler s = new TestScheduler(); + Flowable interval = Flowable.interval(100, TimeUnit.MILLISECONDS, s).publish().refCount(); + + // subscribe list1 + final List list1 = new ArrayList(); + Disposable d1 = interval.subscribe(new Consumer() { + @Override + public void accept(Long t1) { + list1.add(t1); + } + }); + + s.advanceTimeBy(200, TimeUnit.MILLISECONDS); + + assertEquals(2, list1.size()); + assertEquals(0L, list1.get(0).longValue()); + assertEquals(1L, list1.get(1).longValue()); + + // subscribe list2 + final List list2 = new ArrayList(); + Disposable d2 = interval.subscribe(new Consumer() { + @Override + public void accept(Long t1) { + list2.add(t1); + } + }); + + s.advanceTimeBy(300, TimeUnit.MILLISECONDS); + + // list 1 should have 5 items + assertEquals(5, list1.size()); + assertEquals(2L, list1.get(2).longValue()); + assertEquals(3L, list1.get(3).longValue()); + assertEquals(4L, list1.get(4).longValue()); + + // list 2 should only have 3 items + assertEquals(3, list2.size()); + assertEquals(2L, list2.get(0).longValue()); + assertEquals(3L, list2.get(1).longValue()); + assertEquals(4L, list2.get(2).longValue()); + + // unsubscribe list1 + d1.dispose(); + + // advance further + s.advanceTimeBy(300, TimeUnit.MILLISECONDS); + + // list 1 should still have 5 items + assertEquals(5, list1.size()); + + // list 2 should have 6 items + assertEquals(6, list2.size()); + assertEquals(5L, list2.get(3).longValue()); + assertEquals(6L, list2.get(4).longValue()); + assertEquals(7L, list2.get(5).longValue()); + + // unsubscribe list2 + d2.dispose(); + + // advance further + s.advanceTimeBy(1000, TimeUnit.MILLISECONDS); + + // subscribing a new one should start over because the source should have been unsubscribed + // subscribe list3 + final List list3 = new ArrayList(); + interval.subscribe(new Consumer() { + @Override + public void accept(Long t1) { + list3.add(t1); + } + }); + + s.advanceTimeBy(200, TimeUnit.MILLISECONDS); + + assertEquals(2, list3.size()); + assertEquals(0L, list3.get(0).longValue()); + assertEquals(1L, list3.get(1).longValue()); + + } + + @Test + public void testAlreadyUnsubscribedClient() { + Subscriber done = CancelledSubscriber.INSTANCE; + + Subscriber subscriber = TestHelper.mockSubscriber(); + + Flowable result = Flowable.just(1).publish().refCount(); + + result.subscribe(done); + + result.subscribe(subscriber); + + verify(subscriber).onNext(1); + verify(subscriber).onComplete(); + verify(subscriber, never()).onError(any(Throwable.class)); + } + + @Test + public void testAlreadyUnsubscribedInterleavesWithClient() { + ReplayProcessor source = ReplayProcessor.create(); + + Subscriber done = CancelledSubscriber.INSTANCE; + + Subscriber subscriber = TestHelper.mockSubscriber(); + InOrder inOrder = inOrder(subscriber); + + Flowable result = source.publish().refCount(); + + result.subscribe(subscriber); + + source.onNext(1); + + result.subscribe(done); + + source.onNext(2); + source.onComplete(); + + inOrder.verify(subscriber).onNext(1); + inOrder.verify(subscriber).onNext(2); + inOrder.verify(subscriber).onComplete(); + verify(subscriber, never()).onError(any(Throwable.class)); + } + + @Test + public void testConnectDisconnectConnectAndSubjectState() { + Flowable f1 = Flowable.just(10); + Flowable f2 = Flowable.just(20); + Flowable combined = Flowable.combineLatest(f1, f2, new BiFunction() { + @Override + public Integer apply(Integer t1, Integer t2) { + return t1 + t2; + } + }) + .publish().refCount(); + + TestSubscriber ts1 = new TestSubscriber(); + TestSubscriber ts2 = new TestSubscriber(); + + combined.subscribe(ts1); + combined.subscribe(ts2); + + ts1.assertTerminated(); + ts1.assertNoErrors(); + ts1.assertValue(30); + + ts2.assertTerminated(); + ts2.assertNoErrors(); + ts2.assertValue(30); + } + + @Test(timeout = 10000) + public void testUpstreamErrorAllowsRetry() throws InterruptedException { + List errors = TestHelper.trackPluginErrors(); + try { + final AtomicInteger intervalSubscribed = new AtomicInteger(); + Flowable interval = + Flowable.interval(200, TimeUnit.MILLISECONDS) + .doOnSubscribe(new Consumer() { + @Override + public void accept(Subscription s) { + System.out.println("Subscribing to interval " + intervalSubscribed.incrementAndGet()); + } + } + ) + .flatMap(new Function>() { + @Override + public Publisher apply(Long t1) { + return Flowable.defer(new Callable>() { + @Override + public Publisher call() { + return Flowable.error(new TestException("Some exception")); + } + }); + } + }) + .onErrorResumeNext(new Function>() { + @Override + public Publisher apply(Throwable t1) { + return Flowable.error(t1); + } + }) + .publish() + .refCount(); + + interval + .doOnError(new Consumer() { + @Override + public void accept(Throwable t1) { + System.out.println("Subscriber 1 onError: " + t1); + } + }) + .retry(5) + .subscribe(new Consumer() { + @Override + public void accept(String t1) { + System.out.println("Subscriber 1: " + t1); + } + }); + Thread.sleep(100); + interval + .doOnError(new Consumer() { + @Override + public void accept(Throwable t1) { + System.out.println("Subscriber 2 onError: " + t1); + } + }) + .retry(5) + .subscribe(new Consumer() { + @Override + public void accept(String t1) { + System.out.println("Subscriber 2: " + t1); + } + }); + + Thread.sleep(1300); + + System.out.println(intervalSubscribed.get()); + assertEquals(6, intervalSubscribed.get()); + + TestHelper.assertError(errors, 0, OnErrorNotImplementedException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + private enum CancelledSubscriber implements FlowableSubscriber { + INSTANCE; + + @Override public void onSubscribe(Subscription s) { + s.cancel(); + } + + @Override public void onNext(Integer o) { + } + + @Override public void onError(Throwable t) { + } + + @Override public void onComplete() { + } + } + + @Test + public void disposed() { + TestHelper.checkDisposed(Flowable.just(1).publish().refCount()); + } + + @Test + public void noOpConnect() { + final int[] calls = { 0 }; + Flowable f = new ConnectableFlowable() { + @Override + public void connect(Consumer connection) { + calls[0]++; + } + + @Override + protected void subscribeActual(Subscriber subscriber) { + subscriber.onSubscribe(new BooleanSubscription()); + } + }.refCount(); + + f.test(); + f.test(); + + assertEquals(1, calls[0]); + } + + Flowable source; + + @Test + public void replayNoLeak() throws Exception { + Thread.sleep(100); + System.gc(); + Thread.sleep(100); + + long start = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); + + source = Flowable.fromCallable(new Callable() { + @Override + public Object call() throws Exception { + return new byte[100 * 1000 * 1000]; + } + }) + .replay(1) + .refCount(); + + source.subscribe(); + + Thread.sleep(100); + System.gc(); + Thread.sleep(100); + + long after = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); + + source = null; + assertTrue(String.format("%,3d -> %,3d%n", start, after), start + 20 * 1000 * 1000 > after); + } + + @Test + public void replayNoLeak2() throws Exception { + Thread.sleep(100); + System.gc(); + Thread.sleep(100); + + long start = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); + + source = Flowable.fromCallable(new Callable() { + @Override + public Object call() throws Exception { + return new byte[100 * 1000 * 1000]; + } + }).concatWith(Flowable.never()) + .replay(1) + .refCount(); + + Disposable d1 = source.subscribe(); + Disposable d2 = source.subscribe(); + + d1.dispose(); + d2.dispose(); + + d1 = null; + d2 = null; + + Thread.sleep(100); + System.gc(); + Thread.sleep(100); + + long after = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); + + source = null; + assertTrue(String.format("%,3d -> %,3d%n", start, after), start + 20 * 1000 * 1000 > after); + } + + static final class ExceptionData extends Exception { + private static final long serialVersionUID = -6763898015338136119L; + + public final Object data; + + ExceptionData(Object data) { + this.data = data; + } + } + + @Test + public void publishNoLeak() throws Exception { + Thread.sleep(100); + System.gc(); + Thread.sleep(100); + + long start = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); + + source = Flowable.fromCallable(new Callable() { + @Override + public Object call() throws Exception { + throw new ExceptionData(new byte[100 * 1000 * 1000]); + } + }) + .publish() + .refCount(); + + source.subscribe(Functions.emptyConsumer(), Functions.emptyConsumer()); + + Thread.sleep(100); + System.gc(); + Thread.sleep(200); + + long after = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); + + source = null; + + assertTrue(String.format("%,3d -> %,3d%n", start, after), start + 20 * 1000 * 1000 > after); + } + + @Test + public void publishNoLeak2() throws Exception { + System.gc(); + Thread.sleep(100); + + long start = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); + + source = Flowable.fromCallable(new Callable() { + @Override + public Object call() throws Exception { + return new byte[100 * 1000 * 1000]; + } + }).concatWith(Flowable.never()) + .publish() + .refCount(); + + Disposable d1 = source.test(); + Disposable d2 = source.test(); + + d1.dispose(); + d2.dispose(); + + d1 = null; + d2 = null; + + System.gc(); + Thread.sleep(100); + + long after = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); + + source = null; + assertTrue(String.format("%,3d -> %,3d%n", start, after), start + 20 * 1000 * 1000 > after); + } + + @Test + public void replayIsUnsubscribed() { + ConnectableFlowable cf = Flowable.just(1) + .replay(); + + if (cf instanceof Disposable) { + assertTrue(((Disposable)cf).isDisposed()); + + Disposable connection = cf.connect(); + + assertFalse(((Disposable)cf).isDisposed()); + + connection.dispose(); + + assertTrue(((Disposable)cf).isDisposed()); + } + } + + static final class BadFlowableSubscribe extends ConnectableFlowable { + + @Override + public void connect(Consumer connection) { + try { + connection.accept(Disposables.empty()); + } catch (Throwable ex) { + throw ExceptionHelper.wrapOrThrow(ex); + } + } + + @Override + protected void subscribeActual(Subscriber subscriber) { + throw new TestException("subscribeActual"); + } + } + + static final class BadFlowableDispose extends ConnectableFlowable implements Disposable { + + @Override + public void dispose() { + throw new TestException("dispose"); + } + + @Override + public boolean isDisposed() { + return false; + } + + @Override + public void connect(Consumer connection) { + try { + connection.accept(Disposables.empty()); + } catch (Throwable ex) { + throw ExceptionHelper.wrapOrThrow(ex); + } + } + + @Override + protected void subscribeActual(Subscriber subscriber) { + subscriber.onSubscribe(new BooleanSubscription()); + } + } + + static final class BadFlowableConnect extends ConnectableFlowable { + + @Override + public void connect(Consumer connection) { + throw new TestException("connect"); + } + + @Override + protected void subscribeActual(Subscriber subscriber) { + subscriber.onSubscribe(new BooleanSubscription()); + } + } + + @Test + public void badSourceSubscribe() { + List errors = TestHelper.trackPluginErrors(); + try { + BadFlowableSubscribe bo = new BadFlowableSubscribe(); + + try { + bo.refCount() + .test(); + fail("Should have thrown"); + } catch (NullPointerException ex) { + assertTrue(ex.getCause() instanceof TestException); + } + + TestHelper.assertUndeliverable(errors, 0, TestException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void badSourceDispose() { + BadFlowableDispose bf = new BadFlowableDispose(); + + try { + bf.refCount() + .test() + .cancel(); + fail("Should have thrown"); + } catch (TestException expected) { + } + } + + @Test + public void badSourceConnect() { + List errors = TestHelper.trackPluginErrors(); + try { + BadFlowableConnect bf = new BadFlowableConnect(); + + try { + bf.refCount() + .test(); + fail("Should have thrown"); + } catch (NullPointerException ex) { + assertTrue(ex.getCause() instanceof TestException); + } + + TestHelper.assertUndeliverable(errors, 0, TestException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + static final class BadFlowableSubscribe2 extends ConnectableFlowable { + + int count; + + @Override + public void connect(Consumer connection) { + try { + connection.accept(Disposables.empty()); + } catch (Throwable ex) { + throw ExceptionHelper.wrapOrThrow(ex); + } + } + + @Override + protected void subscribeActual(Subscriber subscriber) { + if (++count == 1) { + subscriber.onSubscribe(new BooleanSubscription()); + } else { + throw new TestException("subscribeActual"); + } + } + } + + @Test + public void badSourceSubscribe2() { + List errors = TestHelper.trackPluginErrors(); + try { + BadFlowableSubscribe2 bf = new BadFlowableSubscribe2(); + + Flowable f = bf.refCount(); + f.test(); + try { + f.test(); + fail("Should have thrown"); + } catch (NullPointerException ex) { + assertTrue(ex.getCause() instanceof TestException); + } + + TestHelper.assertUndeliverable(errors, 0, TestException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + static final class BadFlowableConnect2 extends ConnectableFlowable + implements Disposable { + + @Override + public void connect(Consumer connection) { + try { + connection.accept(Disposables.empty()); + } catch (Throwable ex) { + throw ExceptionHelper.wrapOrThrow(ex); + } + } + + @Override + protected void subscribeActual(Subscriber subscriber) { + subscriber.onSubscribe(new BooleanSubscription()); + subscriber.onComplete(); + } + + @Override + public void dispose() { + throw new TestException("dispose"); + } + + @Override + public boolean isDisposed() { + return false; + } + } + + @Test + public void badSourceCompleteDisconnect() { + List errors = TestHelper.trackPluginErrors(); + try { + BadFlowableConnect2 bf = new BadFlowableConnect2(); + + try { + bf.refCount() + .test(); + fail("Should have thrown"); + } catch (NullPointerException ex) { + assertTrue(ex.getCause() instanceof TestException); + } + + TestHelper.assertUndeliverable(errors, 0, TestException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test(timeout = 7500) + public void blockingSourceAsnycCancel() throws Exception { + BehaviorProcessor bp = BehaviorProcessor.createDefault(1); + + Flowable f = bp + .replay(1) + .refCount(); + + f.subscribe(); + + final AtomicBoolean interrupted = new AtomicBoolean(); + + f.switchMap(new Function>() { + @Override + public Publisher apply(Integer v) throws Exception { + return Flowable.create(new FlowableOnSubscribe() { + @Override + public void subscribe(FlowableEmitter emitter) throws Exception { + while (!emitter.isCancelled()) { + Thread.sleep(100); + } + interrupted.set(true); + } + }, BackpressureStrategy.MISSING); + } + }) + .takeUntil(Flowable.timer(500, TimeUnit.MILLISECONDS)) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + + assertTrue(interrupted.get()); + } + + @Test + public void byCount() { + final int[] subscriptions = { 0 }; + + Flowable source = Flowable.range(1, 5) + .doOnSubscribe(new Consumer() { + @Override + public void accept(Subscription s) throws Exception { + subscriptions[0]++; + } + }) + .publish() + .refCount(2); + + for (int i = 0; i < 3; i++) { + TestSubscriber ts1 = source.test(); + + ts1.assertEmpty(); + + TestSubscriber ts2 = source.test(); + + ts1.assertResult(1, 2, 3, 4, 5); + ts2.assertResult(1, 2, 3, 4, 5); + } + + assertEquals(3, subscriptions[0]); + } + + @Test + public void resubscribeBeforeTimeout() throws Exception { + final int[] subscriptions = { 0 }; + + PublishProcessor pp = PublishProcessor.create(); + + Flowable source = pp + .doOnSubscribe(new Consumer() { + @Override + public void accept(Subscription s) throws Exception { + subscriptions[0]++; + } + }) + .publish() + .refCount(500, TimeUnit.MILLISECONDS); + + TestSubscriber ts1 = source.test(0); + + assertEquals(1, subscriptions[0]); + + ts1.cancel(); + + Thread.sleep(100); + + ts1 = source.test(0); + + assertEquals(1, subscriptions[0]); + + Thread.sleep(500); + + assertEquals(1, subscriptions[0]); + + pp.onNext(1); + pp.onNext(2); + pp.onNext(3); + pp.onNext(4); + pp.onNext(5); + pp.onComplete(); + + ts1.requestMore(5) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void letitTimeout() throws Exception { + final int[] subscriptions = { 0 }; + + PublishProcessor pp = PublishProcessor.create(); + + Flowable source = pp + .doOnSubscribe(new Consumer() { + @Override + public void accept(Subscription s) throws Exception { + subscriptions[0]++; + } + }) + .publish() + .refCount(1, 100, TimeUnit.MILLISECONDS); + + TestSubscriber ts1 = source.test(0); + + assertEquals(1, subscriptions[0]); + + ts1.cancel(); + + assertTrue(pp.hasSubscribers()); + + Thread.sleep(200); + + assertFalse(pp.hasSubscribers()); + } + + @Test + public void error() { + Flowable.error(new IOException()) + .publish() + .refCount(500, TimeUnit.MILLISECONDS) + .test() + .assertFailure(IOException.class); + } + + @Test + public void comeAndGo() { + PublishProcessor pp = PublishProcessor.create(); + + Flowable source = pp + .publish() + .refCount(1); + + TestSubscriber ts1 = source.test(0); + + assertTrue(pp.hasSubscribers()); + + for (int i = 0; i < 3; i++) { + TestSubscriber ts2 = source.test(); + ts1.cancel(); + ts1 = ts2; + } + + ts1.cancel(); + + assertFalse(pp.hasSubscribers()); + } + + @Test + public void unsubscribeSubscribeRace() { + for (int i = 0; i < 1000; i++) { + + final Flowable source = Flowable.range(1, 5) + .replay() + .refCount(1) + ; + + final TestSubscriber ts1 = source.test(0); + + final TestSubscriber ts2 = new TestSubscriber(0); + + Runnable r1 = new Runnable() { + @Override + public void run() { + ts1.cancel(); + } + }; + + Runnable r2 = new Runnable() { + @Override + public void run() { + source.subscribe(ts2); + } + }; + + TestHelper.race(r1, r2, Schedulers.single()); + + ts2.requestMore(6) // FIXME RxJava replay() doesn't issue onComplete without request + .withTag("Round: " + i) + .assertResult(1, 2, 3, 4, 5); + } + } + + static final class BadFlowableDoubleOnX extends ConnectableFlowable + implements Disposable { + + @Override + public void connect(Consumer connection) { + try { + connection.accept(Disposables.empty()); + } catch (Throwable ex) { + throw ExceptionHelper.wrapOrThrow(ex); + } + } + + @Override + protected void subscribeActual(Subscriber subscriber) { + subscriber.onSubscribe(new BooleanSubscription()); + subscriber.onSubscribe(new BooleanSubscription()); + subscriber.onComplete(); + subscriber.onComplete(); + subscriber.onError(new TestException()); + } + + @Override + public void dispose() { + } + + @Override + public boolean isDisposed() { + return false; + } + } + + @Test + public void doubleOnX() { + List errors = TestHelper.trackPluginErrors(); + try { + new BadFlowableDoubleOnX() + .refCount() + .test() + .assertResult(); + + TestHelper.assertError(errors, 0, ProtocolViolationException.class); + TestHelper.assertUndeliverable(errors, 1, TestException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void doubleOnXCount() { + List errors = TestHelper.trackPluginErrors(); + try { + new BadFlowableDoubleOnX() + .refCount(1) + .test() + .assertResult(); + + TestHelper.assertError(errors, 0, ProtocolViolationException.class); + TestHelper.assertUndeliverable(errors, 1, TestException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void doubleOnXTime() { + List errors = TestHelper.trackPluginErrors(); + try { + new BadFlowableDoubleOnX() + .refCount(5, TimeUnit.SECONDS, Schedulers.single()) + .test() + .assertResult(); + + TestHelper.assertError(errors, 0, ProtocolViolationException.class); + TestHelper.assertUndeliverable(errors, 1, TestException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void cancelTerminateStateExclusion() { + FlowableRefCount o = (FlowableRefCount)PublishProcessor.create() + .publish() + .refCount(); + + o.cancel(null); + + RefConnection rc = new RefConnection(o); + o.connection = null; + rc.subscriberCount = 0; + o.timeout(rc); + + rc.subscriberCount = 1; + o.timeout(rc); + + o.connection = rc; + o.timeout(rc); + + rc.subscriberCount = 0; + o.timeout(rc); + + // ------------------- + + rc.subscriberCount = 2; + rc.connected = false; + o.connection = rc; + o.cancel(rc); + + rc.subscriberCount = 1; + rc.connected = false; + o.connection = rc; + o.cancel(rc); + + rc.subscriberCount = 2; + rc.connected = true; + o.connection = rc; + o.cancel(rc); + + rc.subscriberCount = 1; + rc.connected = true; + o.connection = rc; + rc.set(null); + o.cancel(rc); + + o.connection = rc; + o.cancel(new RefConnection(o)); + } + + @Test + public void replayRefCountShallBeThreadSafe() { + for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) { + Flowable flowable = Flowable.just(1).replay(1).refCount(); + + TestSubscriber ts1 = flowable + .subscribeOn(Schedulers.io()) + .test(); + + TestSubscriber ts2 = flowable + .subscribeOn(Schedulers.io()) + .test(); + + ts1 + .withTag("" + i) + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1); + + ts2 + .withTag("" + i) + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1); + } + } + + static final class TestConnectableFlowable extends ConnectableFlowable + implements Disposable { + + volatile boolean disposed; + + @Override + public void dispose() { + disposed = true; + } + + @Override + public boolean isDisposed() { + return disposed; + } + + @Override + public void connect(Consumer connection) { + // not relevant + } + + @Override + protected void subscribeActual(Subscriber subscriber) { + // not relevant + } + } + + @Test + public void timeoutDisposesSource() { + FlowableRefCount o = (FlowableRefCount)new TestConnectableFlowable().refCount(); + + RefConnection rc = new RefConnection(o); + o.connection = rc; + + o.timeout(rc); + + assertTrue(((Disposable)o.source).isDisposed()); + } + + @Test + public void disconnectBeforeConnect() { + BehaviorProcessor processor = BehaviorProcessor.create(); + + Flowable flowable = processor + .replay(1) + .refCount(); + + flowable.takeUntil(Flowable.just(1)).test(); + + processor.onNext(2); + + flowable.take(1).test().assertResult(2); + } + + @Test + public void publishRefCountShallBeThreadSafe() { + for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) { + Flowable flowable = Flowable.just(1).publish().refCount(); + + TestSubscriber subscriber1 = flowable + .subscribeOn(Schedulers.io()) + .test(); + + TestSubscriber subscriber2 = flowable + .subscribeOn(Schedulers.io()) + .test(); + + subscriber1 + .withTag("subscriber1 " + i) + .awaitDone(5, TimeUnit.SECONDS) + .assertNoErrors() + .assertComplete(); + + subscriber2 + .withTag("subscriber2 " + i) + .awaitDone(5, TimeUnit.SECONDS) + .assertNoErrors() + .assertComplete(); + } + } +} diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableRefCountTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableRefCountTest.java index 673a0f4add..88aa2b17c1 100644 --- a/src/test/java/io/reactivex/internal/operators/flowable/FlowableRefCountTest.java +++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowableRefCountTest.java @@ -23,7 +23,7 @@ import java.util.concurrent.*; import java.util.concurrent.atomic.*; -import org.junit.Test; +import org.junit.*; import org.mockito.InOrder; import org.reactivestreams.*; @@ -43,6 +43,28 @@ public class FlowableRefCountTest { + // This will undo the workaround so that the plain ObservablePublish is still + // tested. + @Before + @SuppressWarnings({ "rawtypes", "unchecked" }) + public void before() { + RxJavaPlugins.setOnConnectableFlowableAssembly(new Function() { + @Override + public ConnectableFlowable apply(ConnectableFlowable co) throws Exception { + if (co instanceof FlowablePublishAlt) { + FlowablePublishAlt fpa = (FlowablePublishAlt) co; + return FlowablePublish.create(Flowable.fromPublisher(fpa.source()), fpa.publishBufferSize()); + } + return co; + } + }); + } + + @After + public void after() { + RxJavaPlugins.setOnConnectableFlowableAssembly(null); + } + @Test public void testRefCountAsync() { final AtomicInteger subscribeCount = new AtomicInteger(); @@ -653,6 +675,7 @@ protected void subscribeActual(Subscriber subscriber) { @Test public void replayNoLeak() throws Exception { + Thread.sleep(100); System.gc(); Thread.sleep(100); @@ -669,6 +692,7 @@ public Object call() throws Exception { source.subscribe(); + Thread.sleep(100); System.gc(); Thread.sleep(100); @@ -680,6 +704,7 @@ public Object call() throws Exception { @Test public void replayNoLeak2() throws Exception { + Thread.sleep(100); System.gc(); Thread.sleep(100); @@ -703,6 +728,7 @@ public Object call() throws Exception { d1 = null; d2 = null; + Thread.sleep(100); System.gc(); Thread.sleep(100); @@ -724,6 +750,7 @@ static final class ExceptionData extends Exception { @Test public void publishNoLeak() throws Exception { + Thread.sleep(100); System.gc(); Thread.sleep(100); @@ -740,6 +767,7 @@ public Object call() throws Exception { source.subscribe(Functions.emptyConsumer(), Functions.emptyConsumer()); + Thread.sleep(100); System.gc(); Thread.sleep(100); diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservablePublishAltTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservablePublishAltTest.java new file mode 100644 index 0000000000..b268e5b2ed --- /dev/null +++ b/src/test/java/io/reactivex/internal/operators/observable/ObservablePublishAltTest.java @@ -0,0 +1,794 @@ +/** + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.internal.operators.observable; + +import static org.junit.Assert.*; + +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; + +import org.junit.Test; + +import io.reactivex.*; +import io.reactivex.Observable; +import io.reactivex.Observer; +import io.reactivex.disposables.*; +import io.reactivex.exceptions.TestException; +import io.reactivex.functions.*; +import io.reactivex.internal.functions.Functions; +import io.reactivex.internal.fuseable.HasUpstreamObservableSource; +import io.reactivex.observables.ConnectableObservable; +import io.reactivex.observers.TestObserver; +import io.reactivex.plugins.RxJavaPlugins; +import io.reactivex.schedulers.*; +import io.reactivex.subjects.PublishSubject; + +public class ObservablePublishAltTest { + + @Test + public void testPublish() throws InterruptedException { + final AtomicInteger counter = new AtomicInteger(); + ConnectableObservable o = Observable.unsafeCreate(new ObservableSource() { + + @Override + public void subscribe(final Observer observer) { + observer.onSubscribe(Disposables.empty()); + new Thread(new Runnable() { + + @Override + public void run() { + counter.incrementAndGet(); + observer.onNext("one"); + observer.onComplete(); + } + }).start(); + } + }).publish(); + + final CountDownLatch latch = new CountDownLatch(2); + + // subscribe once + o.subscribe(new Consumer() { + + @Override + public void accept(String v) { + assertEquals("one", v); + latch.countDown(); + } + }); + + // subscribe again + o.subscribe(new Consumer() { + + @Override + public void accept(String v) { + assertEquals("one", v); + latch.countDown(); + } + }); + + Disposable connection = o.connect(); + try { + if (!latch.await(1000, TimeUnit.MILLISECONDS)) { + fail("subscriptions did not receive values"); + } + assertEquals(1, counter.get()); + } finally { + connection.dispose(); + } + } + + @Test + public void testBackpressureFastSlow() { + ConnectableObservable is = Observable.range(1, Flowable.bufferSize() * 2).publish(); + Observable fast = is.observeOn(Schedulers.computation()) + .doOnComplete(new Action() { + @Override + public void run() { + System.out.println("^^^^^^^^^^^^^ completed FAST"); + } + }); + + Observable slow = is.observeOn(Schedulers.computation()).map(new Function() { + int c; + + @Override + public Integer apply(Integer i) { + if (c == 0) { + try { + Thread.sleep(500); + } catch (InterruptedException e) { + } + } + c++; + return i; + } + + }).doOnComplete(new Action() { + + @Override + public void run() { + System.out.println("^^^^^^^^^^^^^ completed SLOW"); + } + + }); + + TestObserver to = new TestObserver(); + Observable.merge(fast, slow).subscribe(to); + is.connect(); + to.awaitTerminalEvent(); + to.assertNoErrors(); + assertEquals(Flowable.bufferSize() * 4, to.valueCount()); + } + + // use case from https://github.com/ReactiveX/RxJava/issues/1732 + @Test + public void testTakeUntilWithPublishedStreamUsingSelector() { + final AtomicInteger emitted = new AtomicInteger(); + Observable xs = Observable.range(0, Flowable.bufferSize() * 2).doOnNext(new Consumer() { + + @Override + public void accept(Integer t1) { + emitted.incrementAndGet(); + } + + }); + TestObserver to = new TestObserver(); + xs.publish(new Function, Observable>() { + + @Override + public Observable apply(Observable xs) { + return xs.takeUntil(xs.skipWhile(new Predicate() { + + @Override + public boolean test(Integer i) { + return i <= 3; + } + + })); + } + + }).subscribe(to); + to.awaitTerminalEvent(); + to.assertNoErrors(); + to.assertValues(0, 1, 2, 3); + assertEquals(5, emitted.get()); + System.out.println(to.values()); + } + + // use case from https://github.com/ReactiveX/RxJava/issues/1732 + @Test + public void testTakeUntilWithPublishedStream() { + Observable xs = Observable.range(0, Flowable.bufferSize() * 2); + TestObserver to = new TestObserver(); + ConnectableObservable xsp = xs.publish(); + xsp.takeUntil(xsp.skipWhile(new Predicate() { + + @Override + public boolean test(Integer i) { + return i <= 3; + } + + })).subscribe(to); + xsp.connect(); + System.out.println(to.values()); + } + + @Test(timeout = 10000) + public void testBackpressureTwoConsumers() { + final AtomicInteger sourceEmission = new AtomicInteger(); + final AtomicBoolean sourceUnsubscribed = new AtomicBoolean(); + final Observable source = Observable.range(1, 100) + .doOnNext(new Consumer() { + @Override + public void accept(Integer t1) { + sourceEmission.incrementAndGet(); + } + }) + .doOnDispose(new Action() { + @Override + public void run() { + sourceUnsubscribed.set(true); + } + }).share(); + ; + + final AtomicBoolean child1Unsubscribed = new AtomicBoolean(); + final AtomicBoolean child2Unsubscribed = new AtomicBoolean(); + + final TestObserver to2 = new TestObserver(); + + final TestObserver to1 = new TestObserver() { + @Override + public void onNext(Integer t) { + if (valueCount() == 2) { + source.doOnDispose(new Action() { + @Override + public void run() { + child2Unsubscribed.set(true); + } + }).take(5).subscribe(to2); + } + super.onNext(t); + } + }; + + source.doOnDispose(new Action() { + @Override + public void run() { + child1Unsubscribed.set(true); + } + }).take(5) + .subscribe(to1); + + to1.awaitTerminalEvent(); + to2.awaitTerminalEvent(); + + to1.assertNoErrors(); + to2.assertNoErrors(); + + assertTrue(sourceUnsubscribed.get()); + assertTrue(child1Unsubscribed.get()); + assertTrue(child2Unsubscribed.get()); + + to1.assertValues(1, 2, 3, 4, 5); + to2.assertValues(4, 5, 6, 7, 8); + + assertEquals(8, sourceEmission.get()); + } + + @Test + public void testConnectWithNoSubscriber() { + TestScheduler scheduler = new TestScheduler(); + ConnectableObservable co = Observable.interval(10, 10, TimeUnit.MILLISECONDS, scheduler).take(3).publish(); + co.connect(); + // Emit 0 + scheduler.advanceTimeBy(15, TimeUnit.MILLISECONDS); + TestObserver to = new TestObserver(); + co.subscribe(to); + // Emit 1 and 2 + scheduler.advanceTimeBy(50, TimeUnit.MILLISECONDS); + to.assertValues(1L, 2L); + to.assertNoErrors(); + to.assertTerminated(); + } + + @Test + public void testSubscribeAfterDisconnectThenConnect() { + ConnectableObservable source = Observable.just(1).publish(); + + TestObserver to1 = new TestObserver(); + + source.subscribe(to1); + + Disposable connection = source.connect(); + + to1.assertValue(1); + to1.assertNoErrors(); + to1.assertTerminated(); + + TestObserver to2 = new TestObserver(); + + source.subscribe(to2); + + Disposable connection2 = source.connect(); + + to2.assertValue(1); + to2.assertNoErrors(); + to2.assertTerminated(); + + System.out.println(connection); + System.out.println(connection2); + } + + @Test + public void testNoSubscriberRetentionOnCompleted() { + ObservablePublish source = (ObservablePublish)Observable.just(1).publish(); + + TestObserver to1 = new TestObserver(); + + source.subscribe(to1); + + to1.assertNoValues(); + to1.assertNoErrors(); + to1.assertNotComplete(); + + source.connect(); + + to1.assertValue(1); + to1.assertNoErrors(); + to1.assertTerminated(); + + assertNull(source.current.get()); + } + + @Test + public void testNonNullConnection() { + ConnectableObservable source = Observable.never().publish(); + + assertNotNull(source.connect()); + assertNotNull(source.connect()); + } + + @Test + public void testNoDisconnectSomeoneElse() { + ConnectableObservable source = Observable.never().publish(); + + Disposable connection1 = source.connect(); + Disposable connection2 = source.connect(); + + connection1.dispose(); + + Disposable connection3 = source.connect(); + + connection2.dispose(); + + assertTrue(checkPublishDisposed(connection1)); + assertTrue(checkPublishDisposed(connection2)); + assertFalse(checkPublishDisposed(connection3)); + } + + @SuppressWarnings("unchecked") + static boolean checkPublishDisposed(Disposable d) { + return ((ObservablePublish.PublishObserver)d).isDisposed(); + } + + @Test + public void testConnectIsIdempotent() { + final AtomicInteger calls = new AtomicInteger(); + Observable source = Observable.unsafeCreate(new ObservableSource() { + @Override + public void subscribe(Observer t) { + t.onSubscribe(Disposables.empty()); + calls.getAndIncrement(); + } + }); + + ConnectableObservable conn = source.publish(); + + assertEquals(0, calls.get()); + + conn.connect(); + conn.connect(); + + assertEquals(1, calls.get()); + + conn.connect().dispose(); + + conn.connect(); + conn.connect(); + + assertEquals(2, calls.get()); + } + + @Test + public void testObserveOn() { + ConnectableObservable co = Observable.range(0, 1000).publish(); + Observable obs = co.observeOn(Schedulers.computation()); + for (int i = 0; i < 1000; i++) { + for (int j = 1; j < 6; j++) { + List> tos = new ArrayList>(); + for (int k = 1; k < j; k++) { + TestObserver to = new TestObserver(); + tos.add(to); + obs.subscribe(to); + } + + Disposable connection = co.connect(); + + for (TestObserver to : tos) { + to.awaitTerminalEvent(2, TimeUnit.SECONDS); + to.assertTerminated(); + to.assertNoErrors(); + assertEquals(1000, to.valueCount()); + } + connection.dispose(); + } + } + } + + @Test + public void preNextConnect() { + for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { + + final ConnectableObservable co = Observable.empty().publish(); + + co.connect(); + + Runnable r1 = new Runnable() { + @Override + public void run() { + co.test(); + } + }; + + TestHelper.race(r1, r1); + } + } + + @Test + public void connectRace() { + for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { + + final ConnectableObservable co = Observable.empty().publish(); + + Runnable r1 = new Runnable() { + @Override + public void run() { + co.connect(); + } + }; + + TestHelper.race(r1, r1); + } + } + + @Test + public void selectorCrash() { + Observable.just(1).publish(new Function, ObservableSource>() { + @Override + public ObservableSource apply(Observable v) throws Exception { + throw new TestException(); + } + }) + .test() + .assertFailure(TestException.class); + } + + @Test + public void source() { + Observable o = Observable.never(); + + assertSame(o, (((HasUpstreamObservableSource)o.publish()).source())); + } + + @Test + public void connectThrows() { + ConnectableObservable co = Observable.empty().publish(); + try { + co.connect(new Consumer() { + @Override + public void accept(Disposable d) throws Exception { + throw new TestException(); + } + }); + } catch (TestException ex) { + // expected + } + } + + @Test + public void addRemoveRace() { + for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { + + final ConnectableObservable co = Observable.empty().publish(); + + final TestObserver to = co.test(); + + final TestObserver to2 = new TestObserver(); + + Runnable r1 = new Runnable() { + @Override + public void run() { + co.subscribe(to2); + } + }; + + Runnable r2 = new Runnable() { + @Override + public void run() { + to.cancel(); + } + }; + + TestHelper.race(r1, r2); + } + } + + @Test + public void disposeOnArrival() { + ConnectableObservable co = Observable.empty().publish(); + + co.test(true).assertEmpty(); + } + + @Test + public void dispose() { + TestHelper.checkDisposed(Observable.never().publish()); + + TestHelper.checkDisposed(Observable.never().publish(Functions.>identity())); + } + + @Test + public void empty() { + ConnectableObservable co = Observable.empty().publish(); + + co.connect(); + } + + @Test + public void take() { + ConnectableObservable co = Observable.range(1, 2).publish(); + + TestObserver to = co.take(1).test(); + + co.connect(); + + to.assertResult(1); + } + + @Test + public void just() { + final PublishSubject ps = PublishSubject.create(); + + ConnectableObservable co = ps.publish(); + + TestObserver to = new TestObserver() { + @Override + public void onNext(Integer t) { + super.onNext(t); + ps.onComplete(); + } + }; + + co.subscribe(to); + co.connect(); + + ps.onNext(1); + + to.assertResult(1); + } + + @Test + public void nextCancelRace() { + for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { + + final PublishSubject ps = PublishSubject.create(); + + final ConnectableObservable co = ps.publish(); + + final TestObserver to = co.test(); + + Runnable r1 = new Runnable() { + @Override + public void run() { + ps.onNext(1); + } + }; + + Runnable r2 = new Runnable() { + @Override + public void run() { + to.cancel(); + } + }; + + TestHelper.race(r1, r2); + } + } + + @Test + public void badSource() { + List errors = TestHelper.trackPluginErrors(); + try { + new Observable() { + @Override + protected void subscribeActual(Observer observer) { + observer.onSubscribe(Disposables.empty()); + observer.onNext(1); + observer.onComplete(); + observer.onNext(2); + observer.onError(new TestException()); + observer.onComplete(); + } + } + .publish() + .autoConnect() + .test() + .assertResult(1); + + TestHelper.assertUndeliverable(errors, 0, TestException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void noErrorLoss() { + List errors = TestHelper.trackPluginErrors(); + try { + ConnectableObservable co = Observable.error(new TestException()).publish(); + + co.connect(); + + TestHelper.assertUndeliverable(errors, 0, TestException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void subscribeDisconnectRace() { + for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { + + final PublishSubject ps = PublishSubject.create(); + + final ConnectableObservable co = ps.publish(); + + final Disposable d = co.connect(); + final TestObserver to = new TestObserver(); + + Runnable r1 = new Runnable() { + @Override + public void run() { + d.dispose(); + } + }; + + Runnable r2 = new Runnable() { + @Override + public void run() { + co.subscribe(to); + } + }; + + TestHelper.race(r1, r2); + } + } + + @Test + public void selectorDisconnectsIndependentSource() { + PublishSubject ps = PublishSubject.create(); + + ps.publish(new Function, ObservableSource>() { + @Override + public ObservableSource apply(Observable v) throws Exception { + return Observable.range(1, 2); + } + }) + .test() + .assertResult(1, 2); + + assertFalse(ps.hasObservers()); + } + + @Test(timeout = 5000) + public void selectorLatecommer() { + Observable.range(1, 5) + .publish(new Function, ObservableSource>() { + @Override + public ObservableSource apply(Observable v) throws Exception { + return v.concatWith(v); + } + }) + .test() + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void mainError() { + Observable.error(new TestException()) + .publish(Functions.>identity()) + .test() + .assertFailure(TestException.class); + } + + @Test + public void selectorInnerError() { + PublishSubject ps = PublishSubject.create(); + + ps.publish(new Function, ObservableSource>() { + @Override + public ObservableSource apply(Observable v) throws Exception { + return Observable.error(new TestException()); + } + }) + .test() + .assertFailure(TestException.class); + + assertFalse(ps.hasObservers()); + } + + @Test + public void delayedUpstreamOnSubscribe() { + final Observer[] sub = { null }; + + new Observable() { + @Override + protected void subscribeActual(Observer observer) { + sub[0] = observer; + } + } + .publish() + .connect() + .dispose(); + + Disposable bs = Disposables.empty(); + + sub[0].onSubscribe(bs); + + assertTrue(bs.isDisposed()); + } + + @Test + public void doubleOnSubscribe() { + TestHelper.checkDoubleOnSubscribeObservable(new Function, ObservableSource>() { + @Override + public ObservableSource apply(final Observable o) + throws Exception { + return Observable.never().publish(new Function, ObservableSource>() { + @Override + public ObservableSource apply(Observable v) + throws Exception { + return o; + } + }); + } + } + ); + } + + @Test + public void disposedUpfront() { + ConnectableObservable co = Observable.just(1) + .concatWith(Observable.never()) + .publish(); + + TestObserver to1 = co.test(); + + TestObserver to2 = co.test(true); + + co.connect(); + + to1.assertValuesOnly(1); + + to2.assertEmpty(); + + ((ObservablePublish)co).current.get().remove(null); + } + + @Test + public void altConnectCrash() { + try { + new ObservablePublishAlt(Observable.empty()) + .connect(new Consumer() { + @Override + public void accept(Disposable t) throws Exception { + throw new TestException(); + } + }); + fail("Should have thrown"); + } catch (TestException expected) { + // expected + } + } + + @Test + public void altConnectRace() { + for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) { + final ConnectableObservable co = + new ObservablePublishAlt(Observable.never()); + + Runnable r = new Runnable() { + @Override + public void run() { + co.connect(); + } + }; + + TestHelper.race(r, r); + } + } +} diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservablePublishTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservablePublishTest.java index 2f2a0e677e..7534f07346 100644 --- a/src/test/java/io/reactivex/internal/operators/observable/ObservablePublishTest.java +++ b/src/test/java/io/reactivex/internal/operators/observable/ObservablePublishTest.java @@ -19,7 +19,7 @@ import java.util.concurrent.*; import java.util.concurrent.atomic.*; -import org.junit.Test; +import org.junit.*; import io.reactivex.*; import io.reactivex.Observable; @@ -37,6 +37,27 @@ public class ObservablePublishTest { + // This will undo the workaround so that the plain ObservablePublish is still + // tested. + @Before + @SuppressWarnings({ "rawtypes", "unchecked" }) + public void before() { + RxJavaPlugins.setOnConnectableObservableAssembly(new Function() { + @Override + public ConnectableObservable apply(ConnectableObservable co) throws Exception { + if (co instanceof ObservablePublishAlt) { + return ObservablePublish.create(((ObservablePublishAlt)co).source()); + } + return co; + } + }); + } + + @After + public void after() { + RxJavaPlugins.setOnConnectableObservableAssembly(null); + } + @Test public void testPublish() throws InterruptedException { final AtomicInteger counter = new AtomicInteger(); diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountAltTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountAltTest.java new file mode 100644 index 0000000000..effe8ef206 --- /dev/null +++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountAltTest.java @@ -0,0 +1,1394 @@ +/** + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.internal.operators.observable; + +import static org.junit.Assert.*; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.*; + +import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; + +import org.junit.Test; +import org.mockito.InOrder; + +import io.reactivex.*; +import io.reactivex.Observable; +import io.reactivex.Observer; +import io.reactivex.disposables.*; +import io.reactivex.exceptions.*; +import io.reactivex.functions.*; +import io.reactivex.internal.functions.Functions; +import io.reactivex.internal.operators.observable.ObservableRefCount.RefConnection; +import io.reactivex.internal.util.ExceptionHelper; +import io.reactivex.observables.ConnectableObservable; +import io.reactivex.observers.TestObserver; +import io.reactivex.plugins.RxJavaPlugins; +import io.reactivex.schedulers.*; +import io.reactivex.subjects.*; + +public class ObservableRefCountAltTest { + + @Test + public void testRefCountAsync() { + final AtomicInteger subscribeCount = new AtomicInteger(); + final AtomicInteger nextCount = new AtomicInteger(); + Observable r = Observable.interval(0, 25, TimeUnit.MILLISECONDS) + .doOnSubscribe(new Consumer() { + @Override + public void accept(Disposable d) { + subscribeCount.incrementAndGet(); + } + }) + .doOnNext(new Consumer() { + @Override + public void accept(Long l) { + nextCount.incrementAndGet(); + } + }) + .publish().refCount(); + + final AtomicInteger receivedCount = new AtomicInteger(); + Disposable d1 = r.subscribe(new Consumer() { + @Override + public void accept(Long l) { + receivedCount.incrementAndGet(); + } + }); + + Disposable d2 = r.subscribe(); + + // give time to emit + try { + Thread.sleep(260); + } catch (InterruptedException e) { + } + + // now unsubscribe + d2.dispose(); // unsubscribe s2 first as we're counting in 1 and there can be a race between unsubscribe and one Observer getting a value but not the other + d1.dispose(); + + System.out.println("onNext: " + nextCount.get()); + + // should emit once for both subscribers + assertEquals(nextCount.get(), receivedCount.get()); + // only 1 subscribe + assertEquals(1, subscribeCount.get()); + } + + @Test + public void testRefCountSynchronous() { + final AtomicInteger subscribeCount = new AtomicInteger(); + final AtomicInteger nextCount = new AtomicInteger(); + Observable r = Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9) + .doOnSubscribe(new Consumer() { + @Override + public void accept(Disposable d) { + subscribeCount.incrementAndGet(); + } + }) + .doOnNext(new Consumer() { + @Override + public void accept(Integer l) { + nextCount.incrementAndGet(); + } + }) + .publish().refCount(); + + final AtomicInteger receivedCount = new AtomicInteger(); + Disposable d1 = r.subscribe(new Consumer() { + @Override + public void accept(Integer l) { + receivedCount.incrementAndGet(); + } + }); + + Disposable d2 = r.subscribe(); + + // give time to emit + try { + Thread.sleep(50); + } catch (InterruptedException e) { + } + + // now unsubscribe + d2.dispose(); // unsubscribe s2 first as we're counting in 1 and there can be a race between unsubscribe and one Observer getting a value but not the other + d1.dispose(); + + System.out.println("onNext Count: " + nextCount.get()); + + // it will emit twice because it is synchronous + assertEquals(nextCount.get(), receivedCount.get() * 2); + // it will subscribe twice because it is synchronous + assertEquals(2, subscribeCount.get()); + } + + @Test + public void testRefCountSynchronousTake() { + final AtomicInteger nextCount = new AtomicInteger(); + Observable r = Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9) + .doOnNext(new Consumer() { + @Override + public void accept(Integer l) { + System.out.println("onNext --------> " + l); + nextCount.incrementAndGet(); + } + }) + .take(4) + .publish().refCount(); + + final AtomicInteger receivedCount = new AtomicInteger(); + r.subscribe(new Consumer() { + @Override + public void accept(Integer l) { + receivedCount.incrementAndGet(); + } + }); + + System.out.println("onNext: " + nextCount.get()); + + assertEquals(4, receivedCount.get()); + assertEquals(4, receivedCount.get()); + } + + @Test + public void testRepeat() { + final AtomicInteger subscribeCount = new AtomicInteger(); + final AtomicInteger unsubscribeCount = new AtomicInteger(); + Observable r = Observable.interval(0, 1, TimeUnit.MILLISECONDS) + .doOnSubscribe(new Consumer() { + @Override + public void accept(Disposable d) { + System.out.println("******************************* Subscribe received"); + // when we are subscribed + subscribeCount.incrementAndGet(); + } + }) + .doOnDispose(new Action() { + @Override + public void run() { + System.out.println("******************************* Unsubscribe received"); + // when we are unsubscribed + unsubscribeCount.incrementAndGet(); + } + }) + .publish().refCount(); + + for (int i = 0; i < 10; i++) { + TestObserver to1 = new TestObserver(); + TestObserver to2 = new TestObserver(); + r.subscribe(to1); + r.subscribe(to2); + try { + Thread.sleep(50); + } catch (InterruptedException e) { + } + to1.dispose(); + to2.dispose(); + to1.assertNoErrors(); + to2.assertNoErrors(); + assertTrue(to1.valueCount() > 0); + assertTrue(to2.valueCount() > 0); + } + + assertEquals(10, subscribeCount.get()); + assertEquals(10, unsubscribeCount.get()); + } + + @Test + public void testConnectUnsubscribe() throws InterruptedException { + final CountDownLatch unsubscribeLatch = new CountDownLatch(1); + final CountDownLatch subscribeLatch = new CountDownLatch(1); + + Observable o = synchronousInterval() + .doOnSubscribe(new Consumer() { + @Override + public void accept(Disposable d) { + System.out.println("******************************* Subscribe received"); + // when we are subscribed + subscribeLatch.countDown(); + } + }) + .doOnDispose(new Action() { + @Override + public void run() { + System.out.println("******************************* Unsubscribe received"); + // when we are unsubscribed + unsubscribeLatch.countDown(); + } + }); + + TestObserver observer = new TestObserver(); + o.publish().refCount().subscribeOn(Schedulers.newThread()).subscribe(observer); + System.out.println("send unsubscribe"); + // wait until connected + subscribeLatch.await(); + // now unsubscribe + observer.dispose(); + System.out.println("DONE sending unsubscribe ... now waiting"); + if (!unsubscribeLatch.await(3000, TimeUnit.MILLISECONDS)) { + System.out.println("Errors: " + observer.errors()); + if (observer.errors().size() > 0) { + observer.errors().get(0).printStackTrace(); + } + fail("timed out waiting for unsubscribe"); + } + observer.assertNoErrors(); + } + + @Test + public void testConnectUnsubscribeRaceConditionLoop() throws InterruptedException { + for (int i = 0; i < 100; i++) { + testConnectUnsubscribeRaceCondition(); + } + } + + @Test + public void testConnectUnsubscribeRaceCondition() throws InterruptedException { + final AtomicInteger subUnsubCount = new AtomicInteger(); + Observable o = synchronousInterval() + .doOnDispose(new Action() { + @Override + public void run() { + System.out.println("******************************* Unsubscribe received"); + // when we are unsubscribed + subUnsubCount.decrementAndGet(); + } + }) + .doOnSubscribe(new Consumer() { + @Override + public void accept(Disposable d) { + System.out.println("******************************* SUBSCRIBE received"); + subUnsubCount.incrementAndGet(); + } + }); + + TestObserver observer = new TestObserver(); + + o.publish().refCount().subscribeOn(Schedulers.computation()).subscribe(observer); + System.out.println("send unsubscribe"); + // now immediately unsubscribe while subscribeOn is racing to subscribe + observer.dispose(); + + // this generally will mean it won't even subscribe as it is already unsubscribed by the time connect() gets scheduled + // give time to the counter to update + Thread.sleep(10); + + // make sure we wait a bit in case the counter is still nonzero + int counter = 200; + while (subUnsubCount.get() != 0 && counter-- != 0) { + Thread.sleep(10); + } + // either we subscribed and then unsubscribed, or we didn't ever even subscribe + assertEquals(0, subUnsubCount.get()); + + System.out.println("DONE sending unsubscribe ... now waiting"); + System.out.println("Errors: " + observer.errors()); + if (observer.errors().size() > 0) { + observer.errors().get(0).printStackTrace(); + } + observer.assertNoErrors(); + } + + private Observable synchronousInterval() { + return Observable.unsafeCreate(new ObservableSource() { + @Override + public void subscribe(Observer observer) { + final AtomicBoolean cancel = new AtomicBoolean(); + observer.onSubscribe(Disposables.fromRunnable(new Runnable() { + @Override + public void run() { + cancel.set(true); + } + })); + for (;;) { + if (cancel.get()) { + break; + } + try { + Thread.sleep(100); + } catch (InterruptedException e) { + } + observer.onNext(1L); + } + } + }); + } + + @Test + public void onlyFirstShouldSubscribeAndLastUnsubscribe() { + final AtomicInteger subscriptionCount = new AtomicInteger(); + final AtomicInteger unsubscriptionCount = new AtomicInteger(); + Observable o = Observable.unsafeCreate(new ObservableSource() { + @Override + public void subscribe(Observer observer) { + subscriptionCount.incrementAndGet(); + observer.onSubscribe(Disposables.fromRunnable(new Runnable() { + @Override + public void run() { + unsubscriptionCount.incrementAndGet(); + } + })); + } + }); + Observable refCounted = o.publish().refCount(); + + Disposable first = refCounted.subscribe(); + assertEquals(1, subscriptionCount.get()); + + Disposable second = refCounted.subscribe(); + assertEquals(1, subscriptionCount.get()); + + first.dispose(); + assertEquals(0, unsubscriptionCount.get()); + + second.dispose(); + assertEquals(1, unsubscriptionCount.get()); + } + + @Test + public void testRefCount() { + TestScheduler s = new TestScheduler(); + Observable interval = Observable.interval(100, TimeUnit.MILLISECONDS, s).publish().refCount(); + + // subscribe list1 + final List list1 = new ArrayList(); + Disposable d1 = interval.subscribe(new Consumer() { + @Override + public void accept(Long t1) { + list1.add(t1); + } + }); + + s.advanceTimeBy(200, TimeUnit.MILLISECONDS); + + assertEquals(2, list1.size()); + assertEquals(0L, list1.get(0).longValue()); + assertEquals(1L, list1.get(1).longValue()); + + // subscribe list2 + final List list2 = new ArrayList(); + Disposable d2 = interval.subscribe(new Consumer() { + @Override + public void accept(Long t1) { + list2.add(t1); + } + }); + + s.advanceTimeBy(300, TimeUnit.MILLISECONDS); + + // list 1 should have 5 items + assertEquals(5, list1.size()); + assertEquals(2L, list1.get(2).longValue()); + assertEquals(3L, list1.get(3).longValue()); + assertEquals(4L, list1.get(4).longValue()); + + // list 2 should only have 3 items + assertEquals(3, list2.size()); + assertEquals(2L, list2.get(0).longValue()); + assertEquals(3L, list2.get(1).longValue()); + assertEquals(4L, list2.get(2).longValue()); + + // unsubscribe list1 + d1.dispose(); + + // advance further + s.advanceTimeBy(300, TimeUnit.MILLISECONDS); + + // list 1 should still have 5 items + assertEquals(5, list1.size()); + + // list 2 should have 6 items + assertEquals(6, list2.size()); + assertEquals(5L, list2.get(3).longValue()); + assertEquals(6L, list2.get(4).longValue()); + assertEquals(7L, list2.get(5).longValue()); + + // unsubscribe list2 + d2.dispose(); + + // advance further + s.advanceTimeBy(1000, TimeUnit.MILLISECONDS); + + // subscribing a new one should start over because the source should have been unsubscribed + // subscribe list3 + final List list3 = new ArrayList(); + interval.subscribe(new Consumer() { + @Override + public void accept(Long t1) { + list3.add(t1); + } + }); + + s.advanceTimeBy(200, TimeUnit.MILLISECONDS); + + assertEquals(2, list3.size()); + assertEquals(0L, list3.get(0).longValue()); + assertEquals(1L, list3.get(1).longValue()); + + } + + @Test + public void testAlreadyUnsubscribedClient() { + Observer done = DisposingObserver.INSTANCE; + + Observer o = TestHelper.mockObserver(); + + Observable result = Observable.just(1).publish().refCount(); + + result.subscribe(done); + + result.subscribe(o); + + verify(o).onNext(1); + verify(o).onComplete(); + verify(o, never()).onError(any(Throwable.class)); + } + + @Test + public void testAlreadyUnsubscribedInterleavesWithClient() { + ReplaySubject source = ReplaySubject.create(); + + Observer done = DisposingObserver.INSTANCE; + + Observer o = TestHelper.mockObserver(); + InOrder inOrder = inOrder(o); + + Observable result = source.publish().refCount(); + + result.subscribe(o); + + source.onNext(1); + + result.subscribe(done); + + source.onNext(2); + source.onComplete(); + + inOrder.verify(o).onNext(1); + inOrder.verify(o).onNext(2); + inOrder.verify(o).onComplete(); + verify(o, never()).onError(any(Throwable.class)); + } + + @Test + public void testConnectDisconnectConnectAndSubjectState() { + Observable o1 = Observable.just(10); + Observable o2 = Observable.just(20); + Observable combined = Observable.combineLatest(o1, o2, new BiFunction() { + @Override + public Integer apply(Integer t1, Integer t2) { + return t1 + t2; + } + }) + .publish().refCount(); + + TestObserver to1 = new TestObserver(); + TestObserver to2 = new TestObserver(); + + combined.subscribe(to1); + combined.subscribe(to2); + + to1.assertTerminated(); + to1.assertNoErrors(); + to1.assertValue(30); + + to2.assertTerminated(); + to2.assertNoErrors(); + to2.assertValue(30); + } + + @Test(timeout = 10000) + public void testUpstreamErrorAllowsRetry() throws InterruptedException { + final AtomicInteger intervalSubscribed = new AtomicInteger(); + Observable interval = + Observable.interval(200, TimeUnit.MILLISECONDS) + .doOnSubscribe(new Consumer() { + @Override + public void accept(Disposable d) { + System.out.println("Subscribing to interval " + intervalSubscribed.incrementAndGet()); + } + } + ) + .flatMap(new Function>() { + @Override + public Observable apply(Long t1) { + return Observable.defer(new Callable>() { + @Override + public Observable call() { + return Observable.error(new Exception("Some exception")); + } + }); + } + }) + .onErrorResumeNext(new Function>() { + @Override + public Observable apply(Throwable t1) { + return Observable.error(t1); + } + }) + .publish() + .refCount(); + + interval + .doOnError(new Consumer() { + @Override + public void accept(Throwable t1) { + System.out.println("Observer 1 onError: " + t1); + } + }) + .retry(5) + .subscribe(new Consumer() { + @Override + public void accept(String t1) { + System.out.println("Observer 1: " + t1); + } + }); + Thread.sleep(100); + interval + .doOnError(new Consumer() { + @Override + public void accept(Throwable t1) { + System.out.println("Observer 2 onError: " + t1); + } + }) + .retry(5) + .subscribe(new Consumer() { + @Override + public void accept(String t1) { + System.out.println("Observer 2: " + t1); + } + }); + + Thread.sleep(1300); + + System.out.println(intervalSubscribed.get()); + assertEquals(6, intervalSubscribed.get()); + } + + private enum DisposingObserver implements Observer { + INSTANCE; + + @Override + public void onSubscribe(Disposable d) { + d.dispose(); + } + + @Override + public void onNext(Integer t) { + } + + @Override + public void onError(Throwable t) { + } + + @Override + public void onComplete() { + } + } + + @Test + public void disposed() { + TestHelper.checkDisposed(Observable.just(1).publish().refCount()); + } + + @Test + public void noOpConnect() { + final int[] calls = { 0 }; + Observable o = new ConnectableObservable() { + @Override + public void connect(Consumer connection) { + calls[0]++; + } + + @Override + protected void subscribeActual(Observer observer) { + observer.onSubscribe(Disposables.disposed()); + } + }.refCount(); + + o.test(); + o.test(); + + assertEquals(1, calls[0]); + } + Observable source; + + @Test + public void replayNoLeak() throws Exception { + System.gc(); + Thread.sleep(100); + + long start = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); + + source = Observable.fromCallable(new Callable() { + @Override + public Object call() throws Exception { + return new byte[100 * 1000 * 1000]; + } + }) + .replay(1) + .refCount(); + + source.subscribe(); + + System.gc(); + Thread.sleep(100); + + long after = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); + + source = null; + assertTrue(String.format("%,3d -> %,3d%n", start, after), start + 20 * 1000 * 1000 > after); + } + + @Test + public void replayNoLeak2() throws Exception { + System.gc(); + Thread.sleep(100); + + long start = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); + + source = Observable.fromCallable(new Callable() { + @Override + public Object call() throws Exception { + return new byte[100 * 1000 * 1000]; + } + }).concatWith(Observable.never()) + .replay(1) + .refCount(); + + Disposable d1 = source.subscribe(); + Disposable d2 = source.subscribe(); + + d1.dispose(); + d2.dispose(); + + d1 = null; + d2 = null; + + System.gc(); + Thread.sleep(100); + + long after = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); + + source = null; + assertTrue(String.format("%,3d -> %,3d%n", start, after), start + 20 * 1000 * 1000 > after); + } + + static final class ExceptionData extends Exception { + private static final long serialVersionUID = -6763898015338136119L; + + public final Object data; + + ExceptionData(Object data) { + this.data = data; + } + } + + @Test + public void publishNoLeak() throws Exception { + System.gc(); + Thread.sleep(100); + + long start = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); + + source = Observable.fromCallable(new Callable() { + @Override + public Object call() throws Exception { + throw new ExceptionData(new byte[100 * 1000 * 1000]); + } + }) + .publish() + .refCount(); + + source.subscribe(Functions.emptyConsumer(), Functions.emptyConsumer()); + + System.gc(); + Thread.sleep(100); + + long after = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); + + source = null; + assertTrue(String.format("%,3d -> %,3d%n", start, after), start + 20 * 1000 * 1000 > after); + } + + @Test + public void publishNoLeak2() throws Exception { + System.gc(); + Thread.sleep(100); + + long start = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); + + source = Observable.fromCallable(new Callable() { + @Override + public Object call() throws Exception { + return new byte[100 * 1000 * 1000]; + } + }).concatWith(Observable.never()) + .publish() + .refCount(); + + Disposable d1 = source.test(); + Disposable d2 = source.test(); + + d1.dispose(); + d2.dispose(); + + d1 = null; + d2 = null; + + System.gc(); + Thread.sleep(100); + + long after = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); + + source = null; + assertTrue(String.format("%,3d -> %,3d%n", start, after), start + 20 * 1000 * 1000 > after); + } + + @Test + public void replayIsUnsubscribed() { + ConnectableObservable co = Observable.just(1).concatWith(Observable.never()) + .replay(); + + if (co instanceof Disposable) { + assertTrue(((Disposable)co).isDisposed()); + + Disposable connection = co.connect(); + + assertFalse(((Disposable)co).isDisposed()); + + connection.dispose(); + + assertTrue(((Disposable)co).isDisposed()); + } + } + + static final class BadObservableSubscribe extends ConnectableObservable { + + @Override + public void connect(Consumer connection) { + try { + connection.accept(Disposables.empty()); + } catch (Throwable ex) { + throw ExceptionHelper.wrapOrThrow(ex); + } + } + + @Override + protected void subscribeActual(Observer observer) { + throw new TestException("subscribeActual"); + } + } + + static final class BadObservableDispose extends ConnectableObservable implements Disposable { + + @Override + public void dispose() { + throw new TestException("dispose"); + } + + @Override + public boolean isDisposed() { + return false; + } + + @Override + public void connect(Consumer connection) { + try { + connection.accept(Disposables.empty()); + } catch (Throwable ex) { + throw ExceptionHelper.wrapOrThrow(ex); + } + } + + @Override + protected void subscribeActual(Observer observer) { + observer.onSubscribe(Disposables.empty()); + } + } + + static final class BadObservableConnect extends ConnectableObservable { + + @Override + public void connect(Consumer connection) { + throw new TestException("connect"); + } + + @Override + protected void subscribeActual(Observer observer) { + observer.onSubscribe(Disposables.empty()); + } + } + + @Test + public void badSourceSubscribe() { + BadObservableSubscribe bo = new BadObservableSubscribe(); + + try { + bo.refCount() + .test(); + fail("Should have thrown"); + } catch (NullPointerException ex) { + assertTrue(ex.getCause() instanceof TestException); + } + } + + @Test + public void badSourceDispose() { + BadObservableDispose bo = new BadObservableDispose(); + + try { + bo.refCount() + .test() + .cancel(); + fail("Should have thrown"); + } catch (TestException expected) { + } + } + + @Test + public void badSourceConnect() { + BadObservableConnect bo = new BadObservableConnect(); + + try { + bo.refCount() + .test(); + fail("Should have thrown"); + } catch (NullPointerException ex) { + assertTrue(ex.getCause() instanceof TestException); + } + } + + static final class BadObservableSubscribe2 extends ConnectableObservable { + + int count; + + @Override + public void connect(Consumer connection) { + try { + connection.accept(Disposables.empty()); + } catch (Throwable ex) { + throw ExceptionHelper.wrapOrThrow(ex); + } + } + + @Override + protected void subscribeActual(Observer observer) { + if (++count == 1) { + observer.onSubscribe(Disposables.empty()); + } else { + throw new TestException("subscribeActual"); + } + } + } + + @Test + public void badSourceSubscribe2() { + BadObservableSubscribe2 bo = new BadObservableSubscribe2(); + + Observable o = bo.refCount(); + o.test(); + try { + o.test(); + fail("Should have thrown"); + } catch (NullPointerException ex) { + assertTrue(ex.getCause() instanceof TestException); + } + } + + static final class BadObservableConnect2 extends ConnectableObservable + implements Disposable { + + @Override + public void connect(Consumer connection) { + try { + connection.accept(Disposables.empty()); + } catch (Throwable ex) { + throw ExceptionHelper.wrapOrThrow(ex); + } + } + + @Override + protected void subscribeActual(Observer observer) { + observer.onSubscribe(Disposables.empty()); + observer.onComplete(); + } + + @Override + public void dispose() { + throw new TestException("dispose"); + } + + @Override + public boolean isDisposed() { + return false; + } + } + + @Test + public void badSourceCompleteDisconnect() { + BadObservableConnect2 bo = new BadObservableConnect2(); + + try { + bo.refCount() + .test(); + fail("Should have thrown"); + } catch (NullPointerException ex) { + assertTrue(ex.getCause() instanceof TestException); + } + } + + @Test(timeout = 7500) + public void blockingSourceAsnycCancel() throws Exception { + BehaviorSubject bs = BehaviorSubject.createDefault(1); + + Observable o = bs + .replay(1) + .refCount(); + + o.subscribe(); + + final AtomicBoolean interrupted = new AtomicBoolean(); + + o.switchMap(new Function>() { + @Override + public ObservableSource apply(Integer v) throws Exception { + return Observable.create(new ObservableOnSubscribe() { + @Override + public void subscribe(ObservableEmitter emitter) throws Exception { + while (!emitter.isDisposed()) { + Thread.sleep(100); + } + interrupted.set(true); + } + }); + } + }) + .take(500, TimeUnit.MILLISECONDS) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + + assertTrue(interrupted.get()); + } + + @Test + public void byCount() { + final int[] subscriptions = { 0 }; + + Observable source = Observable.range(1, 5) + .doOnSubscribe(new Consumer() { + @Override + public void accept(Disposable d) throws Exception { + subscriptions[0]++; + } + }) + .publish() + .refCount(2); + + for (int i = 0; i < 3; i++) { + TestObserver to1 = source.test(); + + to1.withTag("to1 " + i); + to1.assertEmpty(); + + TestObserver to2 = source.test(); + + to2.withTag("to2 " + i); + + to1.assertResult(1, 2, 3, 4, 5); + to2.assertResult(1, 2, 3, 4, 5); + } + + assertEquals(3, subscriptions[0]); + } + + @Test + public void resubscribeBeforeTimeout() throws Exception { + final int[] subscriptions = { 0 }; + + PublishSubject ps = PublishSubject.create(); + + Observable source = ps + .doOnSubscribe(new Consumer() { + @Override + public void accept(Disposable d) throws Exception { + subscriptions[0]++; + } + }) + .publish() + .refCount(500, TimeUnit.MILLISECONDS); + + TestObserver to1 = source.test(); + + assertEquals(1, subscriptions[0]); + + to1.cancel(); + + Thread.sleep(100); + + to1 = source.test(); + + assertEquals(1, subscriptions[0]); + + Thread.sleep(500); + + assertEquals(1, subscriptions[0]); + + ps.onNext(1); + ps.onNext(2); + ps.onNext(3); + ps.onNext(4); + ps.onNext(5); + ps.onComplete(); + + to1 + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void letitTimeout() throws Exception { + final int[] subscriptions = { 0 }; + + PublishSubject ps = PublishSubject.create(); + + Observable source = ps + .doOnSubscribe(new Consumer() { + @Override + public void accept(Disposable d) throws Exception { + subscriptions[0]++; + } + }) + .publish() + .refCount(1, 100, TimeUnit.MILLISECONDS); + + TestObserver to1 = source.test(); + + assertEquals(1, subscriptions[0]); + + to1.cancel(); + + assertTrue(ps.hasObservers()); + + Thread.sleep(200); + + assertFalse(ps.hasObservers()); + } + + @Test + public void error() { + Observable.error(new IOException()) + .publish() + .refCount(500, TimeUnit.MILLISECONDS) + .test() + .assertFailure(IOException.class); + } + + @Test + public void comeAndGo() { + PublishSubject ps = PublishSubject.create(); + + Observable source = ps + .publish() + .refCount(1); + + TestObserver to1 = source.test(); + + assertTrue(ps.hasObservers()); + + for (int i = 0; i < 3; i++) { + TestObserver to2 = source.test(); + to1.cancel(); + to1 = to2; + } + + to1.cancel(); + + assertFalse(ps.hasObservers()); + } + + @Test + public void unsubscribeSubscribeRace() { + for (int i = 0; i < 1000; i++) { + + final Observable source = Observable.range(1, 5) + .replay() + .refCount(1) + ; + + final TestObserver to1 = source.test(); + + final TestObserver to2 = new TestObserver(); + + Runnable r1 = new Runnable() { + @Override + public void run() { + to1.cancel(); + } + }; + + Runnable r2 = new Runnable() { + @Override + public void run() { + source.subscribe(to2); + } + }; + + TestHelper.race(r1, r2, Schedulers.single()); + + to2 + .withTag("Round: " + i) + .assertResult(1, 2, 3, 4, 5); + } + } + + static final class BadObservableDoubleOnX extends ConnectableObservable + implements Disposable { + + @Override + public void connect(Consumer connection) { + try { + connection.accept(Disposables.empty()); + } catch (Throwable ex) { + throw ExceptionHelper.wrapOrThrow(ex); + } + } + + @Override + protected void subscribeActual(Observer observer) { + observer.onSubscribe(Disposables.empty()); + observer.onSubscribe(Disposables.empty()); + observer.onComplete(); + observer.onComplete(); + observer.onError(new TestException()); + } + + @Override + public void dispose() { + } + + @Override + public boolean isDisposed() { + return false; + } + } + + @Test + public void doubleOnX() { + List errors = TestHelper.trackPluginErrors(); + try { + new BadObservableDoubleOnX() + .refCount() + .test() + .assertResult(); + + TestHelper.assertError(errors, 0, ProtocolViolationException.class); + TestHelper.assertUndeliverable(errors, 1, TestException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void doubleOnXCount() { + List errors = TestHelper.trackPluginErrors(); + try { + new BadObservableDoubleOnX() + .refCount(1) + .test() + .assertResult(); + + TestHelper.assertError(errors, 0, ProtocolViolationException.class); + TestHelper.assertUndeliverable(errors, 1, TestException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void doubleOnXTime() { + List errors = TestHelper.trackPluginErrors(); + try { + new BadObservableDoubleOnX() + .refCount(5, TimeUnit.SECONDS, Schedulers.single()) + .test() + .assertResult(); + + TestHelper.assertError(errors, 0, ProtocolViolationException.class); + TestHelper.assertUndeliverable(errors, 1, TestException.class); + } finally { + RxJavaPlugins.reset(); + } + } + + @Test + public void cancelTerminateStateExclusion() { + ObservableRefCount o = (ObservableRefCount)PublishSubject.create() + .publish() + .refCount(); + + o.cancel(null); + + o.cancel(new RefConnection(o)); + + RefConnection rc = new RefConnection(o); + o.connection = null; + rc.subscriberCount = 0; + o.timeout(rc); + + rc.subscriberCount = 1; + o.timeout(rc); + + o.connection = rc; + o.timeout(rc); + + rc.subscriberCount = 0; + o.timeout(rc); + + // ------------------- + + rc.subscriberCount = 2; + rc.connected = false; + o.connection = rc; + o.cancel(rc); + + rc.subscriberCount = 1; + rc.connected = false; + o.connection = rc; + o.cancel(rc); + + rc.subscriberCount = 2; + rc.connected = true; + o.connection = rc; + o.cancel(rc); + + rc.subscriberCount = 1; + rc.connected = true; + o.connection = rc; + rc.lazySet(null); + o.cancel(rc); + + o.connection = rc; + o.cancel(new RefConnection(o)); + } + + @Test + public void replayRefCountShallBeThreadSafe() { + for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) { + Observable observable = Observable.just(1).replay(1).refCount(); + + TestObserver observer1 = observable + .subscribeOn(Schedulers.io()) + .test(); + + TestObserver observer2 = observable + .subscribeOn(Schedulers.io()) + .test(); + + observer1 + .withTag("" + i) + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1); + + observer2 + .withTag("" + i) + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1); + } + } + + static final class TestConnectableObservable extends ConnectableObservable + implements Disposable { + + volatile boolean disposed; + + @Override + public void dispose() { + disposed = true; + } + + @Override + public boolean isDisposed() { + return disposed; + } + + @Override + public void connect(Consumer connection) { + // not relevant + } + + @Override + protected void subscribeActual(Observer observer) { + // not relevant + } + } + + @Test + public void timeoutDisposesSource() { + ObservableRefCount o = (ObservableRefCount)new TestConnectableObservable().refCount(); + + RefConnection rc = new RefConnection(o); + o.connection = rc; + + o.timeout(rc); + + assertTrue(((Disposable)o.source).isDisposed()); + } + + @Test + public void disconnectBeforeConnect() { + BehaviorSubject subject = BehaviorSubject.create(); + + Observable observable = subject + .replay(1) + .refCount(); + + observable.takeUntil(Observable.just(1)).test(); + + subject.onNext(2); + + observable.take(1).test().assertResult(2); + } + + @Test + public void publishRefCountShallBeThreadSafe() { + for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) { + Observable observable = Observable.just(1).publish().refCount(); + + TestObserver observer1 = observable + .subscribeOn(Schedulers.io()) + .test(); + + TestObserver observer2 = observable + .subscribeOn(Schedulers.io()) + .test(); + + observer1 + .withTag("observer1 " + i) + .awaitDone(5, TimeUnit.SECONDS) + .assertNoErrors() + .assertComplete(); + + observer2 + .withTag("observer2 " + i) + .awaitDone(5, TimeUnit.SECONDS) + .assertNoErrors() + .assertComplete(); + } + } +} diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountTest.java index 0f0d930d8d..48a9ff2d5a 100644 --- a/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountTest.java +++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountTest.java @@ -23,7 +23,7 @@ import java.util.concurrent.*; import java.util.concurrent.atomic.*; -import org.junit.Test; +import org.junit.*; import org.mockito.InOrder; import io.reactivex.*; @@ -43,6 +43,27 @@ public class ObservableRefCountTest { + // This will undo the workaround so that the plain ObservablePublish is still + // tested. + @Before + @SuppressWarnings({ "rawtypes", "unchecked" }) + public void before() { + RxJavaPlugins.setOnConnectableObservableAssembly(new Function() { + @Override + public ConnectableObservable apply(ConnectableObservable co) throws Exception { + if (co instanceof ObservablePublishAlt) { + return ObservablePublish.create(((ObservablePublishAlt)co).source()); + } + return co; + } + }); + } + + @After + public void after() { + RxJavaPlugins.setOnConnectableObservableAssembly(null); + } + @Test public void testRefCountAsync() { final AtomicInteger subscribeCount = new AtomicInteger();