= Vert.x RxJava
:toc: left

== Vert.x API for RxJava3

https://github.com/ReactiveX/RxJava[RxJava] is a popular library for composing asynchronous and event based programs using
observable sequences for the Java VM.

Vert.x integrates naturally with RxJava, allowing using RxJava wherever you can use streams or asynchronous results.

=== Using Vert.x API for RxJava3

To use Vert.x API for RxJava3, add the following dependency to the _dependencies_ section of your build descriptor:

* Maven (in your `pom.xml`):

[source,xml,subs="+attributes"]
----
<dependency>
 <groupId>io.vertx</groupId>
 <artifactId>vertx-rx-java3</artifactId>
 <version>4.1.2</version>
</dependency>
----

* Gradle (in your `build.gradle` file):

[source,groovy,subs="+attributes"]
----
compile 'io.vertx:vertx-rx-java3:4.1.2'
----

There are two ways for using the RxJava 3 API with Vert.x:

* via the original Vert.x API with helpers class that provides static methods for converting objects between Vert.x core
API and RxJava 3 API
** `link:../../apidocs/io/vertx/rxjava3/RxHelper.html[RxHelper]`
** `link:../../apidocs/io/vertx/rxjava3/ObservableHelper.html[ObservableHelper]`
** `link:../../apidocs/io/vertx/rxjava3/FlowableHelper.html[FlowableHelper]`
** `link:../../apidocs/io/vertx/rxjava3/SingleHelper.html[SingleHelper]`
** `link:../../apidocs/io/vertx/rxjava3/MaybeHelper.html[MaybeHelper]`
** `link:../../apidocs/io/vertx/rxjava3/CompletableHelper.html[CompletableHelper]`
* via the _Rxified_ Vert.x API enhancing the core Vert.x API.

=== Read stream support

RxJava `Flowable` is a perfect match for Vert.x `ReadStream` class : both provide a flow of items.

The `link:../../apidocs/io/vertx/rxjava3/FlowableHelper.html#toFlowable-io.vertx.core.streams.ReadStream-[FlowableHelper.toFlowable]` static methods convert
a Vert.x read stream to a `Flowable`:

[source,java]
----
FileSystem fileSystem = vertx.fileSystem();
fileSystem.open("/data.txt", new OpenOptions(), result -> {
  AsyncFile file = result.result();
  Flowable<Buffer> observable = FlowableHelper.toFlowable(file);
  observable.forEach(data -> System.out.println("Read data: " + data.toString("UTF-8")));
});
----

The _Rxified_ Vert.x API provides a `link:../../apidocs/io/vertx/rxjava3/core/streams/ReadStream.html#toFlowable--[toFlowable]`  method on
`link:../../apidocs/io/vertx/rxjava3/core/streams/ReadStream.html[ReadStream]`:

[source,java]
----
FileSystem fs = vertx.fileSystem();
fs.rxOpen("/data.txt", new OpenOptions())
  .flatMapPublisher(file -> file.toFlowable())
  .subscribe(data -> System.out.println("Read data: " + data.toString("UTF-8")));
----

Such flowables are *hot* flowables, i.e. they will produce notifications regardless of subscriptions because
a `ReadStream` can potentially emit items spontaneously or not, depending on the implementation:

At subscription time, the adapter calls `link:../../apidocs/io/vertx/core/streams/ReadStream.html#handler-io.vertx.core.Handler-[handler]`
to set its own handler.

Some `ReadStream` implementations can start to emit events after this call, others will emit events wether an
handler is set:

- `AsyncFile` produces buffer events after the handler is set
- `HttpServerRequest` produces events independantly of the handler (i.e buffer may be lost if no handler is set)

In both cases, subscribing to the `Flowable` in the same call is safe because the event loop or the worker
verticles cannot be called concurrently, so the subscription will always happens before the handler starts emitting
data.

When you need to delay the subscription, you need to `pause` the `ReadStream` and then `resume` it, which is what
you would do with a `ReadStream`.

