### Question:
TaskBasedSignalOffloader implements SignalOffloader { @Override public <T> Subscriber<? super T> offloadSubscription(final Subscriber<? super T> subscriber) { return new OffloadedSubscriptionSubscriber<>(subscriber, executor); } 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);  }### Answer:
@Test public void offloadSubscriptionRequestN() { Subscription offloaded = offloadSubscription(); requestNAndVerify(offloaded); }
@Test public void offloadSubscriptionRequestNReentrant() { Subscription offloaded = offloadSubscription(); doAnswer(invocation -> { pubSub.onNext(1); return null; }).when(subscription).request(1L); doAnswer(invocation -> { offloaded.request(2); return null; }).when(pubSub).onNext(1); offloaded.request(1); verify(executor.mock).execute(any()); assertThat("Unexpected tasks executed.", executor.executeAllTasks(), is(1)); verify(pubSub).onNext(1); verify(subscription).request(1); verify(subscription).request(2); verifyNoMoreInteractions(executor.mock); }
@Test public void offloadSubscriptionRequestNMany() { Subscription offloaded = offloadSubscription(); requestNAndVerify(offloaded); offloaded.request(1); verify(executor.mock, times(2)).execute(any()); assertThat("Unexpected tasks executed.", executor.executeAllTasks(), is(1)); verifyNoMoreInteractions(pubSub); verify(subscription, times(2)).request(1); verifyNoMoreInteractions(executor.mock); }
@Test public void offloadSubscriptionCancel() { Subscription offloaded = offloadSubscription(); cancelAndVerify(offloaded); }
@Test public void offloadSubscriptionRequestNThenCancel() { Subscription offloaded = offloadSubscription(); requestNAndVerify(offloaded); offloaded.cancel(); verify(executor.mock, times(2)).execute(any()); assertThat("Unexpected tasks executed.", executor.executeAllTasks(), is(1)); verifyNoMoreInteractions(pubSub); verify(subscription).cancel(); verifyNoMoreInteractions(executor.mock); }
@Test public void offloadSubscriptionInvalidRequestN() { Subscription offloaded = offloadSubscription(); offloaded.request(-4); verifyNoMoreInteractions(pubSub); verify(executor.mock).execute(any()); assertThat("Unexpected tasks executed.", executor.executeAllTasks(), is(1)); verifyNoMoreInteractions(pubSub); verify(subscription).request(-4); verifyNoMoreInteractions(executor.mock); }