Phantom DSL 中 Cassandra 连接的运行时配置答案

【问题标题】：Runtime Configuration of Cassandra connection in Phantom DSLPhantom DSL 中 Cassandra 连接的运行时配置
【发布时间】：2018-03-20 01:59:22
【问题描述】：

我正在使用 phantom 连接到 Apache Cassandra 并希望在运行时配置连接器，即我想解析一些配置文件，提取 Cassandra 数据库列表并以某种方式将其传递给我的 Database 对象。

我跟随this guide 在Database 和我的服务之间增加了一层DatabaseProvider。因此，我可以像这样提供静态DatabaseProvider：

object ProdConnector {
  val connector = ContactPoints(Seq("dev-cassndr.containers"), 9042)
    .keySpace("test")
}

object ProdDatabase extends MyDatabase(ProdConnector.connector)

trait ProdDatabaseProvider extends MyDatabaseProvider {
  override def database: MyDatabase = ProdDatabase
}

在我的 main 函数中我这样做

val service = new MessageService with ProdDatabaseProvider {}

如何在没有单例对象的情况下在运行时获得相同的结果？

我尝试了几次，但总是得到NullPointerExceptions。我目前的方法是让 Jackson 从文件中读取 Cassandra 配置对象：

case class CassandraConfigurator(
  contactPoints: Seq[String],
  keySpace: String,
  port: Int = 9042,
) {
  @JsonCreator
  def this() = this(null, null, 9042)

  val connection: CassandraConnection = {
    val p = ContactPoints(contactPoints, port)
    p.keySpace(keySpace)
  }
}

然后我的入口点从fs2 扩展StreamApp

object Main extends StreamApp[IO] {
  override def stream(args: List[String], reqShutdown: IO[Unit])
      : Stream[IO, ExitCode] = {
    val conf: CassandraConfigurator = ???
    val service = new MyService with MyDatabaseProvider {
      override def database: MyDatabase = new MyDatabase(conf.connection)
    }

    service.database.create()
    val api = ApiWithService(service).getApi
    BlazeBuilder[IO].bindHttp(80, "0.0.0.0").mountService(api, "/").serve
  }
}

这会导致以下错误：

12:44:03.436 [pool-10-thread-1] INFO  com.datastax.driver.core.GuavaCompatibility - Detected Guava >= 19 in the classpath, using modern compatibility layer
12:44:03.436 [pool-10-thread-1] INFO  com.datastax.driver.core.GuavaCompatibility - Detected Guava >= 19 in the classpath, using modern compatibility layer
12:44:03.463 [pool-10-thread-1] DEBUG com.datastax.driver.core.SystemProperties - com.datastax.driver.NEW_NODE_DELAY_SECONDS is undefined, using default value 1
12:44:03.463 [pool-10-thread-1] DEBUG com.datastax.driver.core.SystemProperties - com.datastax.driver.NEW_NODE_DELAY_SECONDS is undefined, using default value 1
12:44:03.477 [pool-10-thread-1] DEBUG com.datastax.driver.core.SystemProperties - com.datastax.driver.NOTIF_LOCK_TIMEOUT_SECONDS is undefined, using default value 60
12:44:03.477 [pool-10-thread-1] DEBUG com.datastax.driver.core.SystemProperties - com.datastax.driver.NOTIF_LOCK_TIMEOUT_SECONDS is undefined, using default value 60
java.lang.NullPointerException
    at com.outworkers.phantom.connectors.ContactPoints$.$anonfun$apply$3(ContactPoint.scala:101)
    at com.outworkers.phantom.connectors.DefaultSessionProvider.<init>(DefaultSessionProvider.scala:37)
    at com.outworkers.phantom.connectors.CassandraConnection.provider$lzycompute(CassandraConnection.scala:46)
    at com.outworkers.phantom.connectors.CassandraConnection.provider(CassandraConnection.scala:41)
    at com.outworkers.phantom.connectors.CassandraConnection.session$lzycompute(CassandraConnection.scala:52)
    at com.outworkers.phantom.connectors.CassandraConnection.session(CassandraConnection.scala:52)
    at com.outworkers.phantom.database.Database.session$lzycompute(Database.scala:36)
    at com.outworkers.phantom.database.Database.session(Database.scala:36)
    at com.outworkers.phantom.ops.DbOps.$anonfun$createAsync$2(DbOps.scala:66)
    at com.outworkers.phantom.builder.query.execution.ExecutionHelper$.$anonfun$sequencedTraverse$2(ExecutableStatements.scala:71)
    at scala.concurrent.Future.$anonfun$flatMap$1(Future.scala:304)
    at scala.concurrent.impl.Promise.$anonfun$transformWith$1(Promise.scala:37)
    at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:60)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)

Exception: sbt.TrapExitSecurityException thrown from the UncaughtExceptionHandler in thread "run-main-0"
[error] java.lang.RuntimeException: Nonzero exit code: 1
[error]     at sbt.Run$.executeTrapExit(Run.scala:124)
[error]     at sbt.Run.run(Run.scala:77)
[error]     at sbt.Defaults$.$anonfun$bgRunTask$5(Defaults.scala:1168)
[error]     at sbt.Defaults$.$anonfun$bgRunTask$5$adapted(Defaults.scala:1163)
[error]     at sbt.internal.BackgroundThreadPool.$anonfun$run$1(DefaultBackgroundJobService.scala:366)
[error]     at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:12)
[error]     at scala.util.Try$.apply(Try.scala:209)
[error]     at sbt.internal.BackgroundThreadPool$BackgroundRunnable.run(DefaultBackgroundJobService.scala:289)
[error]     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
[error]     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
[error]     at java.lang.Thread.run(Thread.java:748)

【问题讨论】：

NPE 的原因是什么？你有堆栈跟踪吗？
刚刚用我当前的方法和完整的堆栈跟踪以及发生错误的日志更新了问题。我猜这和stackoverflow.com/questions/18807564/… 有点像，但我不知道如何破解它……

标签： cassandra phantom-dsl http4s fs2

【解决方案1】：

在你的情况下，这里没有任何隐含的东西，运行时的东西看起来应该很简单。您的问题很可能是杰克逊未能从文件中读取。要进行测试，请尝试以下操作，如果它成功尝试连接到本地 Cassandra，那么这就是您的问题。

object Main extends StreamApp[IO] {
  override def stream(args: List[String], reqShutdown: IO[Unit])
      : Stream[IO, ExitCode] = {

    val service = new MyService with MyDatabaseProvider {
      override def database: MyDatabase = new MyDatabase(Connector.default)
    }

    service.database.create()
    val api = ApiWithService(service).getApi
    BlazeBuilder[IO].bindHttp(80, "0.0.0.0").mountService(api, "/").serve
  }
}

您确定val conf: CassandraConfigurator = ??? 已正确初始化吗？如果这就是您在代码中所拥有的，那么您获得 NPE 我并不感到惊讶。

【讨论】：