RxJava/RxScala async code inside Observer.onNext

Question

Let's assume that you want to store events from some stream into the database and a client library to that database is asynchronous.

E.g. there is a method writeEvent(event: MyEvent): Future[Boolean] that you have to call inside onNext of your observer when the next event is emitted. Is there a good way to do this otherwise than blocking on the Future?

The only way that I currently see on how to implement this, is to create some custom Scheduler that allows me to return thread to the pool, until async code inside onNext is complete.

drhr · Accepted Answer · 2016-10-23 06:45:12Z

2

You do not want to block like that inside your onNext subscriber callback, that would defeat Rx. It can be chained together in a more idiomatic way.

I don't work with Futures much, but I wonder if Observable.from(Future) would work:

someStream
    .flatMap(evt => Observable.from(writeEvent(evt)))
    .subscribe(
        next => ...,
        err => ...
)

answered Oct 23, 2016 at 6:45

drhr

2,2912 gold badges18 silver badges36 bronze badges

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

1 Comment

Magomed Abdurakhmanov Over a year ago

Thanks for your answer. It seems that it could work, I'll try it and report later.

Collectives™ on Stack Overflow

RxJava/RxScala async code inside Observer.onNext

1 Answer 1

1 Comment

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Your Answer

Sign up or log in

Post as a guest

Related