[source,java]
----
server.requestHandler(request -> {
  if (request.method() == HttpMethod.POST) {

    // Stop receiving buffers
    request.pause();

    checkAuth(res -> {

      // Now we can receive buffers again
      request.resume();

      if (res.succeeded()) {
        Flowable<Buffer> flowable = request.toFlowable();
        flowable.subscribe(buff -> {
          // Get buffers
        });
      }
    });
  }
});
----

Likewise it is possible to turn an existing `Flowable` into a Vert.x `ReadStream`.

The `link:../../apidocs/io/vertx/rxjava3/FlowableHelper.html#toReadStream-io.reactivex.rxjava3.core.Flowable-[FlowableHelper.toReadStream]`  static methods convert
a `Flowable` to a Vert.x read stream:

[source,java]
----
Flowable<Buffer> observable = getFlowable();
ReadStream<Buffer> readStream = FlowableHelper.toReadStream(observable);
Pump pump = Pump.pump(readStream, response);
pump.start();
----

=== Write stream support

A `link:../../apidocs/io/vertx/core/streams/WriteStream.html[WriteStream]`, like a `org.reactivestreams.Subscriber`, consumes items, and, when it can't keep-up, collaborates with the producer to avoid an ever-growing backlog.

Vert.x provides the `link:../../apidocs/io/vertx/rxjava3/WriteStreamSubscriber.html[WriteStreamSubscriber]` adapter that you can use to send `Flowable` items to any `link:../../apidocs/io/vertx/core/streams/WriteStream.html[WriteStream]`:

.Sending buffers to an HTTP server response
[source,java]
----
response.setChunked(true);
WriteStreamSubscriber<io.vertx.core.buffer.Buffer> subscriber = io.vertx.rxjava3.RxHelper.toSubscriber(response);
flowable.subscribe(subscriber);
----

TIP: There is also a `io.vertx.rxjava3.WriteStreamObserver` adapter for the non-backpressured `io.reactivex.Observable`.
The difference is that this adapter will send items to the `link:../../apidocs/io/vertx/core/streams/WriteStream.html[WriteStream]` even when it can't keep-up with the producer rate.

If you are progamming with the _Rxified_ Vert.x API, the `link:../../apidocs/io/vertx/rxjava3/core/streams/WriteStream.html[WriteStream]` implementations provide a `toSubscriber` method.
The previous example then becomes even more straightforward:

[source,java]
----
response.setChunked(true);
flowable.subscribe(response.toSubscriber());
----

NOTE: When the `Flowable` terminates successfully, the adapter invokes the `link:../../apidocs/io/vertx/rxjava3/core/streams/WriteStream.html#end--[end]` method.

CAUTION: The adapter sets the `link:../../apidocs/io/vertx/rxjava3/core/streams/WriteStream.html[WriteStream]` `drain` and `exception` handlers, so don't use them after subscribing.

The `link:../../apidocs/io/vertx/rxjava3/WriteStreamSubscriber.html[WriteStreamSubscriber]` adapter is able to invoke callbacks when:

* the `Flowable` terminates with an error, or
* the `link:../../apidocs/io/vertx/rxjava3/core/streams/WriteStream.html[WriteStream]` fails (e.g. HTTP connection is closed or filesystem is full), or
* the `link:../../apidocs/io/vertx/rxjava3/core/streams/WriteStream.html[WriteStream]` ends (i.e. all writes done and file is closed), or
* the `link:../../apidocs/io/vertx/rxjava3/core/streams/WriteStream.html[WriteStream]` ends with an error (i.e. all writes done and an error occured when closing the file)

This allows for a more robust program design, as well as scheduling other tasks after the stream has been handled:

[source,java]
----
response.setChunked(true);

WriteStreamSubscriber<Buffer> subscriber = response.toSubscriber();

subscriber.onError(throwable -> {
  if (!response.headWritten() && response.closed()) {
    response.setStatusCode(500).end("oops");
  } else {
    // log error
  }
});

subscriber.onWriteStreamError(throwable -> {
  // log error
});

subscriber.onWriteStreamEnd(() -> {
  // log end of transaction to audit system...
});

flowable.subscribe(subscriber);
----

NOTE: If the `link:../../apidocs/io/vertx/rxjava3/core/streams/WriteStream.html[WriteStream]` fails, the adapter cancels the `org.reactivestreams.Subscription`.

