@Test public void offloadedCompletableSubscriberSignalOverlap() { CompletableSource.Subscriber offloaded = offloader.offloadSubscriber(completableSub); verifyNoMoreInteractions(executor.mock); verifyNoMoreInteractions(completableSub); offloaded.onSubscribe(cancellable); verify(executor.mock).execute(any()); verifyNoMoreInteractions(completableSub); offloaded.onComplete(); verifyNoMoreInteractions(executor.mock); assertThat("Unexpected tasks executed.", executor.executeAllTasks(), is(1)); verify(completableSub).onSubscribe(cancellable); verify(completableSub).onComplete(); verifyNoMoreInteractions(completableSub); verifyNoMoreInteractions(executor.mock); }
@Override public <T> Subscriber<? super T> offloadSubscriber(final Subscriber<? super T> subscriber) { return new OffloadedSubscriber<>(subscriber, executor, publisherSignalQueueInitialCapacity); }
TaskBasedSignalOffloader implements SignalOffloader { @Override public <T> Subscriber<? super T> offloadSubscriber(final Subscriber<? super T> subscriber) { return new OffloadedSubscriber<>(subscriber, executor, publisherSignalQueueInitialCapacity); } }
TaskBasedSignalOffloader implements SignalOffloader { @Override public <T> Subscriber<? super T> offloadSubscriber(final Subscriber<? super T> subscriber) { return new OffloadedSubscriber<>(subscriber, executor, publisherSignalQueueInitialCapacity); } TaskBasedSignalOffloader(final Executor executor);   TaskBasedSignalOffloader(final Executor executor, final int publisherSignalQueueInitialCapacity); }
TaskBasedSignalOffloader implements SignalOffloader { @Override public <T> Subscriber<? super T> offloadSubscriber(final Subscriber<? super T> subscriber) { return new OffloadedSubscriber<>(subscriber, executor, publisherSignalQueueInitialCapacity); } TaskBasedSignalOffloader(final Executor executor);   TaskBasedSignalOffloader(final Executor executor, final int publisherSignalQueueInitialCapacity); @Override Subscriber<? super T> offloadSubscriber(final Subscriber<? super T> subscriber); @Override SingleSource.Subscriber<? super T> offloadSubscriber(
            final SingleSource.Subscriber<? super T> subscriber); @Override CompletableSource.Subscriber offloadSubscriber(final CompletableSource.Subscriber subscriber); @Override Subscriber<? super T> offloadSubscription(final Subscriber<? super T> subscriber); @Override SingleSource.Subscriber<? super T> offloadCancellable(
            final SingleSource.Subscriber<? super T> subscriber); @Override CompletableSource.Subscriber offloadCancellable(final CompletableSource.Subscriber subscriber); @Override void offloadSubscribe(final Subscriber<? super T> subscriber,
                                     final Consumer<Subscriber<? super T>> handleSubscribe); @Override void offloadSubscribe(final SingleSource.Subscriber<? super T> subscriber,
                                     final Consumer<SingleSource.Subscriber<? super T>> handleSubscribe); @Override void offloadSubscribe(final CompletableSource.Subscriber subscriber,
                                 final Consumer<CompletableSource.Subscriber> handleSubscribe); @Override void offloadSignal(final T signal, final Consumer<T> signalConsumer); }
TaskBasedSignalOffloader implements SignalOffloader { @Override public <T> Subscriber<? super T> offloadSubscriber(final Subscriber<? super T> subscriber) { return new OffloadedSubscriber<>(subscriber, executor, publisherSignalQueueInitialCapacity); } TaskBasedSignalOffloader(final Executor executor);   TaskBasedSignalOffloader(final Executor executor, final int publisherSignalQueueInitialCapacity); @Override Subscriber<? super T> offloadSubscriber(final Subscriber<? super T> subscriber); @Override SingleSource.Subscriber<? super T> offloadSubscriber(
            final SingleSource.Subscriber<? super T> subscriber); @Override CompletableSource.Subscriber offloadSubscriber(final CompletableSource.Subscriber subscriber); @Override Subscriber<? super T> offloadSubscription(final Subscriber<? super T> subscriber); @Override SingleSource.Subscriber<? super T> offloadCancellable(
            final SingleSource.Subscriber<? super T> subscriber); @Override CompletableSource.Subscriber offloadCancellable(final CompletableSource.Subscriber subscriber); @Override void offloadSubscribe(final Subscriber<? super T> subscriber,
                                     final Consumer<Subscriber<? super T>> handleSubscribe); @Override void offloadSubscribe(final SingleSource.Subscriber<? super T> subscriber,
                                     final Consumer<SingleSource.Subscriber<? super T>> handleSubscribe); @Override void offloadSubscribe(final CompletableSource.Subscriber subscriber,
                                 final Consumer<CompletableSource.Subscriber> handleSubscribe); @Override void offloadSignal(final T signal, final Consumer<T> signalConsumer);  }