Going Reactive with Eclipse Vert.x and RxJava

package samples;

import io.vertx.core.Vertx;

public class HttpApplication {

    public static void main(String[] args) {
        // 1 - Create a Vert.x instance
        Vertx vertx = Vertx.vertx();

        // 2 - Create the HTTP server
        vertx.createHttpServer()
            // 3 - Attach a request handler processing the requests
            .requestHandler(req -> req.response()
                    .end("Hello, request handled from "
                        + Thread.currentThread().getName()))
            // 4 - Start the server on the port 8080
            .listen(8080);
    }
}

package samples;

import io.vertx.core.Vertx;
import io.vertx.ext.web.client.WebClient;

public class TwitterFeedApplication {

    public static void main(String[] args) {
        Vertx vertx = Vertx.vertx();
        // 1 - Create a Web client
        WebClient client = WebClient.create(vertx);
        vertx.createHttpServer()
            .requestHandler(req -> {
                // 2 - In the request handler, retrieve a Twitter feed
                client
                    .getAbs("https://twitter.com/vertx_project")
                    .send(res -> {
                        // 3 - Write the response based on the result
                        if (res.failed()) {
                            req.response().end("Cannot access "
                                + "the twitter feed: "
                                + res.cause().getMessage());
                        } else {
                            req.response().end(res.result()
                                .bodyAsString());
                        }
                    });
            })
            .listen(8080);
    }
}

package samples;

import io.vertx.reactivex.core.Vertx;
import io.vertx.reactivex.core.http.HttpServer;
import io.vertx.reactivex.ext.web.client.HttpResponse;
import io.vertx.reactivex.ext.web.client.WebClient;

public class RXTwitterFeedApplication {

    public static void main(String[] args) {
        Vertx vertx = Vertx.vertx();
        WebClient client = WebClient.create(vertx);
        HttpServer server = vertx.createHttpServer();
        server
            // 1 - Transform the sequence of request into a stream
            .requestStream().toFlowable()
            // 2 - For each request, call the twitter API
            .flatMapCompletable(req ->
                client.getAbs("https://twitter.com/vertx_project")
                    .rxSend()
                    // 3 - Extract the body as string
                    .map(HttpResponse::bodyAsString)
                    // 4 - In case of a failure
                    .onErrorReturn(t -> "Cannot access the twitter " +
                        "feed: " + t.getMessage())
                    // 5 - Write the response
                    .doOnSuccess(res -> req.response().end(res))
                    // 6 - Just transform the restul into a completable
                    .toCompletable()
            )
            // 7 - Never forget to subscribe to a reactive type,
            // or nothing happens
            .subscribe();

        server.listen(8080);
    }
}

public class BiddingServiceVerticle extends AbstractVerticle {

    private final Logger logger = 
        LoggerFactory.getLogger(BiddingServiceVerticle.class);

    @Override
    public void start(Future<Void> verticleStartFuture) throws Exception {
        Random random = new Random();
        String myId = UUID.randomUUID().toString();
        int portNumber = config().getInteger("port", 3000);

       // (...)
    }
}

Router router = Router.router(vertx);
router.get("/offer").handler(context -> {
    String clientIdHeader = context.request()
        .getHeader("Client-Request-Id");
    String clientId =
        (clientIdHeader != null) ? clientIdHeader : "N/A";
    int myBid = 10 + random.nextInt(20);
    JsonObject payload = new JsonObject()
        .put("origin", myId)
        .put("bid", myBid);
    if (clientIdHeader != null) {
        payload.put("clientRequestId", clientId);
    }
    long artificialDelay = random.nextInt(1000);
    vertx.setTimer(artificialDelay, id -> {
        if (random.nextInt(20) == 1) {
            context.response()
                .setStatusCode(500)
                .end();
            logger.error("{} injects an error (client-id={}, "
                + "artificialDelay={})",
                myId, myBid, clientId, artificialDelay);
        } else {
            context.response()
                .putHeader("Content-Type",
                    "application/json")
                .end(payload.encode());
            logger.info("{} offers {} (client-id={}, " +
                "artificialDelay={})",
                myId, myBid, clientId, artificialDelay);
        }
    });
});

vertx.createHttpServer()
    .requestHandler(router::accept)
    .listen(portNumber, ar -> {
        if (ar.succeeded()) {
            logger.info("Bidding service listening on HTTP " +
                "port {}", portNumber);
            verticleStartFuture.complete();
        } else {
            logger.error("Bidding service failed to start",
                ar.cause());
            verticleStartFuture.fail(ar.cause());
        }
});

