RxJava Observable buffer based on content
<p>I started a project using vertX and RxJava, and I have a problem for which I don't find a solution.</p>

<p>I have an Observable that emits WebSocketFrame for incoming communications,
each WebSocketFrame is composed of a payload (a ByteBuffer) and flags that indicates wether it is the first frame of the message or the last.</p>

<p>I want to make an operation on this Observable to transform it to an Observable that emits ByteBufferd that contain all the frame of a each message.</p>

<p>I tried the <code>buffer</code> method but it seems to be designed to regroup items by an arbitrary criteria (time or another observable).</p>

<p>Another way seems to use <code>compose</code> to subscribe to the WebSocketFrame observable, to add to buffer on non-ending frame, and to "feed" the ByteBuffer Observable on ending frame. But I don't know how to create and feed a buffer manually.</p>

<p>So if someone has already seen this issue (which IMHO seems pretty common) and have enough knowledge of RxJava for proposing an implementation I would be most gratefull.</p>

<p>thank you for reading.</p>