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