fix connection leaking on update and updateAndReturnGeneratedKey

make extractGeneratedKey return Future and thus removing blocking Await.result from mysql implementation
remove some duplication and unnecessary use of EC

Author: jozic

build.sbt

@@ -1,4 +1,4 @@
-lazy val _version = "0.6.0"
+lazy val _version = "0.6.1-SNAPSHOT"
 lazy val scalikejdbcVersion = "2.4.2"
 lazy val mauricioVersion = "0.2.20" // provided
 lazy val postgresqlVersion = "9.4-1201-jdbc41"

core/src/main/scala/scalikejdbc/async/AsyncConnectionPool.scala

@@ -21,23 +21,8 @@ import scalikejdbc._
  * Asynchronous Connection Pool
  */
 abstract class AsyncConnectionPool(
-    val url: String,
-    val user: String,
-    password: String,
     val settings: AsyncConnectionPoolSettings = AsyncConnectionPoolSettings()) {
 
-  type MauricioConfiguration = com.github.mauricio.async.db.Configuration
-
-  private[this] val jdbcUrl = JDBCUrl(url)
-
-  protected val config = new MauricioConfiguration(
-    username = user,
-    host = jdbcUrl.host,
-    port = jdbcUrl.port,
-    password = Option(password).filterNot(_.trim.isEmpty),
-    database = Option(jdbcUrl.database).filterNot(_.trim.isEmpty)
-  )
-
   /**
    * Borrows a connection from pool.
    * @return connection
@@ -70,24 +55,23 @@ object AsyncConnectionPool extends LogSupport {
 
   private[this] val pools = new ConcurrentMap[Any, AsyncConnectionPool]()
 
-  private[this] def ensureInitialized(name: Any): Unit = {
-    if (!isInitialized(name)) {
+  def isInitialized(name: Any = DEFAULT_NAME): Boolean = pools.contains(name)
+
+  def get(name: Any = DEFAULT_NAME): AsyncConnectionPool = {
+    pools.getOrElse(name, {
       val message = ErrorMessage.CONNECTION_POOL_IS_NOT_YET_INITIALIZED + "(name:" + name + ")"
       throw new IllegalStateException(message)
-    }
+    })
   }
 
-  def isInitialized(name: Any = DEFAULT_NAME) = pools.get(name).isDefined
-
-  def get(name: Any = DEFAULT_NAME): AsyncConnectionPool = pools.get(name).orNull
-
   def apply(name: Any = DEFAULT_NAME): AsyncConnectionPool = get(name)
 
   def add(name: Any, url: String, user: String, password: String, settings: CPSettings = AsyncConnectionPoolSettings())(
     implicit factory: CPFactory = AsyncConnectionPoolFactory): Unit = {
     val newPool: AsyncConnectionPool = factory.apply(url, user, password, settings)
     log.debug(s"Registered connection pool (url: ${url}, user: ${user}, settings: ${settings}")
-    pools.put(name, newPool)
+    val replaced = pools.put(name, newPool)
+    replaced.foreach(_.close())
   }
 
   def singleton(url: String, user: String, password: String, settings: CPSettings = AsyncConnectionPoolSettings())(
@@ -96,12 +80,17 @@ object AsyncConnectionPool extends LogSupport {
   }
 
   def borrow(name: Any = DEFAULT_NAME): AsyncConnection = {
-    ensureInitialized(name)
     val pool = get(name)
-    log.debug("Borrowed a new connection from " + pool.toString())
+    log.debug(s"Borrowed a new connection from pool $name")
     pool.borrow()
   }
 
+  def giveBack(connection: NonSharedAsyncConnection, name: Any = DEFAULT_NAME): Unit = {
+    val pool = get(name)
+    log.debug(s"Gave back previously borrowed connection from pool $name")
+    pool.giveBack(connection)
+  }
+
   def close(name: Any = DEFAULT_NAME): Unit = pools.remove(name).foreach(_.close())
 
   def closeAll(): Unit = pools.keys.foreach(name => close(name))

core/src/main/scala/scalikejdbc/async/AsyncDB.scala

@@ -15,47 +15,7 @@
  */
 package scalikejdbc.async
 
-import scala.concurrent._
-import scala.util.{ Failure, Success }
-import scalikejdbc.async.ShortenedNames._
-import scalikejdbc.async.internal.AsyncConnectionCommonImpl
-
 /**
  * Basic Database Accessor
  */
-object AsyncDB {
-
-  /**
-   * Provides a code block which have a connection from ConnectionPool and passes it to the operation.
-   *
-   * @param op operation
-   * @tparam A return type
-   * @return a future value
-   */
-  def withPool[A](op: (SharedAsyncDBSession) => Future[A]): Future[A] = {
-    op.apply(sharedSession)
-  }
-
-  /**
-   * Provides a shared session.
-   *
-   * @return shared session
-   */
-  def sharedSession: SharedAsyncDBSession = SharedAsyncDBSession(AsyncConnectionPool().borrow())
-
-  /**
-   * Provides a future world within a transaction.
-   *
-   * @param op operation
-   * @param cxt execution context
-   * @tparam A return type
-   * @return a future value
-   */
-  def localTx[A](op: (TxAsyncDBSession) => Future[A])(implicit cxt: EC = ECGlobal): Future[A] = {
-    AsyncConnectionPool().borrow().toNonSharedConnection()
-      .map { nonSharedConnection => TxAsyncDBSession(nonSharedConnection) }
-      .flatMap { tx => AsyncTx.inTransaction[A](tx, op) }
-  }
-
-}
-
+object AsyncDB extends NamedAsyncDB()

core/src/main/scala/scalikejdbc/async/AsyncDBSession.scala

@@ -33,7 +33,7 @@ trait AsyncDBSession extends LogSupport {
     withListeners(statement, _parameters) {
       queryLogging(statement, _parameters)
       connection.sendPreparedStatement(statement, _parameters: _*).map { result =>
-        result.rowsAffected.map(_ > 0).getOrElse(false)
+        result.rowsAffected.exists(_ > 0)
       }
     }
   }
@@ -44,8 +44,8 @@ trait AsyncDBSession extends LogSupport {
       queryLogging(statement, _parameters)
       if (connection.isShared) {
         // create local transaction because postgresql-async 0.2.4 seems not to be stable with PostgreSQL without transaction
-        connection.toNonSharedConnection().map(c => TxAsyncDBSession(c)).flatMap { tx: TxAsyncDBSession =>
-          tx.update(statement, _parameters: _*)
+        connection.toNonSharedConnection().flatMap { conn =>
+          AsyncTx.inTransaction(TxAsyncDBSession(conn), (tx: TxAsyncDBSession) => tx.update(statement, _parameters: _*))
         }
       } else {
         connection.sendPreparedStatement(statement, _parameters: _*).map { result =>
@@ -60,11 +60,12 @@ trait AsyncDBSession extends LogSupport {
     withListeners(statement, _parameters) {
       queryLogging(statement, _parameters)
       connection.toNonSharedConnection().flatMap { conn =>
-        conn.sendPreparedStatement(statement, _parameters: _*).map { result =>
-          result.generatedKey.getOrElse {
-            throw new IllegalArgumentException(ErrorMessage.FAILED_TO_RETRIEVE_GENERATED_KEY + " SQL: '" + statement + "'")
-          }
-        }
+        AsyncTx.inTransaction(TxAsyncDBSession(conn), (tx: TxAsyncDBSession) =>
+          tx.connection.sendPreparedStatement(statement, _parameters: _*).flatMap { result =>
+            result.generatedKey.map(_.getOrElse {
+              throw new IllegalArgumentException(ErrorMessage.FAILED_TO_RETRIEVE_GENERATED_KEY + " SQL: '" + statement + "'")
+            })
+          })
       }
     }
   }
@@ -88,7 +89,7 @@ trait AsyncDBSession extends LogSupport {
       results match {
         case Nil => None
         case one :: Nil => Option(one)
-        case _ => throw new TooManyRowsException(1, results.size)
+        case _ => throw TooManyRowsException(1, results.size)
       }
     }
   }
@@ -106,10 +107,9 @@ trait AsyncDBSession extends LogSupport {
 
       def processResultSet(oneToOne: (LinkedHashMap[A, Option[B]]), rs: WrappedResultSet): LinkedHashMap[A, Option[B]] = {
         val o = extractOne(rs)
-        oneToOne.keys.find(_ == o).map {
-          case Some(found) => throw new IllegalRelationshipException(ErrorMessage.INVALID_ONE_TO_ONE_RELATION)
-        }.getOrElse {
-          oneToOne += (o -> extractTo(rs))
+        oneToOne.keys.find(_ == o) match {
+          case Some(_) => throw IllegalRelationshipException(ErrorMessage.INVALID_ONE_TO_ONE_RELATION)
+          case _ => oneToOne += (o -> extractTo(rs))
         }
       }
       connection.sendPreparedStatement(statement, _parameters: _*).map { result =>
@@ -338,7 +338,7 @@ trait AsyncDBSession extends LogSupport {
   protected def ensureAndNormalizeParameters(parameters: Seq[Any]): Seq[Any] = {
     parameters.map {
       case withValue: ParameterBinderWithValue[_] => withValue.value
-      case binder: ParameterBinder => throw new IllegalArgumentException("ParameterBinder is unsupported")
+      case _: ParameterBinder => throw new IllegalArgumentException("ParameterBinder is unsupported")
       case rawValue => rawValue
     }
   }

core/src/main/scala/scalikejdbc/async/AsyncQueryResult.scala

@@ -15,6 +15,8 @@
  */
 package scalikejdbc.async
 
+import scala.concurrent.Future
+
 /**
  * Query Result
  */
@@ -23,7 +25,7 @@ abstract class AsyncQueryResult(
     val statusMessage: Option[String],
     val rows: Option[AsyncResultSet]) {
 
-  val generatedKey: Option[Long]
+  val generatedKey: Future[Option[Long]]
 
 }
 

core/src/main/scala/scalikejdbc/async/AsyncTxQuery.scala

@@ -16,9 +16,7 @@
 package scalikejdbc.async
 
 import scalikejdbc._
-import scalikejdbc.async.internal.AsyncConnectionCommonImpl
 import scala.concurrent._
-import scala.util._
 import scalikejdbc.async.ShortenedNames._
 
 /**
@@ -36,8 +34,7 @@ class AsyncTxQuery(sqls: Seq[SQL[_, _]]) {
       }
     }
     session.connection.toNonSharedConnection
-      .map(conn => TxAsyncDBSession(conn))
-      .flatMap { tx => AsyncTx.inTransaction(tx, op) }
+      .flatMap(conn => AsyncTx.inTransaction(TxAsyncDBSession(conn), op))
   }
 
 }

core/src/main/scala/scalikejdbc/async/NamedAsyncDB.scala

@@ -15,10 +15,7 @@
  */
 package scalikejdbc.async
 
-import scalikejdbc.async.internal.AsyncConnectionCommonImpl
-
 import scala.concurrent._
-import scala.util.{ Failure, Success }
 import scalikejdbc.async.ShortenedNames._
 
 /**
@@ -29,12 +26,12 @@ case class NamedAsyncDB(name: Any = AsyncConnectionPool.DEFAULT_NAME) {
   /**
    * Provides a code block which have a connection from ConnectionPool and passes it to the operation.
    *
-   * @param f operation
+   * @param op operation
    * @tparam A return type
    * @return a Future value
    */
-  def withPool[A](f: (SharedAsyncDBSession) => Future[A]): Future[A] = {
-    f.apply(sharedSession)
+  def withPool[A](op: (SharedAsyncDBSession) => Future[A]): Future[A] = {
+    op.apply(sharedSession)
   }
 
   /**
@@ -54,9 +51,7 @@ case class NamedAsyncDB(name: Any = AsyncConnectionPool.DEFAULT_NAME) {
    */
   def localTx[A](op: (TxAsyncDBSession) => Future[A])(implicit cxt: EC = ECGlobal): Future[A] = {
     AsyncConnectionPool(name).borrow().toNonSharedConnection()
-      .map { txConn => TxAsyncDBSession(txConn) }
-      .flatMap { tx => AsyncTx.inTransaction[A](tx, op) }
+      .flatMap(conn => AsyncTx.inTransaction[A](TxAsyncDBSession(conn), op))
   }
 
 }
-

core/src/main/scala/scalikejdbc/async/internal/AsyncConnectionCommonImpl.scala

@@ -68,6 +68,12 @@ private[scalikejdbc] trait AsyncConnectionCommonImpl extends AsyncConnection {
    * @param cxt  execution context
    * @return optional generated key
    */
-  protected def extractGeneratedKey(queryResult: QueryResult)(implicit cxt: EC = ECGlobal): Option[Long]
+  protected def extractGeneratedKey(queryResult: QueryResult)(implicit cxt: EC = ECGlobal): Future[Option[Long]]
+
+  protected def ensureNonShared(): Unit = {
+    if (!this.isInstanceOf[NonSharedAsyncConnection]) {
+      throw new IllegalStateException("This asynchronous connection must be a non-shared connection.")
+    }
+  }
 
 }

core/src/main/scala/scalikejdbc/async/internal/AsyncConnectionPoolCommonImpl.scala

@@ -0,0 +1,31 @@
+package scalikejdbc.async.internal
+
+import com.github.mauricio.async.db.{Configuration, Connection}
+import com.github.mauricio.async.db.pool.{ConnectionPool, ObjectFactory, PoolConfiguration}
+import scalikejdbc.LogSupport
+import scalikejdbc.async.{AsyncConnectionPool, AsyncConnectionPoolSettings, NonSharedAsyncConnection}
+
+abstract class AsyncConnectionPoolCommonImpl[T <: Connection](
+  url: String,
+  user: String,
+  password: String,
+  factoryF: Configuration => ObjectFactory[T],
+  settings: AsyncConnectionPoolSettings = AsyncConnectionPoolSettings()
+) extends AsyncConnectionPool(settings) with MauricioConfiguration with LogSupport {
+
+  private[this] val factory = factoryF(configuration(url, user, password))
+  private[internal] val pool = new ConnectionPool[T](
+    factory = factory,
+    configuration = PoolConfiguration(
+      maxObjects = settings.maxPoolSize,
+      maxIdle = settings.maxIdleMillis,
+      maxQueueSize = settings.maxQueueSize)
+  )
+
+  override def close(): Unit = pool.disconnect
+
+  override def giveBack(conn: NonSharedAsyncConnection): Unit = conn match {
+    case conn: NonSharedAsyncConnectionImpl => pool.giveBack(conn.underlying.asInstanceOf[T])
+    case _ => log.debug("You don't need to give back this connection to the pool.")
+  }
+}

core/src/main/scala/scalikejdbc/async/internal/MauricioConfiguration.scala

@@ -15,26 +15,22 @@
  */
 package scalikejdbc.async.internal
 
+import com.github.mauricio.async.db.Configuration
 import scalikejdbc.JDBCUrl
-import scalikejdbc.async._
 
 /**
  * Configuration attribute
  */
-private[scalikejdbc] trait MauricioConfiguration { self: AsyncConnection =>
+private[scalikejdbc] trait MauricioConfiguration {
 
-  val url: String
-  val user: String
-  val password: String
-
-  private[scalikejdbc] val configuration = {
+  private[scalikejdbc] def configuration(url: String, user: String, password: String) = {
     val jdbcUrl = JDBCUrl(url)
-    com.github.mauricio.async.db.Configuration(
+    Configuration(
+      username = user,
       host = jdbcUrl.host,
       port = jdbcUrl.port,
-      database = Option(jdbcUrl.database),
-      username = user,
-      password = Option(password)
+      password = Option(password).filterNot(_.trim.isEmpty),
+      database = Option(jdbcUrl.database).filterNot(_.trim.isEmpty)
     )
   }