=== Async result support

You can create an RxJava `Observer` from an existing Vert.x `Handler<AsyncResult<T>>` and subscribe
it:

[source,java]
----
Handler<AsyncResult<String>> handler = getHandler();

// Subscribe to a Single
Single.just("hello").subscribe(SingleHelper.toObserver(handler));
----

[source,java]
----
Handler<AsyncResult<String>> handler = getHandler();

// Subscribe to a Single
Maybe.just("hello").subscribe(MaybeHelper.toObserver(handler));
----

[source,java]
----
Handler<AsyncResult<Void>> handler = getHandler();

// Subscribe to a Single
Completable.complete().subscribe(CompletableHelper.toObserver(handler));
----

The _Rxified_ Vert.x API duplicates each such method with the `rx` prefix that returns an RxJava `Single`,
`Maybe` or `Completable`:

[source,java]
----
Single<HttpServer> single = vertx
  .createHttpServer()
  .rxListen(1234, "localhost");

// Subscribe to bind the server
single.
    subscribe(
        server -> {
          // Server is listening
        },
        failure -> {
          // Server could not start
        }
    );
----

Such single are *cold* singles, and the corresponding API method is called on subscribe.

`Maybe` can produce a result or no result:

[source,java]
----
DnsClient client = vertx.createDnsClient(dnsPort, dnsHost);

// Obtain a maybe that performs the actual reverse lookup on subscribe
Maybe<String> maybe = client.rxReverseLookup(ipAddress);

// Subscribe to perform the lookup
maybe.
  subscribe(
    name -> {
      // Lookup produced a result
    },
    failure -> {
      // Lookup failed
    },
    () -> {
      // Lookup produced no result
    }
  );
----

`Completable` is usually mapped to `Handler<AsyncResult<Void>>`

[source,java]
----
Completable single = server.rxClose();

// Subscribe to close the server
single.
  subscribe(
    () -> {
      // Server is closed
    },
    failure -> {
      // Server closed but encountered issue
    }
  );
----

[TIP]
====
If you cannot use the Vert.x _Rxified_ API or, if you have your own, callback-based, asynchronous methods, Vert.x provides adapters:

* `link:../../apidocs/io/vertx/rxjava3/SingleHelper.html#toSingle-java.util.function.Consumer-[SingleHelper.toSingle]`,
* `link:../../apidocs/io/vertx/rxjava3/MaybeHelper.html#toMaybe-java.util.function.Consumer-[MaybeHelper.toMaybe]` and
* `link:../../apidocs/io/vertx/rxjava3/CompletableHelper.html#toCompletable-java.util.function.Consumer-[CompletableHelper.toCompletable]`.
====

[source,java]
.Adapting Vert.x core _executeBlocking_ method
----
Maybe<String> maybe = MaybeHelper.toMaybe(handler -> {
  vertx.executeBlocking(fut -> fut.complete(invokeBlocking()), handler);
});
----

=== Scheduler support

The reactive extension sometimes needs to schedule actions, for instance `Flowable#timer` creates and returns
a timer that emit periodic events. By default, scheduled actions are managed by RxJava, it means that the
timer threads are not Vert.x threads and therefore not executing in a Vert.x event loop nor on a Vert.x worker thread.

When an RxJava method deals with a scheduler, it accepts an overloaded method accepting an extra `io.reactivex.Scheduler`,
the `link:../../apidocs/io/vertx/rxjava3/RxHelper.html#scheduler-io.vertx.core.Vertx-[RxHelper.scheduler]`  method will return a scheduler that can be used
in such places.

[source,java]
----
Scheduler scheduler = RxHelper.scheduler(vertx);
Observable<Long> timer = Observable.interval(100, 100, TimeUnit.MILLISECONDS, scheduler);
----

For blocking scheduled actions, a scheduler can be created with the `link:../../apidocs/io/vertx/rxjava3/RxHelper.html#blockingScheduler-io.vertx.core.Vertx-[RxHelper.blockingScheduler]`
method:

