@Test(timeout = 60000) public void testCloseListener() throws Exception { Observable<Void> closeListener = channelOpRule.channelOperations.closeListener(); TestSubscriber<Void> subscriber = new TestSubscriber<>(); closeListener.subscribe(subscriber); subscriber.assertNoTerminalEvent(); subscriber.unsubscribe(); subscriber.assertNoTerminalEvent(); channelOpRule.channel.close().sync(); subscriber.assertNoTerminalEvent(); }
@Override public Observable<Void> closeListener() { return Observable.create(new OnSubscribe<Void>() { @Override public void call(final Subscriber<? super Void> subscriber) { final SubscriberToChannelFutureBridge l = new SubscriberToChannelFutureBridge() { @Override protected void doOnSuccess(ChannelFuture future) { subscriber.onCompleted(); } @Override protected void doOnFailure(ChannelFuture future, Throwable cause) { subscriber.onCompleted(); } }; l.bridge(nettyChannel.closeFuture(), subscriber); } }); }
DefaultChannelOperations implements ChannelOperations<W> { @Override public Observable<Void> closeListener() { return Observable.create(new OnSubscribe<Void>() { @Override public void call(final Subscriber<? super Void> subscriber) { final SubscriberToChannelFutureBridge l = new SubscriberToChannelFutureBridge() { @Override protected void doOnSuccess(ChannelFuture future) { subscriber.onCompleted(); } @Override protected void doOnFailure(ChannelFuture future, Throwable cause) { subscriber.onCompleted(); } }; l.bridge(nettyChannel.closeFuture(), subscriber); } }); } }
DefaultChannelOperations implements ChannelOperations<W> { @Override public Observable<Void> closeListener() { return Observable.create(new OnSubscribe<Void>() { @Override public void call(final Subscriber<? super Void> subscriber) { final SubscriberToChannelFutureBridge l = new SubscriberToChannelFutureBridge() { @Override protected void doOnSuccess(ChannelFuture future) { subscriber.onCompleted(); } @Override protected void doOnFailure(ChannelFuture future, Throwable cause) { subscriber.onCompleted(); } }; l.bridge(nettyChannel.closeFuture(), subscriber); } }); } DefaultChannelOperations(final Channel nettyChannel, ConnectionEventListener eventListener,
                                    EventPublisher eventPublisher); }
DefaultChannelOperations implements ChannelOperations<W> { @Override public Observable<Void> closeListener() { return Observable.create(new OnSubscribe<Void>() { @Override public void call(final Subscriber<? super Void> subscriber) { final SubscriberToChannelFutureBridge l = new SubscriberToChannelFutureBridge() { @Override protected void doOnSuccess(ChannelFuture future) { subscriber.onCompleted(); } @Override protected void doOnFailure(ChannelFuture future, Throwable cause) { subscriber.onCompleted(); } }; l.bridge(nettyChannel.closeFuture(), subscriber); } }); } DefaultChannelOperations(final Channel nettyChannel, ConnectionEventListener eventListener,
                                    EventPublisher eventPublisher); @Override Observable<Void> write(final Observable<W> msgs); @Override Observable<Void> write(Observable<W> msgs, final Func1<W, Boolean> flushSelector); @Override Observable<Void> writeAndFlushOnEach(Observable<W> msgs); @Override Observable<Void> writeString(Observable<String> msgs); @Override Observable<Void> writeString(Observable<String> msgs, Func1<String, Boolean> flushSelector); @Override Observable<Void> writeStringAndFlushOnEach(Observable<String> msgs); @Override Observable<Void> writeBytes(Observable<byte[]> msgs); @Override Observable<Void> writeBytes(Observable<byte[]> msgs, Func1<byte[], Boolean> flushSelector); @Override Observable<Void> writeBytesAndFlushOnEach(Observable<byte[]> msgs); @Override Observable<Void> writeFileRegion(Observable<FileRegion> msgs); @Override Observable<Void> writeFileRegion(Observable<FileRegion> msgs, Func1<FileRegion, Boolean> flushSelector); @Override Observable<Void> writeFileRegionAndFlushOnEach(Observable<FileRegion> msgs); @Override ChannelOperations<WW> transformWrite(AllocatingTransformer<WW, W> transformer); @Override void flush(); @Override Observable<Void> close(); @Override Observable<Void> close(boolean flush); @Override void closeNow(); @Override Observable<Void> closeListener(); }
DefaultChannelOperations implements ChannelOperations<W> { @Override public Observable<Void> closeListener() { return Observable.create(new OnSubscribe<Void>() { @Override public void call(final Subscriber<? super Void> subscriber) { final SubscriberToChannelFutureBridge l = new SubscriberToChannelFutureBridge() { @Override protected void doOnSuccess(ChannelFuture future) { subscriber.onCompleted(); } @Override protected void doOnFailure(ChannelFuture future, Throwable cause) { subscriber.onCompleted(); } }; l.bridge(nettyChannel.closeFuture(), subscriber); } }); } DefaultChannelOperations(final Channel nettyChannel, ConnectionEventListener eventListener,
                                    EventPublisher eventPublisher); @Override Observable<Void> write(final Observable<W> msgs); @Override Observable<Void> write(Observable<W> msgs, final Func1<W, Boolean> flushSelector); @Override Observable<Void> writeAndFlushOnEach(Observable<W> msgs); @Override Observable<Void> writeString(Observable<String> msgs); @Override Observable<Void> writeString(Observable<String> msgs, Func1<String, Boolean> flushSelector); @Override Observable<Void> writeStringAndFlushOnEach(Observable<String> msgs); @Override Observable<Void> writeBytes(Observable<byte[]> msgs); @Override Observable<Void> writeBytes(Observable<byte[]> msgs, Func1<byte[], Boolean> flushSelector); @Override Observable<Void> writeBytesAndFlushOnEach(Observable<byte[]> msgs); @Override Observable<Void> writeFileRegion(Observable<FileRegion> msgs); @Override Observable<Void> writeFileRegion(Observable<FileRegion> msgs, Func1<FileRegion, Boolean> flushSelector); @Override Observable<Void> writeFileRegionAndFlushOnEach(Observable<FileRegion> msgs); @Override ChannelOperations<WW> transformWrite(AllocatingTransformer<WW, W> transformer); @Override void flush(); @Override Observable<Void> close(); @Override Observable<Void> close(boolean flush); @Override void closeNow(); @Override Observable<Void> closeListener();  }