trying to use netty EventLoop

Narigo · Narigo · commit 19d3b214e97e · 2013-08-08T08:56:50.000+02:00
diff --git a/build.gradle b/build.gradle
@@ -61,10 +61,10 @@ dependencies {
   provided "org.scala-lang:scala-compiler:$scalaVersion"
   provided "io.vertx:lang-scala:$scalaLangModVersion"
 
-  compile("com.github.mauricio:postgresql-async_2.10:0.2.4") {
+  compile("com.github.mauricio:postgresql-async_2.10:$asyncDriverVersion") {
     exclude group: 'org.scala-lang'
   }
-  compile("com.github.mauricio:mysql-async_2.10:0.2.4") {
+  compile("com.github.mauricio:mysql-async_2.10:$asyncDriverVersion") {
     exclude group: 'org.scala-lang'
   }
 }
diff --git a/gradle.properties b/gradle.properties
@@ -7,6 +7,9 @@ modname=async-mysql-postgresql
 # Your module version
 version=0.1.0-SNAPSHOT
 
+# The version of mauricios async driver
+asyncDriverVersion=0.2.5-SNAPSHOT
+
 # The test timeout in seconds
 testtimeout=5
 
diff --git a/src/main/resources/langs.properties b/src/main/resources/langs.properties
@@ -0,0 +1,2 @@
+scala=io.vertx~lang-scala~2.0.0-SNAPSHOT:org.vertx.scala.platform.impl.ScalaVerticleFactory
+.scala=scala
diff --git a/src/main/scala/com/campudus/vertx/busmod/ScalaBusMod.scala b/src/main/scala/com/campudus/vertx/busmod/ScalaBusMod.scala
@@ -24,7 +24,9 @@ trait ScalaBusMod extends MessageHelper with VertxExecutionContext with (Message
 
     fut map { reply => msg.reply(reply.toJson) } recover {
       // case x: BusException => msg.reply(new JsonObject().putString("status", "error").putString("message", x.getMessage()).putString("id", x.getId()))
-      case x => msg.reply(new JsonObject().putString("status", "error").putString("message", x.getMessage()))
+      case x =>
+        x.printStackTrace(System.err)
+        msg.reply(new JsonObject().putString("status", "error").putString("message", x.getMessage()))
     }
   }
 
diff --git a/src/main/scala/com/campudus/vertx/database/ConnectionHandler.scala b/src/main/scala/com/campudus/vertx/database/ConnectionHandler.scala
@@ -9,13 +9,11 @@ import com.github.mauricio.async.db.Connection
 import com.campudus.vertx.busmod.ScalaBusMod
 import scala.concurrent.Future
 import com.campudus.vertx.database.pool.AsyncConnectionPool
+import org.vertx.scala.core.Vertx
 
-class ConnectionHandler(dbType: String, config: Configuration) extends ScalaBusMod {
+class ConnectionHandler(vertx: Vertx, dbType: String, config: Configuration) extends ScalaBusMod {
 
-  val pool = dbType match {
-    case "postgresql" => new PostgreSQLAsyncConnectionPool(config)
-    case _ => ???
-  }
+  val pool = AsyncConnectionPool(vertx, dbType, config)
 
   override def receive(msg: Message[JsonObject]) = {
     case "query" => pool.withConnection({ c: Connection =>
diff --git a/src/main/scala/com/campudus/vertx/database/Starter.scala b/src/main/scala/com/campudus/vertx/database/Starter.scala
@@ -16,7 +16,7 @@ class Starter extends Verticle {
 
   var connection: String = null
   var configuration: Configuration = null
-  var pool: ConnectionHandler = null
+  var handler: ConnectionHandler = null
 
   override def start(startedResult: org.vertx.scala.core.Future[Void]) = {
 
@@ -37,14 +37,14 @@ class Starter extends Verticle {
       val port = config.getInteger("port", defaultPortFor(connection))
       val username = config.getString("username", defaultUserFor(connection))
       val password = Option(config.getString("password")).orElse(defaultPasswordFor(connection))
-      val database = Option(config.getString("database"))
+      val database = Option(config.getString("database")).orElse(defaultDatabaseFor(connection))
 
       configuration = Configuration(username, host, port, password, database)
 
-      pool = new ConnectionHandler(connection, configuration)
-      vertx.eventBus.registerHandler(address)(pool)
+      handler = new ConnectionHandler(vertx, connection, configuration)
+      vertx.eventBus.registerHandler(address)(handler)
 
-      logger.error("Async database module for MySQL and PostgreSQL started.")
+      logger.error("Async database module for MySQL and PostgreSQL started with config " + configuration)
 
       startedResult.setResult(null)
     } catch {
@@ -55,14 +55,18 @@ class Starter extends Verticle {
   }
 
   override def stop() {
-    Option(pool).map(_.close)
+    Option(handler).map(_.close)
   }
 
   private def defaultPortFor(connection: String): Integer = connection match {
     case "postgresql" => 5432
     case "mysql" => 3306
   }
 
+  private def defaultDatabaseFor(connection: String): Option[String] = connection match {
+    case _ => Some("testdb")
+  }
+
   private def defaultUserFor(connection: String): String = connection match {
     case "postgresql" => "vertx"
     case "mysql" => "root"
diff --git a/src/main/scala/com/campudus/vertx/database/pool/AsyncConnectionPool.scala b/src/main/scala/com/campudus/vertx/database/pool/AsyncConnectionPool.scala
@@ -7,6 +7,8 @@ import scala.concurrent.ExecutionContext
 import scala.concurrent.Promise
 import scala.util.Failure
 import scala.util.Success
+import org.vertx.scala.core.Vertx
+import org.vertx.java.core.impl.EventLoopContext
 
 trait AsyncConnectionPool[ConnType <: Connection] {
 
@@ -36,8 +38,11 @@ trait AsyncConnectionPool[ConnType <: Connection] {
 
 object AsyncConnectionPool {
 
-  def apply(dbType: String, config: Configuration) = dbType match {
-    case "postgresql" => new PostgreSQLAsyncConnectionPool(config)
+  def apply(vertx: Vertx, dbType: String, config: Configuration) = dbType match {
+    case "postgresql" =>
+      new PostgreSQLAsyncConnectionPool(
+        config,
+        vertx.internal.currentContext().asInstanceOf[EventLoopContext].getEventLoop())
     case _ => throw new NotImplementedError
   }
 
diff --git a/src/main/scala/com/campudus/vertx/database/pool/PostgreSQLAsyncConnectionPool.scala b/src/main/scala/com/campudus/vertx/database/pool/PostgreSQLAsyncConnectionPool.scala
@@ -6,10 +6,12 @@ import scala.concurrent.Future
 import scala.concurrent.ExecutionContext
 import com.campudus.vertx.VertxExecutionContext
 import com.github.mauricio.async.db.Connection
+import org.vertx.java.core.impl.EventLoopContext
+import io.netty.channel.EventLoop
 
-class PostgreSQLAsyncConnectionPool(config: Configuration, implicit val executionContext: ExecutionContext = VertxExecutionContext) extends AsyncConnectionPool[PostgreSQLConnection] {
+class PostgreSQLAsyncConnectionPool(config: Configuration, eventLoop: EventLoop, implicit val executionContext: ExecutionContext = VertxExecutionContext) extends AsyncConnectionPool[PostgreSQLConnection] {
 
-  override def take() = new PostgreSQLConnection(config).connect
+  override def take() = new PostgreSQLConnection(configuration = config, group = eventLoop).connect
 
   override def giveBack(connection: Connection) = {
     connection.disconnect map (_ => this) recover {
diff --git a/src/test/scala/com/campudus/test/postgresql/PostgreSQLTest.scala b/src/test/scala/com/campudus/test/postgresql/PostgreSQLTest.scala
@@ -36,6 +36,7 @@ class PostgreSQLTest extends TestVerticle with VertxScalaHelpers {
   }
 
   private def expectOk(q: JsonObject): Future[JsonObject] = ebSend(q) map { reply =>
+    logger.error("got reply: " + reply.encode())
     assertEquals("ok", reply.getString("status"))
     reply
   }

Original file line number	Diff line number	Diff line change
`@@ -61,10 +61,10 @@ dependencies {`
`61`	`61`	`provided "org.scala-lang:scala-compiler:$scalaVersion"`
`62`	`62`	`provided "io.vertx:lang-scala:$scalaLangModVersion"`
`63`	`63`
`64`		`- compile("com.github.mauricio:postgresql-async_2.10:0.2.4") {`
	`64`	`+ compile("com.github.mauricio:postgresql-async_2.10:$asyncDriverVersion") {`
`65`	`65`	`exclude group: 'org.scala-lang'`
`66`	`66`	`}`
`67`		`- compile("com.github.mauricio:mysql-async_2.10:0.2.4") {`
	`67`	`+ compile("com.github.mauricio:mysql-async_2.10:$asyncDriverVersion") {`
`68`	`68`	`exclude group: 'org.scala-lang'`
`69`	`69`	`}`
`70`	`70`	`}`
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+scala=io.vertx~lang-scala~2.0.0-SNAPSHOT:org.vertx.scala.platform.impl.ScalaVerticleFactory`
	`2`	`+.scala=scala`
Original file line number	Diff line number	Diff line change
`@@ -24,7 +24,9 @@ trait ScalaBusMod extends MessageHelper with VertxExecutionContext with (Message`
`24`	`24`
`25`	`25`	`fut map { reply => msg.reply(reply.toJson) } recover {`
`26`	`26`	`// case x: BusException => msg.reply(new JsonObject().putString("status", "error").putString("message", x.getMessage()).putString("id", x.getId()))`
`27`		`- case x => msg.reply(new JsonObject().putString("status", "error").putString("message", x.getMessage()))`
	`27`	`+ case x =>`
	`28`	`+ x.printStackTrace(System.err)`
	`29`	`+ msg.reply(new JsonObject().putString("status", "error").putString("message", x.getMessage()))`
`28`	`30`	`}`
`29`	`31`	`}`
`30`	`32`
Original file line number	Diff line number	Diff line change
`@@ -36,6 +36,7 @@ class PostgreSQLTest extends TestVerticle with VertxScalaHelpers {`
`36`	`36`	`}`
`37`	`37`
`38`	`38`	`private def expectOk(q: JsonObject): Future[JsonObject] = ebSend(q) map { reply =>`
	`39`	`+ logger.error("got reply: " + reply.encode())`
`39`	`40`	`assertEquals("ok", reply.getString("status"))`
`40`	`41`	`reply`
`41`	`42`	`}`