[source,java]
----
Scheduler scheduler = RxHelper.blockingScheduler(vertx);
Observable<Long> timer = Observable.interval(100, 100, TimeUnit.MILLISECONDS, scheduler);
----

RxJava can also be reconfigured to use the Vert.x scheduler:

[source,java]
----
RxJavaPlugins.setComputationSchedulerHandler(s -> RxHelper.scheduler(vertx));
RxJavaPlugins.setIoSchedulerHandler(s -> RxHelper.blockingScheduler(vertx));
RxJavaPlugins.setNewThreadSchedulerHandler(s -> RxHelper.scheduler(vertx));
----

CAUTION: RxJava uses the words _computation_ for non-blocking tasks and _io_ for blocking tasks
which is the opposite of the Vert.x terminology

The _Rxified_ Vert.x API provides also similar method on the `link:../../apidocs/io/vertx/rxjava3/core/RxHelper.html[RxHelper]` class:

[source,java]
----
Scheduler scheduler = RxHelper.scheduler(vertx);
Observable<Long> timer = Observable.interval(100, 100, TimeUnit.MILLISECONDS, scheduler);
----

[source,java]
----
RxJavaPlugins.setComputationSchedulerHandler(s -> RxHelper.scheduler(vertx));
RxJavaPlugins.setIoSchedulerHandler(s -> RxHelper.blockingScheduler(vertx));
RxJavaPlugins.setNewThreadSchedulerHandler(s -> RxHelper.scheduler(vertx));
----

It is also possible to create a scheduler backed by a named worker pool. This can be useful if you want to re-use
the specific thread pool for scheduling blocking actions:

[source,java]
----
Scheduler scheduler = RxHelper.blockingScheduler(workerExecutor);
Observable<Long> timer = Observable.interval(100, 100, TimeUnit.MILLISECONDS, scheduler);
----

=== Json unmarshalling

The `link:../../apidocs/io/vertx/rxjava3/FlowableHelper.html#unmarshaller-java.lang.Class-[FlowableHelper.unmarshaller]` creates an `io.reactivex.rxjava3.FlowableOperator` that
transforms an `Flowable<Buffer>` in json format into an object flowable:

[source,java]
----
fileSystem.open("/data.txt", new OpenOptions(), result -> {
  AsyncFile file = result.result();
  Flowable<Buffer> observable = FlowableHelper.toFlowable(file);
  observable.compose(FlowableHelper.unmarshaller(MyPojo.class)).subscribe(
      mypojo -> {
        // Process the object
      }
  );
});
----

The same can be done with the _Rxified_ helper:

[source,java]
----
fileSystem
  .rxOpen("/data.txt", new OpenOptions())
  .flatMapObservable(file -> file.toObservable())
  .compose(ObservableHelper.unmarshaller((MyPojo.class)))
  .subscribe(mypojo -> {
    // Process the object
  });
----

=== Deploying a Verticle

To deploy existing Verticle instances, you can use `link:../../apidocs/io/vertx/rxjava3/core/RxHelper.html#deployVerticle-io.vertx.rxjava3.core.Vertx-io.vertx.core.Verticle-[RxHelper.deployVerticle]`
, it deploys a `Verticle` and returns an `Single<String>` of the deployment ID.

[source,java]
----
Single<String> deployment = RxHelper.deployVerticle(vertx, verticle);

deployment.subscribe(id -> {
  // Deployed
}, err -> {
  // Could not deploy
});
----

== Rxified API

The _rxified_ API is a code generated version of the Vert.x API. The API uses the `io.vertx.rxjava3` prefix, for instance
the `io.vertx.core.Vertx` class is translated to the `link:../../apidocs/io/vertx/rxjava3/core/Vertx.html[Vertx]` class.

Rxified API expose Vert.x asynchronous methods in two fashion

- the original method translated to a RxJava equivalent returning an eager and cached subscription
- a _rx_ prefixed derived method that invokes the original method at subscription time

```java
// Immediate write
// no need to subscribe
// completion provides the asynchronous result
response.write(buffer);

// No write happened
completion = response.rxWrite(buffer);

// Perform an actual write
completion.subscribe(() -> ..., err -> ...);
```

