= Vert.x for RxJS
:toc: left

https://github.com/Reactive-Extensions/RxJS[RxJS] is a popular library for composing asynchronous and event
based programs using observable sequences for the JavaScript.

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

Vert.x for RxJS comes as an extension for RxJS:

[source,js]
----
var Rx = require("rx.vertx");
----

It provides the `Rx` object we need for creating `Observable`, or other kind of Rx objects.

If you are using Maven or Gradle, add the following dependency to the _dependencies_ section of your build descriptor:

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

[source,xml,subs="+attributes"]
----
<dependency>
  <groupId>{maven-groupId}</groupId>
  <artifactId>{maven-artifactId}</artifactId>
  <version>{maven-version}</version>
</dependency>
----

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

[source,groovy,subs="+attributes"]
----
compile {maven-groupId}:{maven-artifactId}:{maven-version}
----

== Read stream support

RxJS observable is a perfect match for Vert.x read streams : both provides provides a flow of items.
A read stream can be adapted to an observable with the `Rx.Observable.fromReadStream` function:

[source,js]
----
include::read_stream.js[tags=example]
----

== Handler support

The `rx.vertx` module provides an `observableHandler` function:

[source,js]
----
include::handler.js[tags=example]
----

Rx can also turn an existing Observer into an handler:

[source,js]
----
include::observer_to_handler.js[tags=example]
----

== Future support

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

The `rx.vertx` module provides an `observableFuture` function:

[source,js]
----
include::future.js[tags=example]
----

Rx can also turn an existing Observer into an future:

[source,js]
----
include::observer_to_future.js[tags=example]
----

== Scheduler support

RxJS relies on the default context method _timeout_ and _interval_ functions to schedule operations. The
vertx-js integration implements such functions providing an out of the box scheduler support.

== Examples

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

=== EventBus message stream

The event bus message consumer provides naturally an stream of messages:

[source,js]
----
include::event_bus_messages.js[tags=example]
----

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

[source,js]
----
include::event_bus_bodies.js[tags=example]
----

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

[source,js]
----
include::event_bus_map_reduce.js[tags=example]
----

=== Timers

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

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

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

=== Http client requests

The HttpClientRequest provides a one shot callback with the
`http.HttpClientResponse` object. The observable reports a request failure.

[source,js]
----
include::http_client_request.js[tags=example]
----

The response can be processed as an stream of buffer:

[source,js]
----
include::http_client_response.js[tags=example]
----

=== Http server requests

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

[source,js]
----
include::http_server_request.js[tags=example]
----

The `HttpServerRequest` can then be adapted to a buffer observable:

[source,js]
----
include::http_server_request_observable.js[tags=example]
----

=== Websocket client

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

[source,js]
----
include::websocket_client.js[tags=example]
----

The `WebSocket` can then be turned into an observable of buffer easily

[source,js]
----
include::websocket_client_buffer.js[tags=example]
----

=== Websocket server

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

[source,js]
----
include::websocket_server.js[tags=example]
----

The `ServerWebSocket` can be turned into a buffer observable easily:

[source,js]
----
include::websocket_server_buffer.js[tags=example]
----
