= Vert.x for RxGroovy
:toc: left

https://github.com/ReactiveX/RxJava[RxJava] is a popular library for composing asynchronous and event
based programs using observable sequences for the Java VM. https://github.com/ReactiveX/RxGroovy[RxGroovy]
is the Reactive Extensions for Groovy. This adaptor allows `groovy.lang.Closure` functions to be used
and RxJava will know how to invoke them.

Vert.x integrates naturally with RxGroovy, allowing to use observable wherever you can use streams or asynchronous results.

== Read stream support

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

Vert.x API for Groovy provides `io.vertx.groovy.core.stream.ReadStream` objects, the RxGroovy provides a
Groovy extension module that adds the `toObservable` method to the read stream class.

[source,groovy]
----
include::readStream.groovy[tags=example]
----

== Handler support

The RxJava `io.vertx.ext.rx.java.RxHelper` should be used to:
- create an `io.vertx.ext.rx.java.ObservableHandler`,
- transform actions to an handler

The RxGroovy extension module adds the `toHandler` method on the `rx.Observer` class:

[source,groovy]
----
include::toHandler.groovy[tags=example]
----

== Async result support

In Vert.x future objects are modelled as async result handlers and occur as last parameter of asynchronous methods.

The RxGroovy extension module adds the `toFuture` method on the `rx.Observer` class:

[source,groovy]
----
include::toFuture.groovy[tags=example]
----

== Scheduler support

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

When an RxJava method deals with a scheduler, it accepts an overloaded method accepting an extra `Rx.Scheduler`,
the RxGroovy extension module adds to the `Vertx` class the `scheduler()` method will return a scheduler that can be used in such places.

[source,groovy]
----
include::scheduler.groovy[tags=example]
----

For blocking scheduled actions, a scheduler can be created with the `blockingScheduler` method:

[source,groovy]
----
include::blockingScheduler.groovy[tags=example]
----

RxJava can also be configured to use a scheduler by default, the returned scheduler hook uses a blocking
scheduler for IO actions:

[source,groovy]
----
include::defaultScheduler.groovy[tags=example]
----

== Json unmashalling

The `io.vertx.rx.groovy.RxHelper#unmarshaller(java.lang.Class)}` creates an `rx.Observable.Operator` that
transforms an `Observable<Buffer>` in json format into an object observable:

[source,groovy]
----
include::unmarshaller.groovy[tags=example]
----

== Api examples

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

=== EventBus message stream

The event bus `MessageConsumer` provides naturally an `Observable<Message<T>>`:

[source,groovy]
----
include::eventBusMessages.groovy[tags=example]
----

The `MessageConsumer` provides a stream of `Message`.
The `Message#body()` gives access to a new stream of message bodies if needed:

[source,groovy]
----
include::eventBusBodies.groovy[tags=example]
----

RxJava map/reduce composition style can be then be used:

[source,groovy]
----
include::eventBusMapReduce.groovy[tags=example]
----

=== Timers

Timer task can be created with `Vertx#timerStream(long)`:

[source,groovy]
----
include::timer.groovy[tags=example]
----

Periodic task can be created with `Vertx#periodicStream(long)`:

[source,groovy]
----
include::periodic.groovy[tags=example]
----

The observable can be cancelled with an unsubscription:

[source,groovy]
----
include::periodicUnsubscribe.groovy[tags=example]
----

=== Http client requests

`HttpClientRequest#toObservable()` provides a one shot callback with the `HttpClientResponse`}
 object. The observable reports a request failure.

[source,groovy]
----
include::httpClientRequest.groovy[tags=example]
----

 The response can be processed as an `Observable<Buffer>` with the
`HttpClientResponse#toObservable()` method:

[source,groovy]
----
include::httpClientResponse.groovy[tags=example]
----

The same flow can be achieved with the `flatMap` operation:

[source,groovy]
----
include::httpClientResponseFlatMap.groovy[tags=example]
----

We can also unmarshall the `Observable<Buffer>` into an object using the {@link io.vertx.rx.groovy.RxHelpe.RxHelper#unmarshaller(java.lang.Class)}
static method. This method creates an `Rx.Observable.Operator` unmarshalling buffers to an object:

[source,groovy]
----
include::httpClientResponseFlatMapUnmarshall.groovy[tags=example]
----

=== Http server requests

The `HttpServer#requestStream()` provides a callback for each incoming request:

[source,groovy]
----
include::httpServerRequest.groovy[tags=example]
----

The `HttpServerRequest` can then be adapted to an `Observable<Buffer>`:

[source,groovy]
----
include::httpServerRequestObservable.groovy[tags=example]
----

The `io.vertx.rx.groovy.RxHelpe.RxHelper#unmarshaller(java.lang.Class)}` can be used to parse and map a json request to an object:

[source,groovy]
----
include::httpServerRequestObservableUnmarshall.groovy[tags=example]
----

=== Websocket client

The`HttpClient#websocketStream`} provides a single callback when the websocket connects, otherwise a failure:

[source,groovy]
----
include::websocketClient.groovy[tags=example]
----

The `WebSocket` can then be turned into an `Observable<Buffer>` easily

[source,groovy]
----
include::websocketClientBuffer.groovy[tags=example]
----

=== Websocket server

The `HttpServer#websocketStream()` provides a callback for each incoming connection:

[source,groovy]
----
include::websocketServer.groovy[tags=example]
----

The `ServerWebSocket` can be turned into an `Observable<Buffer>` easily:

[source,groovy]
----
include::websocketServerBuffer.groovy[tags=example]
----
