RxJava Observable buffer based on content

Question

I started a project using vertX and RxJava, and I have a problem for which I don't find a solution.

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.

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.

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

Another way seems to use compose 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.

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.

thank you for reading.

Related (in case you need to keep track of state, i.e., buffer based on content spanning multiple emitted items): RxJava buffer/window with custom counting criteria — Steven Jeuris
– Steven Jeuris, Commented Dec 19, 2018 at 17:15

Community · Accepted Answer · 2017-05-23 11:44:17Z

2

I guess you have to use the buffer operator for this (maybe you could do with the simpler buffer, but I'm not sure about that). See also this other question that covers roughly the same topic and this GitHub page for more discussion. Hope this helps you!

edited May 23, 2017 at 11:44

CommunityBot

11 silver badge

answered Jul 11, 2016 at 17:54

RvanHeest

8696 silver badges12 bronze badges

Sign up to request clarification or add additional context in comments.

2 Comments

Steve Marion Over a year ago

Very good answer, I have implemented most of the functionnality, but I still have a problem. the closingObservable get notified first by the publisher, while I need the item to be added before the buffer emits (I cannot wait the next frame to emit the previous message)

Steve Marion Over a year ago

After much try, I found a solution. The problem come from buffer which subscribe on the closing observable before subscribing from the producer. My solution is to chain .delaySubscription(1, TimeUnit.NANOSECONDS) to the closing observable to force subscription after, and hence receive the notification after.

Collectives™ on Stack Overflow

RxJava Observable buffer based on content

1 Answer 1

2 Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

2 Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related