@Test public void onSubscribeOnSubscribeRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final FlowableProcessor<Integer> s = PublishProcessor.<Integer>create().toSerialized(); TestSubscriber<Integer> ts = s.test(); final BooleanSubscription bs1 = new BooleanSubscription(); final BooleanSubscription bs2 = new BooleanSubscription(); Runnable r1 = new Runnable() { @Override public void run() { s.onSubscribe(bs1); } }; Runnable r2 = new Runnable() { @Override public void run() { s.onSubscribe(bs2); } }; TestHelper.race(r1, r2); ts.assertEmpty(); } }
@Override public void onSubscribe(Subscription s) { boolean cancel; if (!done) { synchronized (this) { if (done) { cancel = true; } else { if (emitting) { AppendOnlyLinkedArrayList<Object> q = queue; if (q == null) { q = new AppendOnlyLinkedArrayList<Object>(4); queue = q; } q.add(NotificationLite.subscription(s)); return; } emitting = true; cancel = false; } } } else { cancel = true; } if (cancel) { s.cancel(); } else { actual.onSubscribe(s); emitLoop(); } }
SerializedProcessor extends FlowableProcessor<T> { @Override public void onSubscribe(Subscription s) { boolean cancel; if (!done) { synchronized (this) { if (done) { cancel = true; } else { if (emitting) { AppendOnlyLinkedArrayList<Object> q = queue; if (q == null) { q = new AppendOnlyLinkedArrayList<Object>(4); queue = q; } q.add(NotificationLite.subscription(s)); return; } emitting = true; cancel = false; } } } else { cancel = true; } if (cancel) { s.cancel(); } else { actual.onSubscribe(s); emitLoop(); } } }
SerializedProcessor extends FlowableProcessor<T> { @Override public void onSubscribe(Subscription s) { boolean cancel; if (!done) { synchronized (this) { if (done) { cancel = true; } else { if (emitting) { AppendOnlyLinkedArrayList<Object> q = queue; if (q == null) { q = new AppendOnlyLinkedArrayList<Object>(4); queue = q; } q.add(NotificationLite.subscription(s)); return; } emitting = true; cancel = false; } } } else { cancel = true; } if (cancel) { s.cancel(); } else { actual.onSubscribe(s); emitLoop(); } } SerializedProcessor(final FlowableProcessor<T> actual); }
SerializedProcessor extends FlowableProcessor<T> { @Override public void onSubscribe(Subscription s) { boolean cancel; if (!done) { synchronized (this) { if (done) { cancel = true; } else { if (emitting) { AppendOnlyLinkedArrayList<Object> q = queue; if (q == null) { q = new AppendOnlyLinkedArrayList<Object>(4); queue = q; } q.add(NotificationLite.subscription(s)); return; } emitting = true; cancel = false; } } } else { cancel = true; } if (cancel) { s.cancel(); } else { actual.onSubscribe(s); emitLoop(); } } SerializedProcessor(final FlowableProcessor<T> actual); @Override void onSubscribe(Subscription s); @Override void onNext(T t); @Override void onError(Throwable t); @Override void onComplete(); @Override boolean hasSubscribers(); @Override boolean hasThrowable(); @Override @Nullable Throwable getThrowable(); @Override boolean hasComplete(); }
SerializedProcessor extends FlowableProcessor<T> { @Override public void onSubscribe(Subscription s) { boolean cancel; if (!done) { synchronized (this) { if (done) { cancel = true; } else { if (emitting) { AppendOnlyLinkedArrayList<Object> q = queue; if (q == null) { q = new AppendOnlyLinkedArrayList<Object>(4); queue = q; } q.add(NotificationLite.subscription(s)); return; } emitting = true; cancel = false; } } } else { cancel = true; } if (cancel) { s.cancel(); } else { actual.onSubscribe(s); emitLoop(); } } SerializedProcessor(final FlowableProcessor<T> actual); @Override void onSubscribe(Subscription s); @Override void onNext(T t); @Override void onError(Throwable t); @Override void onComplete(); @Override boolean hasSubscribers(); @Override boolean hasThrowable(); @Override @Nullable Throwable getThrowable(); @Override boolean hasComplete();  }