@Test public void sse() throws Exception { byte[] data = "event:\ndata: foo\ndata:bar\n\ndata: baz\n".getBytes(StandardCharsets.UTF_8); List<ServerSentEvent> events = obs(data) .compose(ByteBufs.sse(10)) .reduce(new ArrayList<ServerSentEvent>(), (acc, v) -> { acc.add(v); return acc; }) .toBlocking() .single(); List<ServerSentEvent> expected = new ArrayList<>(); expected.add(new ServerSentEvent("event", "")); expected.add(new ServerSentEvent("data", "foo")); expected.add(new ServerSentEvent("data", "bar")); expected.add(new ServerSentEvent("data", " baz")); Assert.assertEquals(expected, events); }
public static Observable.Transformer<ByteBuf, ServerSentEvent> sse(int maxLength) { return input -> input .compose(lines(maxLength)) .map(ServerSentEvent::parse) .filter(Objects::nonNull); }
ByteBufs { public static Observable.Transformer<ByteBuf, ServerSentEvent> sse(int maxLength) { return input -> input .compose(lines(maxLength)) .map(ServerSentEvent::parse) .filter(Objects::nonNull); } }
ByteBufs { public static Observable.Transformer<ByteBuf, ServerSentEvent> sse(int maxLength) { return input -> input .compose(lines(maxLength)) .map(ServerSentEvent::parse) .filter(Objects::nonNull); } private  ByteBufs(); }
ByteBufs { public static Observable.Transformer<ByteBuf, ServerSentEvent> sse(int maxLength) { return input -> input .compose(lines(maxLength)) .map(ServerSentEvent::parse) .filter(Objects::nonNull); } 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, ServerSentEvent> sse(int maxLength) { return input -> input .compose(lines(maxLength)) .map(ServerSentEvent::parse) .filter(Objects::nonNull); } 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);  }