= vertx-lang-kotlin-coroutines

The `vertx-lang-kotlin-coroutines` integrates Kotlin _coroutines_ for performing asynchronous operations and processing events.
This results in using a programming model that looks like sequential code, yet it does not block kernel threads.

== Introduction

One of the key advantages of Vert.x over many legacy application platforms is that it is almost entirely non-blocking
(of kernel threads).
This allows Vert.x-based applications to handle a lot of concurrency (e.g., many connections and messages) using a very
small number of kernel threads, which in turns unlocks great scalability.

The non-blocking nature of Vert.x leads to asynchronous APIs. Asynchronous APIs can take various forms including
callbacks, promises, fibers or reactive extensions. Vert.x uses the callback style for the core APIs but it
also supports other models like RxJava.

In some cases, programming using asynchronous APIs can be more challenging than using a classic / sequential style
of code, in particular with several operations need to be done in sequence. Also, error propagation is often more
complex when using asynchronous APIs.

`vertx-lang-kotlin-coroutines` uses _coroutines_. Coroutines are very lightweight threads that do not correspond
to underlying kernel threads, so that when a _coroutine_ needs to _"block_" it gets _suspended_ and frees
its current kernel thread so that another coroutine can process events.

`vertx-lang-kotlin-coroutines` uses https://github.com/Kotlin/kotlinx.coroutines[kotlinx.coroutines] to implement the Coroutines.

NOTE: `vertx-lang-kotlin-coroutines` currently only works with Kotlin and will be out of the https://kotlinlang.org/docs/reference/coroutines.html#experimental-status-of-coroutines[experimental] status
in Kotlin 1.3

== Running a coroutine from a Vert.x context

Having imported `io.vertx.kotlin.coroutines.VertxCoroutine`, the `GlobalScope.launch` method allows to run a block of code as a
coroutine in the "Global" application scope (bounded on the lifetime of the application):

[source,kotlin,indent=0]
----
include::Example.kt[tags=launchCoroutine]
----

The `vertx.dispatcher()` returns a coroutine dispatcher that execute coroutines using the Vert.x event loop.

The `awaitEvent` function suspends the execution of the coroutine until the timer fires and resumes the coroutines
 with the value that was given to the handler.

More details are given in the next sections on handlers, events and stream of events.

== Extending CoroutineVerticle

You can deploy your code as instances of `io.vertx.kotlin.coroutines.CoroutineVerticle`, a specialized type of verticle
for Kotlin coroutines. The `CoroutineVerticle` class implements the `kotlinx.coroutines.experimental.CoroutineScope` interface,
making all coroutines builder methods bounded by default to the verticle context. You should override the suspending `start()`
and (optionally) the suspending `stop()` methods of the verticle:

[source,kotlin,indent=0]
----
include::Example.kt[tags=CoroutineVerticle]
----

All code examples below assume to be run inside a `CoroutineVerticle` implementation, but you can replace all `<builder> { .. }`
calls with `GlobalScope.<builder> { .. }` to use the application scope instead.

== Getting one-shot asynchronous results

Many asynchronous operations in Vert.x take a `Handler<AsyncResult<T>>` as the last argument.
An example would be executing an object retrieval using the Vert.x Mongo client, or sending an event bus message then
awaiting for a reply.

This is achieved by using the `awaitResult` method which returns the value or throws an exception.

The coroutine is being suspended until the event is being processed, and no kernel thread is being blocked.

The method is executed by specifying the asynchronous operation that needs to be executed in the form of a block that
is passed to the handler at run-time.

Here is an example:

[source,kotlin,indent=0]
----
include::Example.kt[tags=awaitResult]
----

When the block produces a failure, the caller can handle it as an exception using the usual exception
`try`/`catch` constructs:

[source,kotlin,indent=0]
----
include::Example.kt[tags=awaitResultFailure]
----

== Getting one-shot events

Processing a one-shot event (and not the next occurrences, if any) is achieved using the `awaitEvent` function:

[source,kotlin,indent=0]
----
include::Example.kt[tags=awaitEvent]
----

== Getting one-shot worker results

Processing a blocking computation is achieved using the `awaitBlocking` function:

[source,kotlin,indent=0]
----
include::Example.kt[tags=awaitBlocking]
----

== Streams of events

In many places in Vert.x APIs, streams of events are processed through handlers.
Examples include event bus message consumers and HTTP server requests.

The `ReceiveChannelHandler` class allows receiving events through the (suspendable) `receive` method:

