Add support for async drivers (#3168)

* Add async codegen and runtime

Add async drivers for R2DBC and sqljs workers

* Split sync and async runtime modules

* Generate suspend functions for queries

Update async runtime to use suspend functions
Add Closeable to async runtime
Update sqljs worker and r2dbc drivers

* Add runtime module tasks to gradle plugin tests

* Run spotless

* Fix up broken async codegen tests

* Pull async gen flag from SqlDelightFile

* Add missing prop in intellij tests

* Update async mysql integration test

Add other async codegen compiler tests

* Add async runtime tests

Remove async extensions from coroutines-extensions (to revisit later)

* Fix spotless errors

* Fix broken AsyncTest

* Fix js worker test

Fix sample builds with new runtime modules

* Skip js web worker tests on node

* Add driver-async-test

Add some brief documentation on implementing async drivers

* Disable mingwX86 on driver-async-test

* Run spotless (again)

* Apply suggestions from code review

Co-authored-by: Alec Strong <AlecStrong@users.noreply.github.com>

* Remove mingwX86 from runtime-async entirely

* Clean things up

Add RuntimeTypes class for codegen types
Add default async runtime types
Remove suspend modifier from generated functions that return an `AsyncQuery`

* Make Closeable.close() suspendable

* Add missing suspend modifier

* Clean up driver-async-test to handle suspend funs

* Handle even more suspending test teardowns

* Run spotless

* Fix one last close function

* Fix up and enable remaining js worker tests

* Remove driver-async-test

Run spotless again again again

* Apply suggestions from code review

Co-authored-by: Alec Strong <AlecStrong@users.noreply.github.com>

* Clean up async runtime

Disable async codegen for HSQL
Remove unused functions in R2DBC driver
Move LogAsyncSqlDriver to tests

* Delete async runtime tests

Co-authored-by: Alec Strong <AlecStrong@users.noreply.github.com>

CONTRIBUTING.md

@@ -42,6 +42,10 @@ or join the [Jetbrains Platform Slack](https://blog.jetbrains.com/platform/2019/
 If you're interested in creating your own driver, you can do so outside of the SQLDelight repository using the `runtime` artifact. To test the driver
 you can depend on the `driver-test` and extend `DriverTest` and `TransactionTest` to ensure it works as SQLDelight would expect.
 
+#### Asynchronous Drivers
+
+Drivers that make asynchronous calls can be implemented by using the `runtime-async` artifact. 
+
 ### Gradle
 
 If you're encountering a gradle issue, start by creating a test fixture in `sqldelight-gradle-plugin/src/test` similar to the other folders there

adapters/primitive-adapters/build.gradle

@@ -12,7 +12,7 @@ kotlin {
   sourceSets {
     commonMain {
       dependencies {
-        api project(':runtime')
+        api project(':runtime:runtime')
       }
     }
     commonTest {

dialects/hsql/src/main/kotlin/app/cash/sqldelight/dialects/hsql/HsqlDialect.kt

@@ -1,5 +1,6 @@
 package app.cash.sqldelight.dialects.hsql
 
+import app.cash.sqldelight.dialect.api.RuntimeTypes
 import app.cash.sqldelight.dialect.api.SqlDelightDialect
 import app.cash.sqldelight.dialect.api.TypeResolver
 import app.cash.sqldelight.dialects.hsql.grammar.HsqlParserUtil
@@ -13,9 +14,15 @@ import com.squareup.kotlinpoet.ClassName
  * Base dialect for JDBC implementations.
  */
 class HsqlDialect : SqlDelightDialect {
-  override val driverType = ClassName("app.cash.sqldelight.driver.jdbc", "JdbcDriver")
-  override val cursorType = ClassName("app.cash.sqldelight.driver.jdbc", "JdbcCursor")
-  override val preparedStatementType = ClassName("app.cash.sqldelight.driver.jdbc", "JdbcPreparedStatement")
+  override val runtimeTypes: RuntimeTypes = RuntimeTypes(
+    ClassName("app.cash.sqldelight.driver.jdbc", "JdbcDriver"),
+    ClassName("app.cash.sqldelight.driver.jdbc", "JdbcCursor"),
+    ClassName("app.cash.sqldelight.driver.jdbc", "JdbcPreparedStatement")
+  )
+
+  override val asyncRuntimeTypes: RuntimeTypes
+    get() = throw UnsupportedOperationException("HSQL does not support an async driver")
+
   override val icon = AllIcons.Providers.Hsqldb
 
   override fun setup() {

dialects/mysql/src/main/kotlin/app/cash/sqldelight/dialects/mysql/MySqlDialect.kt

@@ -1,6 +1,7 @@
 package app.cash.sqldelight.dialects.mysql
 
 import app.cash.sqldelight.dialect.api.ConnectionManager
+import app.cash.sqldelight.dialect.api.RuntimeTypes
 import app.cash.sqldelight.dialect.api.SqlDelightDialect
 import app.cash.sqldelight.dialect.api.TypeResolver
 import app.cash.sqldelight.dialects.mysql.grammar.MySqlParserUtil
@@ -16,9 +17,18 @@ import com.squareup.kotlinpoet.ClassName
  * Base dialect for JDBC implementations.
  */
 class MySqlDialect : SqlDelightDialect {
-  override val driverType = ClassName("app.cash.sqldelight.driver.jdbc", "JdbcDriver")
-  override val cursorType = ClassName("app.cash.sqldelight.driver.jdbc", "JdbcCursor")
-  override val preparedStatementType = ClassName("app.cash.sqldelight.driver.jdbc", "JdbcPreparedStatement")
+  override val runtimeTypes: RuntimeTypes = RuntimeTypes(
+    ClassName("app.cash.sqldelight.driver.jdbc", "JdbcDriver"),
+    ClassName("app.cash.sqldelight.driver.jdbc", "JdbcCursor"),
+    ClassName("app.cash.sqldelight.driver.jdbc", "JdbcPreparedStatement")
+  )
+
+  override val asyncRuntimeTypes: RuntimeTypes = RuntimeTypes(
+    ClassName("app.cash.sqldelight.driver.r2dbc", "R2dbcDriver"),
+    ClassName("app.cash.sqldelight.driver.r2dbc", "R2dbcCursor"),
+    ClassName("app.cash.sqldelight.driver.r2dbc", "R2dbcPreparedStatement"),
+  )
+
   override val icon = AllIcons.Providers.Mysql
   override val connectionManager: ConnectionManager by lazy { MySqlConnectionManager() }
 

dialects/postgresql/src/main/kotlin/app/cash/sqldelight/dialects/postgresql/PostgreSqlDialect.kt

@@ -1,6 +1,7 @@
 package app.cash.sqldelight.dialects.postgresql
 
 import app.cash.sqldelight.dialect.api.ConnectionManager
+import app.cash.sqldelight.dialect.api.RuntimeTypes
 import app.cash.sqldelight.dialect.api.SqlDelightDialect
 import app.cash.sqldelight.dialect.api.TypeResolver
 import app.cash.sqldelight.dialects.postgresql.grammar.PostgreSqlParserUtil
@@ -15,9 +16,18 @@ import com.squareup.kotlinpoet.ClassName
  * Base dialect for JDBC implementations.
  */
 class PostgreSqlDialect : SqlDelightDialect {
-  override val driverType = ClassName("app.cash.sqldelight.driver.jdbc", "JdbcDriver")
-  override val cursorType = ClassName("app.cash.sqldelight.driver.jdbc", "JdbcCursor")
-  override val preparedStatementType = ClassName("app.cash.sqldelight.driver.jdbc", "JdbcPreparedStatement")
+  override val runtimeTypes: RuntimeTypes = RuntimeTypes(
+    ClassName("app.cash.sqldelight.driver.jdbc", "JdbcDriver"),
+    ClassName("app.cash.sqldelight.driver.jdbc", "JdbcCursor"),
+    ClassName("app.cash.sqldelight.driver.jdbc", "JdbcPreparedStatement"),
+  )
+
+  override val asyncRuntimeTypes: RuntimeTypes = RuntimeTypes(
+    ClassName("app.cash.sqldelight.driver.r2dbc", "R2dbcDriver"),
+    ClassName("app.cash.sqldelight.driver.r2dbc", "R2dbcCursor"),
+    ClassName("app.cash.sqldelight.driver.r2dbc", "R2dbcPreparedStatement"),
+  )
+
   override val allowsReferenceCycles = false
   override val icon = AllIcons.Providers.Postgresql
   override val connectionManager: ConnectionManager by lazy { PostgresConnectionManager() }

dialects/sqlite-3-18/src/main/kotlin/app/cash/sqldelight/dialects/sqlite_3_18/SqliteDialect.kt

@@ -13,16 +13,12 @@ import com.intellij.ide.plugins.PluginManagerCore
 import com.intellij.openapi.application.ApplicationManager
 import com.intellij.openapi.extensions.PluginId
 import com.intellij.psi.stubs.StubElementTypeHolderEP
-import com.squareup.kotlinpoet.ClassName
 import timber.log.Timber
 
 /**
  * A dialect for SQLite.
  */
 open class SqliteDialect : SqlDelightDialect {
-  override val driverType = ClassName("app.cash.sqldelight.db", "SqlDriver")
-  override val cursorType = ClassName("app.cash.sqldelight.db", "SqlCursor")
-  override val preparedStatementType = ClassName("app.cash.sqldelight.db", "SqlPreparedStatement")
   override val isSqlite = true
   override val icon = AllIcons.Providers.Sqlite
   override val migrationStrategy = SqliteMigrationStrategy()

drivers/android-driver/build.gradle

@@ -45,7 +45,7 @@ dependencies {
 // workaround for https://youtrack.jetbrains.com/issue/KT-27059
 configurations.all {
     resolutionStrategy.dependencySubstitution {
-        substitute module("${project.property("GROUP")}:runtime-jvm:${project.property("VERSION_NAME")}") with project(':runtime')
+        substitute module("${project.property("GROUP")}:runtime-jvm:${project.property("VERSION_NAME")}") with project(':runtime:runtime')
     }
 }
 

drivers/driver-test/build.gradle

@@ -11,7 +11,7 @@ kotlin {
   sourceSets {
     commonMain {
       dependencies {
-        api project(':runtime')
+        api project(':runtime:runtime')
 
         implementation deps.kotlin.test.common
         implementation deps.kotlin.test.commonAnnotations

drivers/jdbc-driver/build.gradle

@@ -7,7 +7,7 @@ plugins {
 archivesBaseName = 'sqldelight-jdbc-driver'
 
 dependencies {
-  api project(':runtime')
+  api project(':runtime:runtime')
 }
 
 apply from: "$rootDir/gradle/gradle-mvn-push.gradle"

drivers/native-driver/build.gradle

@@ -30,7 +30,7 @@ kotlin {
   sourceSets {
     commonMain {
       dependencies {
-        api project (':runtime')
+        api project (':runtime:runtime')
       }
     }
     commonTest {

drivers/native-driver/src/nativeTest/kotlin/com/squareup/sqldelight/drivers/native/connectionpool/BaseConcurrencyTest.kt

@@ -8,7 +8,6 @@ import co.touchlab.sqliter.DatabaseFileContext
 import co.touchlab.sqliter.JournalMode
 import co.touchlab.testhelp.concurrency.currentTimeMillis
 import co.touchlab.testhelp.concurrency.sleep
-import platform.posix.sleep
 import kotlin.native.concurrent.AtomicInt
 import kotlin.native.concurrent.Worker
 import kotlin.test.AfterTest

drivers/r2dbc-driver/build.gradle

@@ -0,0 +1,16 @@
+plugins {
+  alias(deps.plugins.kotlin.jvm)
+  alias(deps.plugins.publish)
+  alias(deps.plugins.dokka)
+}
+
+archivesBaseName = 'sqldelight-r2dbc-driver'
+
+dependencies {
+  api project(':runtime:runtime-async')
+  implementation deps.r2dbc
+  implementation deps.kotlin.coroutines.core
+  implementation deps.kotlin.coroutines.reactive
+}
+
+apply from: "$rootDir/gradle/gradle-mvn-push.gradle"

drivers/r2dbc-driver/gradle.properties

@@ -0,0 +1,4 @@
+POM_ARTIFACT_ID=r2dbc-driver
+POM_NAME=SQLDelight R2DBC Driver
+POM_DESCRIPTION=R2DBC driver for SQLDelight
+POM_PACKAGING=jar

drivers/r2dbc-driver/src/main/kotlin/app/cash/sqldelight/driver/r2dbc/R2dbcDriver.kt

@@ -0,0 +1,159 @@
+package app.cash.sqldelight.driver.r2dbc
+
+import app.cash.sqldelight.async.AsyncQuery
+import app.cash.sqldelight.async.AsyncTransacter
+import app.cash.sqldelight.async.db.AsyncSqlCursor
+import app.cash.sqldelight.async.db.AsyncSqlDriver
+import app.cash.sqldelight.async.db.AsyncSqlPreparedStatement
+import io.r2dbc.spi.Connection
+import io.r2dbc.spi.Statement
+import kotlinx.coroutines.reactive.awaitLast
+import kotlinx.coroutines.reactive.awaitSingle
+
+class R2dbcDriver(private val connection: Connection) : AsyncSqlDriver {
+  override suspend fun <R> executeQuery(
+    identifier: Int?,
+    sql: String,
+    mapper: (AsyncSqlCursor) -> R,
+    parameters: Int,
+    binders: (AsyncSqlPreparedStatement.() -> Unit)?
+  ): R {
+    val prepared = connection.createStatement(sql).also { statement ->
+      R2dbcPreparedStatement(statement).apply { if (binders != null) this.binders() }
+    }
+    val result = prepared.execute().awaitSingle()
+
+    val rowSet = mutableListOf<Map<Int, Any?>>()
+    result.map { row, rowMetadata ->
+      rowSet.add(rowMetadata.columnMetadatas.mapIndexed { index, _ -> index to row.get(index) }.toMap())
+    }.awaitLast()
+
+    return mapper(R2dbcCursor(rowSet))
+  }
+
+  override suspend fun execute(
+    identifier: Int?,
+    sql: String,
+    parameters: Int,
+    binders: (AsyncSqlPreparedStatement.() -> Unit)?
+  ): Long {
+    val prepared = connection.createStatement(sql).also { statement ->
+      R2dbcPreparedStatement(statement).apply { if (binders != null) this.binders() }
+    }
+
+    val result = prepared.execute().awaitSingle()
+    return result.rowsUpdated.awaitSingle()
+  }
+
+  private val transactions = ThreadLocal<Transaction>()
+  private var transaction: Transaction?
+    get() = transactions.get()
+    set(value) {
+      transactions.set(value)
+    }
+
+  override suspend fun newTransaction(): AsyncTransacter.Transaction {
+    val enclosing = transaction
+    val transaction = Transaction(enclosing, connection)
+    connection.beginTransaction().awaitSingle()
+
+    return transaction
+  }
+
+  override fun currentTransaction(): AsyncTransacter.Transaction? = transaction
+
+  override fun addListener(listener: AsyncQuery.Listener, queryKeys: Array<String>) = Unit
+  override fun removeListener(listener: AsyncQuery.Listener, queryKeys: Array<String>) = Unit
+  override fun notifyListeners(queryKeys: Array<String>) = Unit
+
+  override suspend fun close() {
+    connection.close().awaitSingle()
+  }
+
+  private class Transaction(
+    override val enclosingTransaction: AsyncTransacter.Transaction?,
+    private val connection: Connection
+  ) : AsyncTransacter.Transaction() {
+    override suspend fun endTransaction(successful: Boolean) {
+      if (enclosingTransaction == null) {
+        if (successful) connection.commitTransaction().awaitSingle()
+        else connection.rollbackTransaction().awaitSingle()
+      }
+    }
+  }
+}
+
+open class R2dbcPreparedStatement(private val statement: Statement) : AsyncSqlPreparedStatement {
+  override fun bindBytes(index: Int, bytes: ByteArray?) {
+    if (bytes == null) {
+      statement.bindNull(index - 1, ByteArray::class.java)
+    } else {
+      statement.bind(index - 1, bytes)
+    }
+  }
+
+  override fun bindLong(index: Int, long: Long?) {
+    if (long == null) {
+      statement.bindNull(index - 1, Long::class.java)
+    } else {
+      statement.bind(index - 1, long)
+    }
+  }
+
+  override fun bindDouble(index: Int, double: Double?) {
+    if (double == null) {
+      statement.bindNull(index - 1, Double::class.java)
+    } else {
+      statement.bind(index - 1, double)
+    }
+  }
+
+  override fun bindString(index: Int, string: String?) {
+    if (string == null) {
+      statement.bindNull(index - 1, String::class.java)
+    } else {
+      statement.bind(index - 1, string)
+    }
+  }
+
+  override fun bindBoolean(index: Int, boolean: Boolean?) {
+    if (boolean == null) {
+      statement.bindNull(index - 1, Boolean::class.java)
+    } else {
+      statement.bind(index - 1, boolean)
+    }
+  }
+
+  fun bindObject(index: Int, any: Any?) {
+    if (any == null) {
+      statement.bindNull(index - 1, Any::class.java)
+    } else {
+      statement.bind(index - 1, any)
+    }
+  }
+}
+
+/**
+ * TODO: Write a better async cursor API
+ */
+open class R2dbcCursor(val rowSet: List<Map<Int, Any?>>) : AsyncSqlCursor {
+  var row = -1
+    private set
+
+  override fun next(): Boolean = ++row < rowSet.size
+
+  override fun getString(index: Int): String? = rowSet[row][index] as String?
+
+  override fun getLong(index: Int): Long? = (rowSet[row][index] as Number?)?.toLong()
+
+  override fun getBytes(index: Int): ByteArray? = rowSet[row][index] as ByteArray?
+
+  override fun getDouble(index: Int): Double? = rowSet[row][index] as Double?
+
+  override fun getBoolean(index: Int): Boolean? = rowSet[row][index] as Boolean?
+
+  inline fun <reified T : Any> getObject(index: Int): T? = rowSet[row][index] as T?
+
+  @Suppress("UNCHECKED_CAST")
+  fun <T> getArray(index: Int): Array<T>? = rowSet[row][index] as Array<T>?
+}

drivers/sqljs-driver/build.gradle

@@ -24,12 +24,15 @@ kotlin {
   }
 
   sourceSets["main"].dependencies {
-    api project(':runtime')
+    api project(':runtime:runtime')
+    api project(':runtime:runtime-async')
+    implementation deps.kotlin.coroutines.core
   }
   sourceSets["test"].dependencies {
     implementation deps.kotlin.test.js
     implementation npm('sql.js', deps.versions.sqljs.get())
     implementation devNpm("copy-webpack-plugin", "9.1.0")
+    implementation deps.kotlin.coroutines.test
   }
 }
 

drivers/sqljs-driver/karma.config.d/wasm.js

@@ -1,12 +1,21 @@
 const path = require("path");
-const abs = path.resolve("../../node_modules/sql.js/dist/sql-wasm.wasm")
+const dist = path.resolve("../../node_modules/sql.js/dist/")
+const wasm = path.join(dist, "sql-wasm.wasm")
+const worker = path.join(dist, "worker.sql-wasm.js")
 
 config.files.push({
-    pattern: abs,
+    pattern: wasm,
+    served: true,
+    watched: false,
+    included: false,
+    nocache: false,
+}, {
+    pattern: worker,
     served: true,
     watched: false,
     included: false,
     nocache: false,
 });
 
-config.proxies["/sql-wasm.wasm"] = `/absolute${abs}`
+config.proxies["/sql-wasm.wasm"] = `/absolute${wasm}`
+config.proxies["/worker.sql-wasm.js"] = `/absolute${worker}`

drivers/sqljs-driver/src/main/kotlin/app/cash/sqldelight/driver/sqljs/JsSqlDriver.kt

@@ -132,7 +132,7 @@ private class JsSqlCursor(private val statement: Statement) : SqlCursor {
   fun close() { statement.freemem() }
 }
 
-private class JsSqlPreparedStatement : SqlPreparedStatement {
+internal class JsSqlPreparedStatement : SqlPreparedStatement {
 
   val parameters = mutableListOf<Any?>()