This pull request generates suspending methods of the same name for methods that has kind METHOD_FUTURE.

I will add some tests and clean up the template file later.

I am new to pretty much everything in this pull request, please bear with me if I made some stupid mistakes.

Current Kotlin suspend generated method are named with an Await suffix:

vertx.deployVerticle(verticle, Handler { ... }) 

// becomes

val id = vertx.deployVerticleAwait(verticle);

this issue gathers proposal and opinions about the best naming for 3.6

See minimal reproducer project:

https://github.com/lukehutch/kotlinproj/tree/master

[ERROR] Failed to execute goal org.jetbrains.kotlin:kotlin-maven-plugin:1.3.21:compile (compile) on project kotlinproj: Compilation failure: Compilation failure: 
[ERROR] /tmp/kotlin-proj/src/main/java/kotlinproj/kotlinproj/Test.kt:[18,37] Unresolved reference: await
[ERROR] /tmp/kotlin-proj/src/main/java/kotlinproj/kotlinproj/Test.kt:[21,72] Unresolved reference: await

This is with 4.0-SNAPSHOT versions.

I used the same test structure as in the web-client library.

I didn't create the extension on HttpRequest<T> because the Kotlin compiler is not smart enough to understand that the first type is already available in the receiver, making necessary to do the call like requestWitBufferResponse.like<Buffer, ToClass>().send(..)

• Use an internal channel instead of a ReadStream to implement the ReceiveChannelHandler.
• Removed usage of internal ArrayChannel.

In progress @vietj:

• Unfortunately, I'm not able to figure it out what is the problem with the HttpParserTest, if I change line 68 from chunk.length() - 2 to chunk.length() the test actually pass, but I don't understand why. I'd appreciate if you could help me there.
• I think that the change I made in the ReadStream<T>.toChannel resumes the stream immediately after a receive/receiveOrNull/poll is successful, but I'm not sure this is what you wanted.

• CoroutineVerticle implements now CoroutineScope based on https://github.com/Kotlin/kotlinx.coroutines/issues/410
• Used GlobalScope.launch over launch when a verticle doesn't exist.
• Renamed tests for improving readability.
• Reverted to coroutines 0.30.0 to avoid warnings from atomicfu-common:0.11.10 that uses kotlin-stdlib-common:1.3.0-rc-131.

Next steps:

• Adjust documentation for coroutines module. (how should I update the examples in index.adoc?)
• Remove deprecated ArrayChannel in ReceiveChannelHandler.
• Make project depend on Kotlin 1.3.x and support the latest coroutines version.

This is a basic coroutine support for Kotlin using only core dependencies (no kotlinx.coroutine).

I am not Vert.x expert, so any suggestion is well accepted.

Please send me your feedback.

3.5.1-SNAPSHOT will ship with version 1.2.20 of kotlin. see #44 Latest version as of today is 1.2.21.

It would be great if we were able to specify the kotlin version in our projects. Right now, I am excluding kotlin dependencies from vertx-lang-kotlin:

compile(group = "io.vertx", name = "vertx-lang-kotlin", version = vertx_version) { exclude(group = "org.jetbrains.kotlin") }

Will this lead to incompatibilities, or is this is a reasonable way to keep kotlin version up to date?

Also upgraded the Kotlin coroutines library.

ReceiveChannelHandler had to be added a field and a method due to additions in ReceiveChannel. These are internal APIs so I made them throw NotImplementedError (through TODO("...")) and all tests still pass so it doesn't seem to be on a typical call path.

Kotliners (Kotlin developers) using Vert.x don't have a clear way to do testing. The new testing guide provides guidance on how to test Vert.x code that is Kotlinic (idiomatic Kotlin). Currently the guide is a draft version and hasn't been converted to AsciiDoc. More important to ensure that the approach for the guide is correct before getting the formatting right.

Version

• vertx 3.9.7
• kotlin 1.5.0 (tested also on 1.4.32)
• open-jdk-14

Context

I observed a resource leak in JDBCClient, namely in the vertx-lang-kotlin:3.9.7 function

suspend fun SQLClient.getConnectionAwait(): SQLConnection 

The problem is if coroutine calling getConnectionAwait() is cancelled before the handler on the following method is called:

SQLClient getConnection(Handler<AsyncResult<SQLConnection>> handler);

If such situation happens (e.g., calling REST connection is terminated / timed out), the obtained connection, passed to the handler, is not closed and returned to the pool. This causes leakage, depleting all free connections (if using with connection pool like C3P0). The program ends up in the terminal state without usable DB connection.

I believe that the problem lies in the https://github.com/vert-x3/vertx-lang-kotlin/blob/d991b1683c05c6d5081ee2f6513838c9c1dc977c/vertx-lang-kotlin-coroutines/src/main/java/io/vertx/kotlin/coroutines/VertxCoroutine.kt#L63

suspend fun <T> awaitEvent(block: (h: Handler<T>) -> Unit): T {
  return suspendCancellableCoroutine { cont: CancellableContinuation<T> ->
    try {
      block.invoke(Handler { t ->
        cont.resume(t) // <---- here, if cont is cancelled, t remains opened
      })
    } catch (e: Exception) {
      cont.resumeWithException(e)
    }
  }
}