public class BestOfferServiceVerticle extends AbstractVerticle {

    private static final JsonArray DEFAULT_TARGETS = new JsonArray()
        .add(new JsonObject()
            .put("host", "localhost")
            .put("port", 3000)
            .put("path", "/offer"))
        .add(new JsonObject()
            .put("host", "localhost")
            .put("port", 3001)
            .put("path", "/offer"))
        .add(new JsonObject()
            .put("host", "localhost")
            .put("port", 3002)
            .put("path", "/offer"));
    private final Logger logger = LoggerFactory
        .getLogger(BestOfferServiceVerticle.class);
    private List<JsonObject> targets;
    private WebClient webClient;

    @Override
    public void start(Future<Void> startFuture) throws Exception {
        webClient = WebClient.create(vertx);

        targets = config().getJsonArray("targets",
            DEFAULT_TARGETS)
            .stream()
            .map(JsonObject.class::cast)
            .collect(Collectors.toList());

        vertx.createHttpServer()
            .requestHandler(this::findBestOffer)
            .rxListen(8080)
            .subscribe((server, error) -> {
                if (error != null) {
                    logger.error("Could not start the best offer " +
                        "service", error);
                    startFuture.fail(error);
                } else {
                    logger.info("The best offer service is running " +
                        "on port 8080");
                    startFuture.complete();
                }
            });
}

private final AtomicLong requestIds = new AtomicLong();
private static final JsonObject EMPTY_RESPONSE = new JsonObject()
    .put("empty", true)
    .put("bid", Integer.MAX_VALUE);

private void findBestOffer(HttpServerRequest request) {
    String requestId = String.valueOf(requestIds.getAndIncrement());

    List<Single<JsonObject>> responses = targets.stream()
        .map(t -> webClient
            .get(t.getInteger("port"),
                t.getString("host"),
                t.getString("path"))
            .putHeader("Client-Request-Id",
                String.valueOf(requestId))
            .as(BodyCodec.jsonObject())
            .rxSend()
            .retry(1)
            .timeout(500, TimeUnit.MILLISECONDS,
                RxHelper.scheduler(vertx))
            .map(HttpResponse::body)
            .map(body -> {
                logger.info("#{} received offer {}", requestId,
                    body.encodePrettily());
                return body;
            })
            .onErrorReturnItem(EMPTY_RESPONSE))
        .collect(Collectors.toList());

    Single.merge(responses)
        .reduce((acc, next) -> {
            if (next.containsKey("bid") && isHigher(acc, next)) {
                return next;
            }
            return acc;
        })
        .flatMapSingle(best -> {
            if (!best.containsKey("empty")) {
                return Single.just(best);
            } else {
                return Single.error(new Exception("No offer " +
                    "could be found for requestId=" + requestId));
            }
        })
        .subscribe(best -> {
            logger.info("#{} best offer: {}", requestId,
                best.encodePrettily());
            request.response()
                .putHeader("Content-Type",
                    "application/json")
                .end(best.encode());
        }, error -> {
            logger.error("#{} ends in error", requestId, error);
            request.response()
                .setStatusCode(502)
                .end();
        });
}

public class MainVerticle extends AbstractVerticle {

  @Override
  public void start() {
    vertx.deployVerticle(new BiddingServiceVerticle());

    vertx.deployVerticle(new BiddingServiceVerticle(), 
                         new DeploymentOptions().setConfig(
                             new JsonObject().put("port", 3001)));

    vertx.deployVerticle(new BiddingServiceVerticle(), 
                         new DeploymentOptions().setConfig(
                             new JsonObject().put("port", 3002)));

    vertx.deployVerticle("samples.BestOfferServiceVerticle", 
                         new DeploymentOptions().setInstances(2));
  }
}

$ http GET localhost:3000/offer 'Client-Request-Id:1234' --verbose
GET /offer HTTP/1.1
Accept: */*
Accept-Encoding: gzip, deflate
Client-Request-Id: 1234
Connection: keep-alive
Host: localhost:3000
User-Agent: HTTPie/0.9.9

HTTP/1.1 200 OK
Content-Length: 83
Content-Type: application/json

{
       "bid": 21,
       "clientRequestId": "1234",
       "origin": "fe299565-34be-4a7b-ac09-d88fcc1e42e2"
}