CODEWITHSUNDEEP
javarx-javareactive-programmingvert.x
Sparky
Sparky

Reputation: 2764

Equivalent of VertX CompositeFuture in RxJava

The VertX example for when you need to query multiple asynchronous resources and use them all in a single operation is:

Future<HttpServer> httpServerFuture = Future.future();
httpServer.listen(httpServerFuture.completer());

Future<NetServer> netServerFuture = Future.future();
netServer.listen(netServerFuture.completer());

CompositeFuture.all(httpServerFuture, netServerFuture).setHandler(ar -> {
  if (ar.succeeded()) {
    // All servers started
  } else {
    // At least one server failed
  }
});

We need to query two different databases and then use the results in business logic, but the flow is equivalent.

What's the VertX/RxJava equivalent?

Currently people are doing this by nesting a new .flatMap() call every time they need a new variable. I'm left feeling there must be a better way...

We don't actually need the queries to be concurrent but we need to cache both results and pass them to the business logic at the same time some how.

Upvotes: 0

Views: 812

Answers (1)

homerman
homerman

Reputation: 3569

there are many ways to do this, but i've tried to pick an approach that tacks closely to your sample:

@Override
public void start(Future<Void> startFuture) throws Exception {
    final HttpServer httpServer = vertx.createHttpServer();
    final Completable initializeHttpServer = httpServer.rxListen().toCompletable();

    final NetServer netServer = vertx.createNetServer();
    final Completable initializeNetServer = netServer.rxListen().toCompletable();

    initializeHttpServer.andThen(initializeNetServer)
        .subscribe(
            ()    -> { /* All servers started */ },
            error -> { /* At least one server failed */ }
        );
}

the rxListen() invocations are converted into Completable instances, which are then run serially upon subscription.

  • the subscriber's onComplete callback will be invoked when both servers are done binding to their respective ports, or...
  • the onError callback will be invoked if an exception occurs

(also, fwiw, "nesting" flatMap operations for something as trivial as this shouldn't be necessary. "chaining" such operations, however, would be idiomatic usage).

hope that helps!

--UPDATE--

having read the question more carefully, i now see that you were actually asking about how to handle the results of two discrete asynchronous operations.

an alternative to flatMap'ing your way to combining the results would be to use the zip operator, like so:

    @Override
    public void start(Future<Void> startFuture) throws Exception {
        final Single<String> dbQuery1 = Single.fromCallable(() -> { return "db-query-result-1"; });
        final Single<String> dbQuery2 = Single.fromCallable(() -> { return "db-query-result-2"; });

        Single.zip(dbQuery1, dbQuery2, (result1, result2) -> {
            // handle the results from both db queries
            // (with Pair being a standard tuple-like class)
           return new Pair(result1, result2);
        })
            .subscribe(
                pair -> {
                    // handle the results
                },
                error -> {
                    // something went wrong
                }
            );
    }

per the docs, zip allows you to specify a series of reactive types (Single, Observable, etc) along with a function to transform all the results at once, with the central idea being that it will not emit anything until all the sources have emitted once (or more, depending on the reactive type).

Upvotes: 2

Related Questions