You can use the original method or the rxified method depending on your needs, e.g when you
don't want to subscribe or you don't care of the result you can call the original method.

== Embedding Rxfified Vert.x

Just use the `link:../../apidocs/io/vertx/rxjava3/core/Vertx.html#vertx--[Vertx.vertx]` methods:

[source,java]
----
Vertx vertx = io.vertx.rxjava3.core.Vertx.vertx();
----

== As a Verticle

Extend the `link:../../apidocs/io/vertx/rxjava3/core/AbstractVerticle.html[AbstractVerticle]` class, it will wrap it for you:

[source,java]
----
class MyVerticle extends AbstractVerticle {
  public void start() {
    // Use Rxified Vertx here
  }
}
----

Deploying an RxJava verticle is still performed by the Java deployer and does not need a specified
deployer.

Verticles having an asynchronous start can override instead the `rxStart` method and return a `Completable`:

[source,java]
----
class MyVerticle extends AbstractVerticle {
  public Completable rxStart() {
    return vertx.createHttpServer()
      .requestHandler(req -> req.response().end("Hello World"))
      .rxListen()
      .ignoreElement();
  }
}
----

== Api examples

Let's study now a few examples of using Vert.x with RxJava.

=== EventBus message stream

The event bus `link:../../apidocs/io/vertx/rxjava3/core/eventbus/MessageConsumer.html[MessageConsumer]` provides naturally an `Observable<Message<T>>`:

[source,java]
----
EventBus eb = vertx.eventBus();
MessageConsumer<String> consumer = eb.<String>consumer("the-address");
Flowable<Message<String>> flowable = consumer.toFlowable();
Disposable sub = flowable.subscribe(msg -> {
  // Got message
});

// Unregisters the stream after 10 seconds
vertx.setTimer(10000, id -> {
  sub.dispose();
});
----

The `link:../../apidocs/io/vertx/rxjava3/core/eventbus/MessageConsumer.html[MessageConsumer]` provides a stream of `link:../../apidocs/io/vertx/rxjava3/core/eventbus/Message.html[Message]`.
The `link:../../apidocs/io/vertx/rxjava3/core/eventbus/Message.html#body--[body]` gives access to a new stream of message bodies if needed:

[source,java]
----
EventBus eb = vertx.eventBus();
MessageConsumer<String> consumer = eb.<String>consumer("the-address");
Flowable<String> flowable = consumer.bodyStream().toFlowable();
----

RxJava map/reduce composition style can then be used:

[source,java]
----
Flowable<Double> flowable = vertx.eventBus().
    <Double>consumer("heat-sensor").
    bodyStream().
    toFlowable();

flowable.
    buffer(1, TimeUnit.SECONDS).
    map(samples -> samples.
        stream().
        collect(Collectors.averagingDouble(d -> d))).
    subscribe(heat -> {
      vertx.eventBus().send("news-feed", "Current heat is " + heat);
    });
----

=== Timers

Timer task can be created with `link:../../apidocs/io/vertx/rxjava3/core/Vertx.html#timerStream-long-[timerStream]`:

[source,java]
----
vertx.timerStream(1000).
    toObservable().
    subscribe(
        id -> {
          System.out.println("Callback after 1 second");
        }
    );
----

Periodic task can be created with `link:../../apidocs/io/vertx/rxjava3/core/Vertx.html#periodicStream-long-[periodicStream]`:

[source,java]
----
vertx.periodicStream(1000).
    toObservable().
    subscribe(
        id -> {
          System.out.println("Callback every second");
        }
    );
----

The observable can be cancelled with an unsubscription:

[source,java]
----
vertx.periodicStream(1000).
    toObservable().
    subscribe(new Observer<Long>() {
      private Disposable sub;
      public void onSubscribe(@NonNull Disposable d) {
        sub = d;
      }
      public void onNext(Long aLong) {
        // Callback
        sub.dispose();
      }
      public void onError(Throwable e) {}
      public void onComplete() {}
    });
----

=== Http client requests

You can easily use the http client to create a request and process the response:

