@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)); }
private void cancel(SubscriptionImpl forSubscription) { if (forSubscription != subscription) { return; } resetSubscription(); pending = null; if (fatalError == null) { fatalError = StacklessClosedChannelException.newInstance(NettyChannelPublisher.class, "cancel"); } closeChannelInbound(); }
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 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); }
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);  }
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);   }