0

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

我按照本指南在我的服务DatabaseProvider之间增加了一层。Database因此，我可以提供这样的静态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。我目前的方法是让杰克逊从文件中读取一个 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)
4

1 回答 1

0

在您的情况下，这里没有任何隐含的内容，运行时的东西看起来应该很简单。您的问题很可能是杰克逊未能从文件中读取。要进行测试，请尝试以下方法，如果它成功尝试连接到本地 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 我并不感到惊讶。

于 2018-03-19T17:59:22.177 回答