RxJava2 : Multiple exit criteria for gouping

[DareUrDream]

HI, I have the below situation where I want to determine whether my group is complete or not based my takeUntil operator. But recently we got into a situation where the record being streamed by the database was delayed by few seconds (due to network) and we ended up processing incomplete group of records as the last record reached before the other needed records. So, now I am planning to implement a solution where I wish to wait for X few seconds once the takeUntil operator returns true.

Any help is much appreciated in how to achieve the same.

Flowable<List<DBRecord>> flowable = observable
					.toFlowable(BackpressureStrategy.BUFFER)
					.groupBy(callrecord -> callrecord.getSessionid())
					.flatMapSingle(group -> 
						group
							.timeout(1, TimeUnit.HOURS) // Certain cases records can come within one
							.onErrorResumeNext(Flowable.never())
							.takeUntil(cr -> cr instanceof CCDR && ((CCDR) cr).isCallCompleted())
							.toList()
					);

P.S - Is it possible to take the group members out in a list and analyze the records in a custom function to determine if the group is complete. In other words something like collectUntil/takeUntil the custom predicate accepting the list of all elements returns true.

Cheers !!
DareUrDream

[akarnokd]

Hi. Questions such as "How do I do X with RxJava" are better suited StackOverflow.

we ended up processing incomplete group of records as the last record reached before the other needed records

I don't undestand this given the example. If a group ends when the takeUntil indicates, how can it be incomplete? Groups end when the main source completes.

[DareUrDream]

In our case, it is the database that is streaming records as and when CRUD operations are performed. For us a group is logically complete when I have a set of N different kind records(not known in advance but can be determined If I can access the records), whose end is indicate by the takeUntil operator. Now since the database is remote, it so happened that the one of the record did not reach us until when the takeUntil was evaluated. In other words, I received the last record before the another needed record.

P.S - Sure will use StackOverflow for these kinds of questions going forward.

Cheers !!
DareUrDream

[davidmoten]

@DareUrDream terminating an inner stream in a groupBy can mean loss of in-flight items. Don't do it. Use buffering of some sort.

[DareUrDream]

Sure @davidmoten Thanks...

[DareUrDream]

@davidmoten Would appreciate your suggestion on reactor/reactor-core#1543 ... They are related issues and I am doing the needful in RxJava

[davidmoten]

@DareUrDream Your use case is not clear to me. Can you describe your input and desired output without reference to RxJava operators? Abstraction helps too (I don't really care that the input came from a database for example).

[DareUrDream]

Closing the question