### Question:
DefaultChannelOperations implements ChannelOperations<W> { @Override public Observable<Void> writeAndFlushOnEach(Observable<W> msgs) { return _write(msgs, flushOnEachSelector); } 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();  }### Answer:
@Test(timeout = 60000) public void testWriteAndFlushOnEach() throws Exception { final String msg1 = "Hello1"; final String msg2 = "Hello2"; Observable<Void> writeO = channelOpRule.channelOperations.writeAndFlushOnEach(ChannelOpRule.bbJust(msg1, msg2)); _testWithFlushSelector(writeO, msg1, msg2); }