I solved the problem by implementing the coroutine bridging on my own with a if-branch handling cancelled coroutine, such as

    try {
      block.invoke(Handler { t ->
        if (cont.isActive)
          cont.resume(t)
        } else {
          t.close()  // <--- close connection if coroutine is already cancelled when handler is invoked
        } 
      })
    } catch (e: Exception) {
      cont.resumeWithException(e)
    }

I noticed that this method is deprecated https://github.com/vert-x3/vertx-jdbc-client/issues/196 https://github.com/vert-x3/vertx-lang-kotlin/blob/c1938f890b711734d3e191d274422d3f3098726a/vertx-lang-kotlin/src/main/kotlin/io/vertx/kotlin/ext/sql/SQLClient.kt#L64-L65 but this may help others having the same issues if they didn't migrate yet.

Also, vertx 3 API is planned to be supported also in vertx 4 so this race condition can potentially affect a lot of users, making their servers unresponsive. Btw was it deprecated also because of this potential issue?

The similar issues can be present also elsewhere in the same pattern, i.e., call await method (coroutine), wait for handler, when handler is called, coroutine is already cancelled, thus resource obtained & returned in the handler is not properly closed.

Do you have a reproducer?

Not yet, I can create one if description is not enough.

Steps to reproduce

1. setup MySQL JDBC-backed demo
2. spawn 2000 worker threads simulating clients, each querying the database (code below), let clients timeout randomly
3. watch number of failures / timeouts

val rnd = Random()
withTimeout(rnd.nextLong().rem(1000L)) {
    dbClient.getConnectionAwait().use {
        it.queryAwait("SELECT 1")
    }
}

It would be great to have Mutiny support in this project for CoroutineVerticle as Mutiny is now a preferred Reactive API and has support for Coroutines. This would allow mixing of Reactive APIs with Mutiny with the Coroutines via suspend fun.

Here is some working example code to make it possible to use the io.smallrye.mutiny.coroutines.awaitSuspending with the CoroutineVerticle.

import io.vertx.core.Context
import io.vertx.kotlin.coroutines.CoroutineVerticle
import io.vertx.kotlin.coroutines.dispatcher
import io.vertx.mutiny.core.Vertx
import kotlinx.coroutines.SupervisorJob
import kotlin.coroutines.CoroutineContext

abstract class MutinyCoroutineVerticle : CoroutineVerticle() {

    protected lateinit var vertx: Vertx
    private lateinit var context: Context

    override val coroutineContext: CoroutineContext by lazy { context.dispatcher() + SupervisorJob() }

    override fun init(vertx: io.vertx.core.Vertx, context: Context) {
        this.vertx = Vertx(vertx)
        this.context = context
    }
}

Here is an example of using awaitSuspending().

import io.smallrye.mutiny.coroutines.awaitSuspending
import io.vertx.core.DeploymentOptions
import io.vertx.mutiny.core.Vertx
import kotlin.Array
import kotlin.String

suspend fun main(args: Array<String>) {
    val vertx = Vertx.vertx()
    println("Deployment Starting")
    vertx.deployVerticle({ MainVerticle() }, DeploymentOptions()).awaitSuspending()
    println("Deployment completed")
}

class MainVerticle : MutinyCoroutineVerticle() {
    private var counter = 0L

    override suspend fun start() {

        vertx.periodicStream(2000L)
            .toMulti()
            .subscribe().with { tick -> counter++ }

        vertx.createHttpServer()
                .requestHandler { req -> req.response().endAndForget("@$counter") }
                .listen(8080)
                .awaitSuspending()
   }
}

Motivation:

Hi, Vert.x Maintainers. We have been developing our projects in Kotlin with the Vert.x toolkit as the core of the backend. It's been quite smooth and this project provides a lot of useful extensions in Kotlin. During our development, we came up with a few more common extensions in our shared code repository, which we thought could be ported back to this project and made universally available.

We have tried to pick the less opinionated code from our common project in this PR so it's generally useful to most developers using Vert.x in Kotlin. If there is any code that is not appropriate to be added back to this repository, just feel free to remove it or make further changes as it suits your needs. Or if there is none to be added, just close this PR.

Thank you.

Questions

Starting a coroutine on a vert.x worker thread ends up running it inline, that is on the very same thread, instead of the event loop thread. This seems like a bug because it contradicts the KDoc on Context.dispatcher()

Version

vert.x 4.1.5, kotlin 1.5.31, coroutines 1.5.2

Context

VertxCoroutineExecutor

private class VertxCoroutineExecutor(val vertxContext: Context) : AbstractExecutorService(), ScheduledExecutorService {

  override fun execute(command: Runnable) {
    if (Vertx.currentContext() != vertxContext) {
      vertxContext.runOnContext { command.run() }
    } else {
      command.run()
    }
  }

I suspect the code might be lacking a check for isOnWorkerThread e.g. the if-statement should look like:

if (Vertx.currentContext() != vertxContext || Context.isOnWorkerThread()) {
      vertxContext.runOnContext { command.run() }

Do you have a reproducer?

            awaitBlocking {
                launch {
                    println(Thread.currentThread().name)
                }
            }

Prints

vert.x-worker-thread-0 @coroutine#23