### Question:
MultiReferenceCountSubscriber extends MultiOperatorProcessor<T, T> { @Override public void cancel() { super.cancel(); if (done.compareAndSet(false, true)) { parent.cancel(connection); } } MultiReferenceCountSubscriber(MultiSubscriber<? super T> downstream, MultiReferenceCount<T> parent,
            ConnectableMultiConnection connection); @Override void onItem(T t); @Override void onFailure(Throwable failure); @Override void onCompletion(); @Override void cancel();  }### Answer:
@Test public void testCancellationWithASingleSubscriber() { Multi<Integer> multi = Multi.createFrom().items(1, 2, 3, 4, 5) .broadcast().withCancellationAfterLastSubscriberDeparture().toAllSubscribers(); AssertSubscriber<Integer> subscriber = multi.subscribe() .withSubscriber(AssertSubscriber.create(1)); subscriber.request(2).cancel(); subscriber.assertReceived(1, 2, 3); }
@Test public void testCancellationWithATwoSubscribers() { Multi<Integer> multi = Multi.createFrom().items(1, 2, 3, 4, 5) .broadcast().withCancellationAfterLastSubscriberDeparture().toAllSubscribers(); AssertSubscriber<Integer> subscriber1 = multi.subscribe() .withSubscriber(AssertSubscriber.create(0)); AssertSubscriber<Integer> subscriber2 = multi.subscribe() .withSubscriber(AssertSubscriber.create(0)); subscriber2.request(3); subscriber1.request(3).cancel(); subscriber1.assertReceived(1, 2, 3); subscriber2.request(20); subscriber2.assertReceived(1, 2, 3, 4, 5); }