Using option for the timeout, letting connections declare their event loops and execution contexts

Author: mauricio

db-async-common/src/main/scala/com/github/mauricio/async/db/Configuration.scala

@@ -18,7 +18,7 @@ package com.github.mauricio.async.db
 
 import java.nio.charset.Charset
 
-import io.netty.buffer.{AbstractByteBufAllocator, PooledByteBufAllocator}
+import io.netty.buffer.{ByteBufAllocator, PooledByteBufAllocator}
 import io.netty.util.CharsetUtil
 
 import scala.concurrent.duration._
@@ -43,6 +43,11 @@ object Configuration {
  * OOM or eternal loop attacks the client could have, defaults to 16 MB. You can set this
  * to any value you would like but again, make sure you know what you are doing if you do
  * change it.
+ * @param allocator the netty buffer allocator to be used
+ * @param connectTimeout the timeout for connecting to servers
+ * @param testTimeout the timeout for connection tests performed by pools
+ * @param queryTimeout the optional query timeout
+ *
  */
 
 case class Configuration(username: String,
@@ -52,7 +57,7 @@ case class Configuration(username: String,
  database: Option[String] = None,
  charset: Charset = Configuration.DefaultCharset,
  maximumMessageSize: Int = 16777216,
- allocator: AbstractByteBufAllocator = PooledByteBufAllocator.DEFAULT,
+ allocator: ByteBufAllocator = PooledByteBufAllocator.DEFAULT,
  connectTimeout: Duration = 5.seconds,
  testTimeout: Duration = 5.seconds,
- queryTimeout: Duration = Duration.Inf)
+ queryTimeout: Option[Duration] = None)

db-async-common/src/main/scala/com/github/mauricio/async/db/pool/TimeoutScheduler.scala

@@ -2,36 +2,62 @@ package com.github.mauricio.async.db.pool
 
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.{TimeUnit, TimeoutException, ScheduledFuture}
-import com.github.mauricio.async.db.util.NettyUtils
+import io.netty.channel.EventLoopGroup
 import scala.concurrent.{ExecutionContext, Promise}
 import scala.concurrent.duration.Duration
 
 trait TimeoutScheduler {
- implicit val internalPool: ExecutionContext
+
+ private var isTimeoutedBool = new AtomicBoolean(false)
+
+ /**
+ *
+ * The event loop group to be used for scheduling.
+ *
+ * @return
+ */
+
+ def eventLoopGroup : EventLoopGroup
+
+ /**
+ * Implementors should decide here what they want to do when a timeout occur
+ */
+
  def onTimeout // implementors should decide here what they want to do when a timeout occur
- private var isTimeoutedBool = new AtomicBoolean(false);
- def isTimeouted = isTimeoutedBool.get // We need this property as isClosed takes time to complete and
- // we don't want the connection to be used again.
 
- def addTimeout[A](promise: Promise[A], duration: Duration) : Option[ScheduledFuture[_]] = {
- if (duration != Duration.Inf) {
- val scheduledFuture = schedule(
+ /**
+ *
+ * We need this property as isClosed takes time to complete and
+ * we don't want the connection to be used again.
+ *
+ * @return
+ */
+
+ def isTimeouted : Boolean =
+ isTimeoutedBool.get
+
+ def addTimeout[A](
+ promise: Promise[A],
+ durationOption: Option[Duration])
+ (implicit executionContext : ExecutionContext) : Option[ScheduledFuture[_]] = {
+ durationOption.map {
+ duration =>
+ val scheduledFuture = schedule(
  {
  if (promise.tryFailure(new TimeoutException(s"Operation is timeouted after it took too long to return (${duration})"))) {
  isTimeoutedBool.set(true)
  onTimeout
  }
  },
  duration)
- promise.future.onComplete(x => scheduledFuture.cancel(false))
+  promise.future.onComplete(x => scheduledFuture.cancel(false))
 
- return Some(scheduledFuture)
+  scheduledFuture
  }
- return None
  }
 
  def schedule(block: => Unit, duration: Duration) : ScheduledFuture[_] =
- NettyUtils.DefaultEventLoopGroup.schedule(new Runnable {
+ eventLoopGroup.schedule(new Runnable {
  override def run(): Unit = block
  }, duration.toMillis, TimeUnit.MILLISECONDS)
 }

db-async-common/src/test/scala/com/github/mauricio/async/db/pool/DummyTimeoutScheduler.scala

@@ -17,7 +17,9 @@
 package com.github.mauricio.async.db.pool
 
 import java.util.concurrent.atomic.AtomicInteger
-import com.github.mauricio.async.db.util.ExecutorServiceUtils
+import com.github.mauricio.async.db.util.{NettyUtils, ExecutorServiceUtils}
+import io.netty.channel.EventLoopGroup
+
 /**
  * Implementation of TimeoutScheduler used for testing
  */
@@ -26,4 +28,5 @@ class DummyTimeoutScheduler extends TimeoutScheduler {
  private val timeOuts = new AtomicInteger
  override def onTimeout = timeOuts.incrementAndGet
  def timeoutCount = timeOuts.get()
+ def eventLoopGroup : EventLoopGroup = NettyUtils.DefaultEventLoopGroup
 }

db-async-common/src/test/scala/com/github/mauricio/async/db/pool/TimeoutSchedulerSpec.scala

@@ -28,11 +28,10 @@ class TimeoutSchedulerSpec extends SpecificationWithJUnit {
 
  val TIMEOUT_DID_NOT_PASS = "timeout did not pass"
 
-
  "test timeout did not pass" in {
  val timeoutScheduler = new DummyTimeoutScheduler()
  val promise = Promise[String]()
- val scheduledFuture = timeoutScheduler.addTimeout(promise,Duration(1000, MILLISECONDS))
+ val scheduledFuture = timeoutScheduler.addTimeout(promise,Some(Duration(1000, MILLISECONDS)))
  Thread.sleep(100);
  promise.isCompleted === false
  promise.success(TIMEOUT_DID_NOT_PASS)
@@ -46,7 +45,7 @@ class TimeoutSchedulerSpec extends SpecificationWithJUnit {
  val timeoutMillis = 100
  val promise = Promise[String]()
  val timeoutScheduler = new DummyTimeoutScheduler()
- val scheduledFuture = timeoutScheduler.addTimeout(promise,Duration(timeoutMillis, MILLISECONDS))
+ val scheduledFuture = timeoutScheduler.addTimeout(promise,Some(Duration(timeoutMillis, MILLISECONDS)))
  Thread.sleep(1000)
  promise.isCompleted === true
  scheduledFuture.get.isCancelled === false
@@ -55,11 +54,10 @@ class TimeoutSchedulerSpec extends SpecificationWithJUnit {
  promise.future.value.get.get must throwA[TimeoutException](message = s"Operation is timeouted after it took too long to return \\(${timeoutMillis} milliseconds\\)")
  }
 
-
  "test no timeout" in {
  val timeoutScheduler = new DummyTimeoutScheduler()
  val promise = Promise[String]()
- val scheduledFuture = timeoutScheduler.addTimeout(promise,Duration.Inf)
+ val scheduledFuture = timeoutScheduler.addTimeout(promise,None)
  Thread.sleep(1000)
  scheduledFuture === None
  promise.isCompleted === false

mysql-async/src/main/scala/com/github/mauricio/async/db/mysql/MySQLConnection.scala

@@ -43,7 +43,7 @@ class MySQLConnection(
  configuration: Configuration,
  charsetMapper: CharsetMapper = CharsetMapper.Instance,
  group : EventLoopGroup = NettyUtils.DefaultEventLoopGroup,
- executionContext : ExecutionContext = ExecutorServiceUtils.CachedExecutionContext
+ implicit val executionContext : ExecutionContext = ExecutorServiceUtils.CachedExecutionContext
  )
  extends MySQLHandlerDelegate
  with Connection
@@ -55,10 +55,8 @@ class MySQLConnection(
  // validate that this charset is supported
  charsetMapper.toInt(configuration.charset)
 
-
  private final val connectionCount = MySQLConnection.Counter.incrementAndGet()
  private final val connectionId = s"[mysql-connection-$connectionCount]"
- override implicit val internalPool = executionContext
 
  private final val connectionHandler = new MySQLConnectionHandler(
  configuration,
@@ -80,6 +78,8 @@ class MySQLConnection(
  def lastException : Throwable = this._lastException
  def count : Long = this.connectionCount
 
+ override def eventLoopGroup : EventLoopGroup = group
+
  def connect: Future[Connection] = {
  this.connectionHandler.connect.onFailure {
  case e => this.connectionPromise.tryFailure(e)

mysql-async/src/test/scala/com/github/mauricio/async/db/mysql/QueryTimeoutSpec.scala

@@ -66,7 +66,7 @@ class QueryTimeoutSpec extends Specification with ConnectionHelper {
  port = 3306,
  password = Some("root"),
  database = Some("mysql_async_tests"),
- queryTimeout = Duration(1,NANOSECONDS)
+ queryTimeout = Some(Duration(1,NANOSECONDS))
  )
 
  def longTimeoutConfiguration = new Configuration(
@@ -75,6 +75,6 @@ class QueryTimeoutSpec extends Specification with ConnectionHelper {
  port = 3306,
  password = Some("root"),
  database = Some("mysql_async_tests"),
- queryTimeout = Duration(5,SECONDS)
+ queryTimeout = Some(Duration(5,SECONDS))
  )
 }

postgresql-async/src/main/scala/com/github/mauricio/async/db/postgresql/PostgreSQLConnection.scala

@@ -46,7 +46,7 @@ class PostgreSQLConnection
  encoderRegistry: ColumnEncoderRegistry = PostgreSQLColumnEncoderRegistry.Instance,
  decoderRegistry: ColumnDecoderRegistry = PostgreSQLColumnDecoderRegistry.Instance,
  group : EventLoopGroup = NettyUtils.DefaultEventLoopGroup,
- executionContext : ExecutionContext = ExecutorServiceUtils.CachedExecutionContext
+ implicit val executionContext : ExecutionContext = ExecutorServiceUtils.CachedExecutionContext
  )
  extends PostgreSQLConnectionDelegate
  with Connection
@@ -65,7 +65,6 @@ class PostgreSQLConnection
 
  private final val currentCount = Counter.incrementAndGet()
  private final val preparedStatementsCounter = new AtomicInteger()
- override implicit val internalPool = executionContext
 
  private val parameterStatus = new scala.collection.mutable.HashMap[String, String]()
  private val parsedStatements = new scala.collection.mutable.HashMap[String, PreparedStatementHolder]()
@@ -82,6 +81,7 @@ class PostgreSQLConnection
 
  private var queryResult: Option[QueryResult] = None
 
+ override def eventLoopGroup : EventLoopGroup = group
  def isReadyForQuery: Boolean = this.queryPromise.isEmpty
 
  def connect: Future[Connection] = {

postgresql-async/src/test/scala/com/github/mauricio/async/db/postgresql/PostgreSQLConnectionSpec.scala

@@ -285,16 +285,12 @@ class PostgreSQLConnectionSpec extends Specification with DatabaseTestHelper {
  try {
  withHandler(configuration, {
  handler =>
- executeQuery(handler, "SELECT 0")
- throw new IllegalStateException("should not have come here")
+ val result = executeQuery(handler, "SELECT 0")
+ throw new IllegalStateException("should not have arrived")
  })
  } catch {
- case e: GenericDatabaseException => {
+ case e: GenericDatabaseException =>
  e.errorMessage.fields(InformationMessage.Routine) === "auth_failed"
- }
- case e: Exception => {
- throw new IllegalStateException("should not have come here")
- }
  }
 
  }