ListenableFuture to scala Future

Question

I am in the process of writing a small scala wrapper around a java library.

The java library has an object QueryExecutor exposing 2 methods:

• execute(query): Result
• asyncExecute(query): ListenableFuture[Result]

ListenableFuture in this context is the one from the guava library.

I want my scala wrapper to return a Future[Result] instead of the java object, but I am not sure what is the best way to implement that. Here are 2 solutions I came up with:

future {
  executor.execute(query)
}

and

val p = promise[Result]
val guavaFuture = executor.asyncExecute(query)

Futures.addCallback(guavaFuture, new FutureCallback[Result] {
  def onFailure(t: Throwable) {
    p.failure(t)
  }

  def onSuccess(result: Result) {
    p.success(result)
  }
})

p.future

I am wondering which method is the best. My intuition is that the first one, while returning a Future, will still block a thread while the call to execute waits for a response, the second one looks like it should be really non blocking. Any comment on the pros/cons of each method ?

Joni

Lets assume you have 4 processors. In this case default ExecutionContext consists of 4 workers. Each future { executor.execute(query) } blocks 1 worker, so 4 "futures" will block your program entirely. You could create additional ExecutionContext for blocking operations, but there would be some overhead. – Vtarj

Thanks @senia, that's what I thought. The first code is async from the caller's point of view but will still block a thread of the ExecutionContext, while the second one is really non blocking (assuming that asyncExecute uses non blocking IO). I feel like this is a very basic question but I am not very familiar with Promises. – Joni

I found this to be helpful in a similar (or possible even identical) requirment : github.com/eigengo/activator-akka-cassandra/blob/master/src/… – Illailladvised

Answer 1

The second option is best, it keeps everything asynchronous. but... you can do one better and abstract the solution into a reusable pattern:

implicit class RichListenableFuture[T](lf: ListenableFuture[T]) {
  def asScala: Future[T] = {
    val p = Promise[T]()
    Futures.addCallback(lf, new FutureCallback[T] {
      def onFailure(t: Throwable): Unit = p failure t
      def onSuccess(result: T): Unit    = p success result
    })
    p.future
  }    
}

You can then simply call:

executor.asyncExecute(query).asScala

Ascidium

Awesome solution. Can't believe how much cleaner my ListenableFuture interfacing code is with this implicit – Etz

Very nice. could be even more abstract using Promise[T] – Crinite

Can't help but point out that toPromise actually returns a Future. ;) Otherwise, love it! – Paternalism

Yeah... fixed that :) – Ascidium

nice solution.. I am trying to create test cases based on executeAsync, please can someone guide me on that. my code looks like this: val selectResult: ResultSet = stub[ResultSet] val dbSession = stub[Session]val rows = new util.ArrayList[Row](); val row = stub[Row] val resultSetFuture: ResultSetFuture = stub[ResultSetFuture] (dbSession.executeAsync(_: String)).when(query).returns(resultSetFuture) – Midweek

What are the import(s) ? – Clitoris

In 2019, .addCallback needs an executor as a third argument. Gave it MoreExecutors.directExecutor(). Initialised that once as a private. – Sibling

Answer 2

Another, slightly shorter solution:

implicit class ListenableFutureDecorator[T](val f: ListenableFuture[T]) extends AnyVal {
  def asScala(implicit e: Executor): Future[T] = {
    val p = Promise[T]()
    f.addListener(() => p.complete(Try(f.get())), e)
    p.future
  }
}

Rebroadcast

Looks like implicitly[Executor] could just be e in the above – Portauprince

Thanks, @Portauprince - I've updated the answer above. – Rebroadcast