BinaryChannelSupplier extends AbstractAsyncCloseable { public final <T> ChannelSupplier<T> parseStream(ByteBufsDecoder<T> decoder) { return ChannelSupplier.of( () -> parse(decoder) .thenEx((value, e) -> { if (e == null) return Promise.of(value); if (e == UNEXPECTED_END_OF_STREAM_EXCEPTION && bufs.isEmpty()) return Promise.of(null); return Promise.ofException(e); }), this); } protected  BinaryChannelSupplier(ByteBufQueue bufs); protected  BinaryChannelSupplier(); ByteBufQueue getBufs(); abstract Promise<Void> needMoreData(); abstract Promise<Void> endOfStream(); static BinaryChannelSupplier ofIterable(Iterable<ByteBuf> iterable); static BinaryChannelSupplier ofIterator(Iterator<ByteBuf> iterator); static BinaryChannelSupplier of(ChannelSupplier<ByteBuf> input); static BinaryChannelSupplier ofProvidedQueue(ByteBufQueue queue,
			AsyncSupplier<Void> get, AsyncSupplier<Void> complete, AsyncCloseable closeable); final Promise<T> parse(ByteBufsDecoder<T> decoder); final Promise<T> parseRemaining(ByteBufsDecoder<T> decoder); final ChannelSupplier<T> parseStream(ByteBufsDecoder<T> decoder); Promise<Void> bindTo(BinaryChannelInput input); static final Exception UNEXPECTED_DATA_EXCEPTION; static final Exception UNEXPECTED_END_OF_STREAM_EXCEPTION; }
@Test public void testParseStream() { ByteBuf buf = await(BinaryChannelSupplier.of(ChannelSupplier.of(wrapUtf8("Hello\r\n World\r\n"))) .parseStream(ofCrlfTerminatedBytes()) .toCollector(ByteBufQueue.collector())); assertEquals("Hello World", buf.asString(UTF_8)); }