幻影DSL中Cassandra连接的运行时配置

Question

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

我遵循本指南，在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 {}

没有单例对象，我该如何在运行时获得相同的结果？

我做了几个attempst，但总是有 NullPointerException s。我当前的方法是拥有一个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)

Answer 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。