@Test(timeout = 60000) public void testFlush() throws Exception { String msg = "Hello"; channelOpRule.channel.write(Unpooled.buffer().writeBytes(msg.getBytes())); channelOpRule.channelOperations.flush(); channelOpRule.verifyOutboundMessages(msg); }
@Override public void flush() { if (eventPublisher.publishingEnabled()) { final long startTimeNanos = Clock.newStartTimeNanos(); eventListener.onFlushStart(); if (nettyChannel.eventLoop().inEventLoop()) { _flushInEventloop(startTimeNanos); } else { nettyChannel.eventLoop() .execute(new Runnable() { @Override public void run() { _flushInEventloop(startTimeNanos); } }); } } else { nettyChannel.flush(); } }
DefaultChannelOperations implements ChannelOperations<W> { @Override public void flush() { if (eventPublisher.publishingEnabled()) { final long startTimeNanos = Clock.newStartTimeNanos(); eventListener.onFlushStart(); if (nettyChannel.eventLoop().inEventLoop()) { _flushInEventloop(startTimeNanos); } else { nettyChannel.eventLoop() .execute(new Runnable() { @Override public void run() { _flushInEventloop(startTimeNanos); } }); } } else { nettyChannel.flush(); } } }
DefaultChannelOperations implements ChannelOperations<W> { @Override public void flush() { if (eventPublisher.publishingEnabled()) { final long startTimeNanos = Clock.newStartTimeNanos(); eventListener.onFlushStart(); if (nettyChannel.eventLoop().inEventLoop()) { _flushInEventloop(startTimeNanos); } else { nettyChannel.eventLoop() .execute(new Runnable() { @Override public void run() { _flushInEventloop(startTimeNanos); } }); } } else { nettyChannel.flush(); } } DefaultChannelOperations(final Channel nettyChannel, ConnectionEventListener eventListener,
                                    EventPublisher eventPublisher); }
DefaultChannelOperations implements ChannelOperations<W> { @Override public void flush() { if (eventPublisher.publishingEnabled()) { final long startTimeNanos = Clock.newStartTimeNanos(); eventListener.onFlushStart(); if (nettyChannel.eventLoop().inEventLoop()) { _flushInEventloop(startTimeNanos); } else { nettyChannel.eventLoop() .execute(new Runnable() { @Override public void run() { _flushInEventloop(startTimeNanos); } }); } } else { nettyChannel.flush(); } } 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 void flush() { if (eventPublisher.publishingEnabled()) { final long startTimeNanos = Clock.newStartTimeNanos(); eventListener.onFlushStart(); if (nettyChannel.eventLoop().inEventLoop()) { _flushInEventloop(startTimeNanos); } else { nettyChannel.eventLoop() .execute(new Runnable() { @Override public void run() { _flushInEventloop(startTimeNanos); } }); } } else { nettyChannel.flush(); } } 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();  }