### Question:
NettyChannelPublisher extends SubscribablePublisher<T> { private void cancel(SubscriptionImpl forSubscription) { if (forSubscription != subscription) { return; } resetSubscription(); pending = null; if (fatalError == null) { fatalError = StacklessClosedChannelException.newInstance(NettyChannelPublisher.class, "cancel"); } closeChannelInbound(); } NettyChannelPublisher(Channel channel, Predicate<T> terminalSignalPredicate, CloseHandler closeHandler);   }### Answer:
@Test public void testCancelBeforeTerminal() { toSource(publisher).subscribe(subscriber); subscriber.request(3); fireChannelRead(1); assertThat(subscriber.takeItems(), contains(1)); subscriber.cancel(); assertThat("Channel not closed post cancel.", channel.closeFuture().isDone(), is(true)); }
@Test public void testCancelAfterTerminal() { toSource(publisher).subscribe(subscriber); subscriber.request(3); nextItemTerminal = true; fireChannelRead(1); assertThat(subscriber.takeItems(), contains(1)); assertThat(subscriber.takeTerminal(), is(complete())); subscriber.cancel(); assertThat("Channel closed on cancel post terminate.", channel.closeFuture().isDone(), is(false)); }
@Test public void testDelayedCancel() { nextItemTerminal = true; toSource(publisher).subscribe(subscriber); subscriber.request(3); final Subscription firstSubscription = requireNonNull(subscriber.subscription()); fireChannelRead(1); assertThat(subscriber.takeItems(), contains(1)); assertThat(subscriber.takeTerminal(), is(complete())); nextItemTerminal = false; fireChannelReadToBuffer(2); toSource(publisher).subscribe(subscriber2); subscriber2.request(3); assertThat(subscriber2.takeItems(), contains(2)); firstSubscription.cancel(); nextItemTerminal = true; fireChannelRead(3); assertThat(subscriber2.takeItems(), contains(3)); }