@Test public void json() throws Exception { ByteBuf[] bufs = new ByteBuf[7]; byte[] data0 = "[{\"a\":1},{\"a\":2},{\"a\":3".getBytes(StandardCharsets.UTF_8); byte[] data1 = "},{\"a\":4},{\"a\":5}]".getBytes(StandardCharsets.UTF_8); bufs[0] = Unpooled.wrappedBuffer(data0); bufs[1] = Unpooled.wrappedBuffer(data1); Observable.just(bufs[0], bufs[1]) .compose(ByteBufs.json()).toBlocking().forEach(new Action1<ByteBuf>() { private int i = 0; @Override public void call(ByteBuf byteBuf) { bufs[2 + i] = byteBuf; String obj = byteBuf.toString(StandardCharsets.UTF_8); byteBuf.release(); Assert.assertEquals(String.format("{\"a\":%d}", ++i), obj); } }); bufs[0].release(); bufs[1].release(); for (ByteBuf buf : bufs) { Assert.assertEquals(0, buf.refCnt()); } }
public static Observable.Transformer<ByteBuf, ByteBuf> json() { return json(1024 * 1024); }
ByteBufs { public static Observable.Transformer<ByteBuf, ByteBuf> json() { return json(1024 * 1024); } }
ByteBufs { public static Observable.Transformer<ByteBuf, ByteBuf> json() { return json(1024 * 1024); } private  ByteBufs(); }
ByteBufs { public static Observable.Transformer<ByteBuf, ByteBuf> json() { return json(1024 * 1024); } private  ByteBufs(); static Observable.Transformer<ByteBuf, ByteBuf> gzip(); static Observable.Transformer<ByteBuf, ByteBuf> gunzip(); static Observable.Transformer<ByteBuf, ByteBuf> json(); static Observable.Transformer<ByteBuf, ByteBuf> json(int maxLength); static Observable<ByteBuf> autoReleaseCopy(Observable<ByteBuf> stream); static Observable.Transformer<ByteBuf, ByteBuf> lines(int maxLength); static Observable.Transformer<ByteBuf, ServerSentEvent> sse(int maxLength); static Func1<ByteBuf, byte[]> toByteArray(); static Observable.Transformer<ByteBuf, byte[]> aggrByteArray(); static Observable.Transformer<ByteBuf, String> toString(String enc); }
ByteBufs { public static Observable.Transformer<ByteBuf, ByteBuf> json() { return json(1024 * 1024); } private  ByteBufs(); static Observable.Transformer<ByteBuf, ByteBuf> gzip(); static Observable.Transformer<ByteBuf, ByteBuf> gunzip(); static Observable.Transformer<ByteBuf, ByteBuf> json(); static Observable.Transformer<ByteBuf, ByteBuf> json(int maxLength); static Observable<ByteBuf> autoReleaseCopy(Observable<ByteBuf> stream); static Observable.Transformer<ByteBuf, ByteBuf> lines(int maxLength); static Observable.Transformer<ByteBuf, ServerSentEvent> sse(int maxLength); static Func1<ByteBuf, byte[]> toByteArray(); static Observable.Transformer<ByteBuf, byte[]> aggrByteArray(); static Observable.Transformer<ByteBuf, String> toString(String enc);  }