Implement Connection Pool. Introduce PostgresDriverUnit. Only restart…

… connection on fatal error. Use K2 compiler and gradle 8.4. Tests now using shared pool. (#30)

README.md

@@ -1,17 +1,19 @@
-
 [![Kotlin Experimental](https://kotl.in/badges/experimental.svg)](https://kotlinlang.org/docs/components-stability.html)
 [![CI](https://github.com/miguel-moreira/pgkn/actions/workflows/blank.yml/badge.svg?branch=main)](https://github.com/miguel-moreira/pgkn/actions/workflows/blank.yml)
 [![GitHub license](https://img.shields.io/badge/license-Apache%20License%202.0-blue.svg?style=flat)](http://www.apache.org/licenses/LICENSE-2.0)
 [![Maven Central](https://img.shields.io/maven-central/v/io.github.moreirasantos/pgkn)](https://central.sonatype.com/artifact/io.github.moreirasantos/pgkn/)
 [![Kotlin](https://img.shields.io/badge/kotlin-1.9.0-blue.svg?logo=kotlin)](http://kotlinlang.org)
 
 # pgkn
+
 PostgreSQL Kotlin/Native Driver
 
 ## Usage
+
 ```
 implementation("io.github.moreirasantos:pgkn:1.0.0")
 ```
+
 ```kotlin
 fun main() {
     val driver = PostgresDriver(
@@ -42,14 +44,27 @@ fun main() {
     }
 }
 ```
+
 ## Features
-## Named Parameters
+
+### Connection Pool
+
+PGKN has a connection pool, its size being configurable in `PostgresDriver()` - 20 by default.  
+It will refresh connection units if the query fails fatally, but it still needs more fine-grained status checks.
+
+
+You can also use a single connection with `PostgresDriverUnit`, which currently is not `suspend`
+but probably will be in the future.
+
+### Named Parameters
+
 ```kotlin
 driver.execute(
     "select name from my_table where name = :one OR email = :other",
     mapOf("one" to "your_name", "other" to "[email protected]")
 ) { it.getString(0) }
 ```
+
 Named Parameters provides an alternative to the traditional syntax using `?` to specify parameters.
 Under the hood, it substitutes the named parameters to a query placeholder.
 

build.gradle.kts

@@ -3,17 +3,18 @@ import com.bmuschko.gradle.docker.tasks.container.DockerRemoveContainer
 import com.bmuschko.gradle.docker.tasks.container.DockerStartContainer
 import com.bmuschko.gradle.docker.tasks.image.DockerPullImage
 import io.gitlab.arturbosch.detekt.Detekt
+import org.gradle.api.tasks.testing.logging.TestExceptionFormat
 
 plugins {
-    val kotlinVersion = "1.9.20"
+    val kotlinVersion = "1.9.21"
     kotlin("multiplatform") version kotlinVersion
     id("com.bmuschko.docker-remote-api") version "9.3.7"
     id("io.gitlab.arturbosch.detekt").version("1.23.0")
     id("convention.publication")
 }
 
 group = "io.github.moreirasantos"
-version = "1.0.2"
+version = "1.1.0"
 
 repositories {
     mavenCentral()
@@ -45,6 +46,7 @@ kotlin {
         val commonMain by getting {
             dependencies {
                 implementation("org.jetbrains.kotlinx:kotlinx-datetime:0.4.1")
+                implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.7.3")
                 implementation("io.github.oshai:kotlin-logging:5.1.0")
             }
         }
@@ -99,6 +101,15 @@ tasks {
     }
 }
 
+tasks.withType<Test> {
+    testLogging {
+        events("PASSED", "FAILED", "SKIPPED")
+        exceptionFormat = TestExceptionFormat.FULL
+        showStandardStreams = true
+        showStackTraces = true
+    }
+}
+
 detekt {
     buildUponDefaultConfig = true // preconfigure defaults
     config.setFrom("$projectDir/config/detekt.yml")

gradle.properties

@@ -1 +1,4 @@
 kotlin.code.style=official
+kotlin.native.binary.sourceInfoType=libbacktrace
+kotlin.experimental.tryK2=true
+kapt.use.k2=true

gradle/wrapper/gradle-wrapper.properties

@@ -1,6 +1,7 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.2.1-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.4-bin.zip
 networkTimeout=10000
+validateDistributionUrl=true
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

src/commonMain/kotlin/io/github/moreirasantos/pgkn/PostgresDriver.kt

@@ -2,6 +2,7 @@ package io.github.moreirasantos.pgkn
 
 import io.github.moreirasantos.pgkn.paramsource.MapSqlParameterSource
 import io.github.moreirasantos.pgkn.paramsource.SqlParameterSource
+import io.github.moreirasantos.pgkn.pool.ConnectionPool
 import io.github.moreirasantos.pgkn.resultset.PostgresResultSet
 import io.github.moreirasantos.pgkn.resultset.ResultSet
 import io.github.moreirasantos.pgkn.sql.buildValueArray
@@ -16,20 +17,55 @@ import libpq.*
  * You can pass an [SqlParameterSource] to register your own Postgres types.
  */
 sealed interface PostgresDriver {
-    fun <T> execute(sql: String, namedParameters: Map<String, Any?> = emptyMap(), handler: (ResultSet) -> T): List<T>
+    suspend fun <T> execute(
+        sql: String,
+        namedParameters: Map<String, Any?> = emptyMap(),
+        handler: (ResultSet) -> T
+    ): List<T>
+
+    suspend fun <T> execute(sql: String, paramSource: SqlParameterSource, handler: (ResultSet) -> T): List<T>
+    suspend fun execute(sql: String, namedParameters: Map<String, Any?> = emptyMap()): Long
+    suspend fun execute(sql: String, paramSource: SqlParameterSource): Long
+}
+
+sealed interface PostgresDriverUnit {
+    fun <T> execute(
+        sql: String,
+        namedParameters: Map<String, Any?> = emptyMap(),
+        handler: (ResultSet) -> T
+    ): List<T>
+
     fun <T> execute(sql: String, paramSource: SqlParameterSource, handler: (ResultSet) -> T): List<T>
     fun execute(sql: String, namedParameters: Map<String, Any?> = emptyMap()): Long
     fun execute(sql: String, paramSource: SqlParameterSource): Long
 }
 
+@Suppress("LongParameterList")
 @OptIn(ExperimentalForeignApi::class)
 fun PostgresDriver(
     host: String,
     port: Int = 5432,
     database: String,
     user: String,
     password: String,
-): PostgresDriver = PostgresDriverImpl(
+    poolSize: Int = 20
+): PostgresDriver = PostgresDriverPool(
+    host = host,
+    port = port,
+    database = database,
+    user = user,
+    password = password,
+    poolSize = poolSize
+)
+
+@OptIn(ExperimentalForeignApi::class)
+fun PostgresDriverUnit(
+    host: String,
+    port: Int = 5432,
+    database: String,
+    user: String,
+    password: String
+): PostgresDriverUnit = PostgresDriverImpl(
     host = host,
     port = port,
     database = database,
@@ -38,15 +74,53 @@ fun PostgresDriver(
 )
 
 @ExperimentalForeignApi
-private class PostgresDriverImpl(
+private class PostgresDriverPool(
     host: String,
-    port: Int,
+    port: Int = 5432,
     database: String,
     user: String,
     password: String,
+    poolSize: Int
 ) : PostgresDriver {
 
-    private val connection = PQsetdbLogin(
+    private val pool = ConnectionPool((1..poolSize).map {
+        PostgresDriverImpl(
+            host = host,
+            port = port,
+            database = database,
+            user = user,
+            password = password,
+        )
+    })
+
+    override suspend fun <T> execute(
+        sql: String,
+        namedParameters: Map<String, Any?>,
+        handler: (ResultSet) -> T
+    ) = pool.invoke { it.execute(sql, namedParameters, handler) }
+
+    override suspend fun <T> execute(sql: String, paramSource: SqlParameterSource, handler: (ResultSet) -> T) =
+        pool.invoke { it.execute(sql, paramSource, handler) }
+
+    override suspend fun execute(sql: String, namedParameters: Map<String, Any?>) =
+        pool.invoke { it.execute(sql, namedParameters) }
+
+    override suspend fun execute(sql: String, paramSource: SqlParameterSource) =
+        pool.invoke { it.execute(sql, paramSource) }
+}
+
+@ExperimentalForeignApi
+private class PostgresDriverImpl(
+    private val host: String,
+    private val port: Int,
+    private val database: String,
+    private val user: String,
+    private val password: String,
+) : PostgresDriverUnit {
+
+    var connection = initConnection()
+
+    private fun initConnection() = PQsetdbLogin(
         pghost = host,
         pgport = port.toString(),
         dbName = database,
@@ -126,14 +200,19 @@ private class PostgresDriverImpl(
     private fun CPointer<PGresult>?.check(): CPointer<PGresult> {
         val status = PQresultStatus(this)
         check(status == PGRES_TUPLES_OK || status == PGRES_COMMAND_OK || status == PGRES_COPY_IN) {
-            connection.error()
+            val message = connection.error()
+            if (status == PGRES_FATAL_ERROR) {
+                PQfinish(connection)
+                connection = initConnection()
+            }
+            message
         }
         return this!!
     }
 }
 
 @ExperimentalForeignApi
-private fun CPointer<PGconn>?.error(): String = PQerrorMessage(this)!!.toKString().also { PQfinish(this) }
+private fun CPointer<PGconn>?.error(): String = PQerrorMessage(this)!!.toKString()
 
 private const val TEXT_RESULT_FORMAT = 0
 

src/commonMain/kotlin/io/github/moreirasantos/pgkn/pool/ConnectionPool.kt

@@ -0,0 +1,29 @@
+package io.github.moreirasantos.pgkn.pool
+
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.Semaphore
+import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.sync.withPermit
+
+/**
+ * Pool with Semaphore/Mutex pattern.
+ * Using semaphore with max permits being the size of connection pool to throttle.
+ * Mutex to make sure one connection is not used at the same time-
+ */
+internal class ConnectionPool<T>(connections: List<T>) {
+
+    private val mutex = Mutex()
+    private val semaphore = Semaphore(permits = connections.size)
+    private val connections = connections.toMutableList()
+
+    suspend operator fun <U> invoke(handler: suspend (T) -> U): U {
+        semaphore.withPermit {
+            val borrowed = mutex.withLock { connections.removeLast() }
+            try {
+                return handler(borrowed)
+            } finally {
+                mutex.withLock { connections.add(borrowed) }
+            }
+        }
+    }
+}

src/commonTest/kotlin/io/github/moreirasantos/pgkn/DriverUnitTest.kt

@@ -0,0 +1,20 @@
+package io.github.moreirasantos.pgkn
+
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+class DriverUnitTest {
+    @Test
+    fun `single driver should work`() {
+        val driver = PostgresDriverUnit(
+            host = "localhost",
+            port = 5678,
+            database = "postgres",
+            user = "postgres",
+            password = "postgres"
+        )
+        assertEquals("echo", driver
+            .execute("select 'echo'") { it.getString(0) }
+            .first())
+    }
+}