OpenWebNetObservable { static Func1<OpenContext, Observable<OpenContext>> write(String value) { return context -> { try { byte[] message = new String(value).getBytes(); ByteBuffer buffer = ByteBuffer.wrap(message); Integer count = context.getClient().write(buffer); log(String.format("write: %d|%s", count, value)); return Observable.just(context); } catch (IOException e) { return Observable.error(e); } }; } private  OpenWebNetObservable();   }
@Test public void testWrite() throws IOException { String value = "myFrameValue"; ByteBuffer buffer = ByteBuffer.wrap(value.getBytes()); when(mockedSocketChannel.write(buffer)).then(invocation -> value.length()); TestSubscriber<OpenContext> tester = new TestSubscriber<>(); Observable.just(mockedContext) .flatMap(OpenWebNetObservable.write(value)) .subscribe(tester); verify(mockedSocketChannel, times(1)).write(buffer); tester.assertValue(mockedContext); tester.assertCompleted(); }