[source,java]
----
HttpClient client = vertx.createHttpClient();
client.rxRequest(HttpMethod.GET, 8080, "localhost", "/")
  .flatMap(request -> request
    .rxSend()
    .flatMap(response -> {
      if (response.statusCode() == 200) {
        return response.body();
      } else {
        return Single.error(new NoStackTraceThrowable("Invalid response"));
      }
    }))
  .subscribe(body -> {
    // Process the body
  });
----

When you need to process large streaming response, you can get a `Flowable<Buffer>` from the http response

[source,java]
----
HttpClient client = vertx.createHttpClient();
client.rxRequest(HttpMethod.GET, 8080, "localhost", "/")
  .flatMapPublisher(request -> request
    .rxSend()
    .flatMapPublisher(response -> {
      if (response.statusCode() == 200) {
        return response.toFlowable();
      } else {
        return Flowable.error(new NoStackTraceThrowable("Invalid response"));
      }
    }))
  .subscribe(chunk -> {
    // Process the response chunks
  });
----

You can also use the http://vertx.io/docs/vertx-web-client/java/#_rxjava_api[Vert.x Web Client].

=== Http server requests

The `link:../../apidocs/io/vertx/rxjava3/core/http/HttpServer.html#requestStream--[requestStream]` provides a callback for each incoming
request:

[source,java]
----
Flowable<HttpServerRequest> requestFlowable = server.requestStream().toFlowable();
requestFlowable.subscribe(request -> {
  // Process request
});
----

The `link:../../apidocs/io/vertx/core/http/HttpServerRequest.html[HttpServerRequest]` can then be adapted to an `Observable<Buffer>`:

[source,java]
----
Flowable<HttpServerRequest> requestFlowable = server.requestStream().toFlowable();
requestFlowable.subscribe(request -> {
  Observable<Buffer> observable = request.toObservable();
});
----

The `link:../../apidocs/io/vertx/rxjava3/ObservableHelper.html#unmarshaller-java.lang.Class-[ObservableHelper.unmarshaller]` can be used to parse and map
a json request to an object:

[source,java]
----
Flowable<HttpServerRequest> requestFlowable = server.requestStream().toFlowable();
requestFlowable.subscribe(request -> {
  Flowable<MyPojo> flowable = request.
    toFlowable().
    compose(FlowableHelper.unmarshaller(MyPojo.class));
});
----

=== WebSocket client

The `link:../../apidocs/io/vertx/rxjava3/core/http/HttpClient.html#webSocket-int-java.lang.String-java.lang.String-[webSocket]` provides a single callback when the WebSocket
connects, otherwise a failure:

[source,java]
----
HttpClient client = vertx.createHttpClient(new HttpClientOptions());
client.rxWebSocket(8080, "localhost", "/the_uri").subscribe(
    ws -> {
      // Use the websocket
    },
    error -> {
      // Could not connect
    }
);
----

The `link:../../apidocs/io/vertx/rxjava3/core/http/WebSocket.html[WebSocket]` can then be turned into an `Observable<Buffer>` easily:

[source,java]
----
socketObservable.subscribe(
    socket -> {
      Flowable<Buffer> dataObs = socket.toFlowable();
      dataObs.subscribe(buffer -> {
        System.out.println("Got message " + buffer.toString("UTF-8"));
      });
    }
);
----

=== WebSocket server

The `link:../../apidocs/io/vertx/rxjava3/core/http/HttpServer.html#webSocketStream--[webSocketStream]` provides a callback for each incoming
connection:

[source,java]
----
Flowable<ServerWebSocket> socketFlowable = server.webSocketStream().toFlowable();
socketFlowable.subscribe(
    socket -> System.out.println("Web socket connect"),
    failure -> System.out.println("Should never be called"),
    () -> {
      System.out.println("Subscription ended or server closed");
    }
);
----

The `link:../../apidocs/io/vertx/core/http/ServerWebSocket.html[ServerWebSocket]` can be turned into an `Observable<Buffer>` easily:

[source,java]
----
socketObservable.subscribe(
    socket -> {
      Flowable<Buffer> dataObs = socket.toFlowable();
      dataObs.subscribe(buffer -> {
        System.out.println("Got message " + buffer.toString("UTF-8"));
      });
    }
);
----