@Test public void closeAsyncGracefullyWithoutNettyConnectionChannelHandler() throws InterruptedException { when(mockClosableAttribute.getAndSet(any())).thenReturn(null); Completable completable = closeAsyncGracefully(fixture, 100, SECONDS); verify(channel, never()).close(); TestCollectingCompletableSubscriber subscriber = new TestCollectingCompletableSubscriber(); toSource(completable).subscribe(subscriber); verify(channel).close(); subscriber.awaitOnComplete(); }
@Override public Completable closeAsyncGracefully() { return new SubscribableCompletable() { @Override protected void handleSubscribe(final Subscriber subscriber) { if (!stateUpdater.compareAndSet(ChannelSet.this, OPEN, GRACEFULLY_CLOSING)) { toSource(onClose).subscribe(subscriber); return; } if (channelMap.isEmpty()) { toSource(onClose).subscribe(subscriber); onCloseProcessor.onComplete(); return; } CompositeCloseable closeable = newCompositeCloseable().appendAll(() -> onClose); for (final Channel channel : channelMap.values()) { Attribute<AsyncCloseable> closeableAttribute = channel.attr(CHANNEL_CLOSEABLE_KEY); AsyncCloseable channelCloseable = closeableAttribute.getAndSet(null); if (channelCloseable != null) { closeable.merge(new AsyncCloseable() { @Override public Completable closeAsync() { return channelCloseable.closeAsync().publishOnOverride(immediate()); } @Override public Completable closeAsyncGracefully() { return channelCloseable.closeAsyncGracefully().publishOnOverride(immediate()); } }); } else { channel.close(); } } toSource(closeable.closeAsyncGracefully()).subscribe(subscriber); } }; }
ChannelSet implements ListenableAsyncCloseable { @Override public Completable closeAsyncGracefully() { return new SubscribableCompletable() { @Override protected void handleSubscribe(final Subscriber subscriber) { if (!stateUpdater.compareAndSet(ChannelSet.this, OPEN, GRACEFULLY_CLOSING)) { toSource(onClose).subscribe(subscriber); return; } if (channelMap.isEmpty()) { toSource(onClose).subscribe(subscriber); onCloseProcessor.onComplete(); return; } CompositeCloseable closeable = newCompositeCloseable().appendAll(() -> onClose); for (final Channel channel : channelMap.values()) { Attribute<AsyncCloseable> closeableAttribute = channel.attr(CHANNEL_CLOSEABLE_KEY); AsyncCloseable channelCloseable = closeableAttribute.getAndSet(null); if (channelCloseable != null) { closeable.merge(new AsyncCloseable() { @Override public Completable closeAsync() { return channelCloseable.closeAsync().publishOnOverride(immediate()); } @Override public Completable closeAsyncGracefully() { return channelCloseable.closeAsyncGracefully().publishOnOverride(immediate()); } }); } else { channel.close(); } } toSource(closeable.closeAsyncGracefully()).subscribe(subscriber); } }; } }
ChannelSet implements ListenableAsyncCloseable { @Override public Completable closeAsyncGracefully() { return new SubscribableCompletable() { @Override protected void handleSubscribe(final Subscriber subscriber) { if (!stateUpdater.compareAndSet(ChannelSet.this, OPEN, GRACEFULLY_CLOSING)) { toSource(onClose).subscribe(subscriber); return; } if (channelMap.isEmpty()) { toSource(onClose).subscribe(subscriber); onCloseProcessor.onComplete(); return; } CompositeCloseable closeable = newCompositeCloseable().appendAll(() -> onClose); for (final Channel channel : channelMap.values()) { Attribute<AsyncCloseable> closeableAttribute = channel.attr(CHANNEL_CLOSEABLE_KEY); AsyncCloseable channelCloseable = closeableAttribute.getAndSet(null); if (channelCloseable != null) { closeable.merge(new AsyncCloseable() { @Override public Completable closeAsync() { return channelCloseable.closeAsync().publishOnOverride(immediate()); } @Override public Completable closeAsyncGracefully() { return channelCloseable.closeAsyncGracefully().publishOnOverride(immediate()); } }); } else { channel.close(); } } toSource(closeable.closeAsyncGracefully()).subscribe(subscriber); } }; } ChannelSet(Executor offloadingExecutor); }
ChannelSet implements ListenableAsyncCloseable { @Override public Completable closeAsyncGracefully() { return new SubscribableCompletable() { @Override protected void handleSubscribe(final Subscriber subscriber) { if (!stateUpdater.compareAndSet(ChannelSet.this, OPEN, GRACEFULLY_CLOSING)) { toSource(onClose).subscribe(subscriber); return; } if (channelMap.isEmpty()) { toSource(onClose).subscribe(subscriber); onCloseProcessor.onComplete(); return; } CompositeCloseable closeable = newCompositeCloseable().appendAll(() -> onClose); for (final Channel channel : channelMap.values()) { Attribute<AsyncCloseable> closeableAttribute = channel.attr(CHANNEL_CLOSEABLE_KEY); AsyncCloseable channelCloseable = closeableAttribute.getAndSet(null); if (channelCloseable != null) { closeable.merge(new AsyncCloseable() { @Override public Completable closeAsync() { return channelCloseable.closeAsync().publishOnOverride(immediate()); } @Override public Completable closeAsyncGracefully() { return channelCloseable.closeAsyncGracefully().publishOnOverride(immediate()); } }); } else { channel.close(); } } toSource(closeable.closeAsyncGracefully()).subscribe(subscriber); } }; } ChannelSet(Executor offloadingExecutor); boolean addIfAbsent(final Channel channel); @Override Completable closeAsync(); @Override Completable closeAsyncGracefully(); @Override Completable onClose(); }
ChannelSet implements ListenableAsyncCloseable { @Override public Completable closeAsyncGracefully() { return new SubscribableCompletable() { @Override protected void handleSubscribe(final Subscriber subscriber) { if (!stateUpdater.compareAndSet(ChannelSet.this, OPEN, GRACEFULLY_CLOSING)) { toSource(onClose).subscribe(subscriber); return; } if (channelMap.isEmpty()) { toSource(onClose).subscribe(subscriber); onCloseProcessor.onComplete(); return; } CompositeCloseable closeable = newCompositeCloseable().appendAll(() -> onClose); for (final Channel channel : channelMap.values()) { Attribute<AsyncCloseable> closeableAttribute = channel.attr(CHANNEL_CLOSEABLE_KEY); AsyncCloseable channelCloseable = closeableAttribute.getAndSet(null); if (channelCloseable != null) { closeable.merge(new AsyncCloseable() { @Override public Completable closeAsync() { return channelCloseable.closeAsync().publishOnOverride(immediate()); } @Override public Completable closeAsyncGracefully() { return channelCloseable.closeAsyncGracefully().publishOnOverride(immediate()); } }); } else { channel.close(); } } toSource(closeable.closeAsyncGracefully()).subscribe(subscriber); } }; } ChannelSet(Executor offloadingExecutor); boolean addIfAbsent(final Channel channel); @Override Completable closeAsync(); @Override Completable closeAsyncGracefully(); @Override Completable onClose(); static final AttributeKey<AsyncCloseable> CHANNEL_CLOSEABLE_KEY; }