How do I map an Observable with a function from Future to Future?

Question

Suppose I have an event stream of elements with type In:

val observableIn: Observable[In] = ???

And a function for transforming objects of type In into objects of type Out, but "in the future":

val futureInToFutureOut: (Future[In]) => Future[Out] = ???

At this point I want to transform the elements of my observableIn according to my function futureInToFutureOut. That is, I want as my result an event stream of elements of type Out, matching the elements of the original stream, but converted via function futureInToFutureOut.

I think this should work:

val observableOut: Observable[Out] = observableIn flatMap { in =>
  Observable.from(futureInToFutureOut(Future(in)))
}

Is this right? Is there a better way to do this?

score 2 · Accepted Answer · edited May 23 '17 at 12:08

EDIT:

Your solution is correct as far as I can tell. If you want to improve performance a bit, consider:

val observableOut: Observable[Out] = observableIn.flatMap { in =>
  val p = Promise.successful(in)
  Observable.from(futureInToFutureOut(p.future))
}

This is a bit faster, as it does not create an asynchronous computation to map the future like Future.apply does.

OLD:

I am leaving my old suggestion below, which works only in the case that you are mapping a single event in the Observable.

import scala.concurrent._
val promiseIn = Promise[Int]()
observableIn.foreach(x => promiseIn.trySuccess(x))
val observableOut = Observable.create { observer =>
  promiseIn.future.map(futureInToFutureOut).foreach { y =>
    observer.onNext(y)
    observer.onCompleted()
  }
}

Explanation

Since you are starting out with an Observable[In] object (i.e. an event stream), you need to find a way to transfer an event from that Observable to a future. The typical way to create a new future is to create its Promise first -- the input side of the future object. You then use foreach on the Observable to invoke trySuccess on the future when the first event arrives:

observableIn ---x---> foreach ---x---> promiseIn.trySuccess(x)

Once the event on the Observable arrives, the promise will be asynchronously completed. We can now get the reading side of the promise, i.e. the future by calling its future method; and then call map on the future -- promiseIn.future.map(futureInToFutureOut). Graphically:

promiseIn.future ---x---> map ---f(x)---> futureOut

The resulting future is asynchronously completed with futureInToFutureOut(x). At this point we need to find a way to emit this value back through an Observable[Out]. A typical way to create new Observable is to call the Observable.create factory method. This method gives as the Observable's writing end -- the Observer, which we use to emit events to by calling onNext:

futureOut ---f(x)---> foreach ---f(x)---> observer.onNext(f(x))

Since we know that a future emits at most a single event, we call the onCompleted on the observer, to close the output Observable.

Edit: if you want to master Rx and Scala futures, you may want to consider this book, which deals with these topics. Disclaimer: I'm the author.

Can you really achieve this with a single `Promise`? Maybe my question wasn't clear, but my `Observable[In]` contains multiple elements in the stream, and I want to map them all into the `Observable[Out]` stream. — john sullivan, Jul 13 '15 at 19:27
If there is something wrong with my proposed solution, could you explain what the problem is? — john sullivan, Jul 13 '15 at 19:27
No, in case you want to map all of the events from the input stream, your solution should be correct, modulo the `flapMap` typo. — axel22, Jul 13 '15 at 19:32
Thanks so much. Your original answer was very informative, if not actually a solution to my problem. I'm actually working on something else for which your example of using `Observable.create` is very helpful. I'm going to read your book as well, it should help me flesh out my understanding of this stuff. — john sullivan, Jul 13 '15 at 19:41
in your OLD example, I think you need to add a `Subscription()` before the final `}`. on something similar I get a "found: Unit required: Subscription" compiler error. — john sullivan, Jul 14 '15 at 15:02
Yes, of course - you can add an empty subscription to fix the return type, although, note that calling `unsubscribe` will not actually prevent the future from propagating the event. — axel22, Jul 14 '15 at 16:26
Let us [continue this discussion in chat](http://chat.stackoverflow.com/rooms/83253/discussion-between-john-sullivan-and-axel22). — john sullivan, Jul 14 '15 at 19:36

How do I map an Observable with a function from Future to Future?

1 Answers1

Explanation