[source,kotlin,indent=0]
----
include::Example.kt[tags=streamExample]
----

== Awaiting the completion of Vert.x asynchronous results

Vert.x 4 provides futures and `Future` has an `await()` suspending method latching the asynchronous result.

The `await` extension method on instances of Vert.x asynchronous results suspend coroutines until they have completed, in
which case the method returns the corresponding `AsyncResult<T>` object.

[source,kotlin,indent=0]
----
include::Example.kt[tags=awaitingFuture]
----

== Suspending extension methods

Vert.x 3 generated extension methods to simplify writing coroutines with Vert.x asynchronous APIs. It relieved
the user from using the idiomatic `awaitResult` and makes the code more natural and readable.

Since Vert.x 4 provides a future based API, such extension methods still exists for compatibility but
are _deprecated_.

[source,kotlin,indent=0]
----
include::Example.kt[tags=generatedSuspendingExtensionMethod]
----

== Channels

Channels are similar to Java `BlockingQueue` except that they do not block and instead suspend the coroutine
instead of blocking:

- sending a value to full channel suspends the coroutine
- receving a value from an empty channels also suspends the coroutine

Vert.x `ReadStream` and `WriteStream` can be adapted to channels with the `toChannel` extension method.

These adapters take care of managing the back-pressure and the stream termination

* `ReadStream<T>` is adapted to a `ReceiveChannel<T>`
* `WriteStream<T>` is adapted to a `SendChannel<T>`

=== Receiving data

Channel can be really useful when you need to handle a stream of correlated values:

[source,kotlin,indent=0]
----
include::Example.kt[tags=channel0]
----

It can also be useful for parsing protocols. We will build a non blocking HTTP request parser to show
the power of channels.

We will rely on the http://vertx.io/docs/apidocs/io/vertx/core/parsetools/RecordParser.html[`RecordParser`]
to slice the stream of buffer to a stream of buffer delimited by `\r\n`.

Here is the initial version of the parser, that handles only the HTTP request-line

[source,kotlin,indent=0]
----
include::Example.kt[tags=channel1]
----

Parsing the request-line is as simple as calling `receive` on the channel.

The next step parses HTTP headers by receiving chunks until we get an empty one

[source,kotlin,indent=0]
----
include::Example.kt[tags=channel2]
----

Finally we terminate the parser by handling optional request bodies

[source,kotlin,indent=0]
----
include::Example.kt[tags=channel3]
----

=== Sending data

Using a channel to send data is quite straightforward:

[source,kotlin,indent=0]
----
include::Example.kt[tags=sendChannel]
----

Both `SendChannel#send` and `WriteStream#write` are non blocking operations, however unlike `SendChannel#send`
can suspend the execution when the channel is full, the equivalent without a channel would look like

[source,kotlin,indent=0]
----
include::Example.kt[tags=sendChannel]
----

== Delay, cancellation and timeouts

Vertx dispatcher fully supports coroutine `delay` function via Vert.x timers:

[source,kotlin,indent=0]
----
include::Example.kt[tags=delay]
----

Timers support cancellation

[source,kotlin,indent=0]
----
include::Example.kt[tags=cancellation]
----

cancellation is https://github.com/Kotlin/kotlinx.coroutines/blob/master/coroutines-guide.md#cancellation-is-cooperative[cooperative]

You can also schedule timeout with the `withTimeout` function

[source,kotlin,indent=0]
----
include::Example.kt[tags=withTimeout]
----

== Coroutine builders

Vert.x works will all coroutine builders, as long as an instance of `CoroutineScope` is available: `launch`, `async`,
`produce', ... . A couple of important things to remember:

- The `runBlocking` doesn't need a `CoroutineScope` and must not be used from a Vert.x event loop thread.
- To avoid memory leaks, always use `coroutineScope {..}` to define a child scope. In this way, if a coroutine fails
inside the scope, all the others, defined inside that scope, will be cancelled too.

== Coroutine interoperability

Vert.x integration has been designed to be fully interoperable with Kotlin coroutines

* `kotlinx.coroutines.experimental.sync.Mutex` are executed on the event loop thread when using the vertx dispatcher

== RxJava interoperability

The module `vertx-lang-kotlin-coroutines` does not have specific integration with RxJava.
However, Kotlin coroutines provide integration with RxJava, which works out nicely with `vertx-lang-kotlin-coroutines`.

You can read more about it in https://github.com/Kotlin/kotlinx.coroutines/tree/master/reactive[Coroutines for reactive streams].
