1

私はAkka Streamsで遊んでいて、MongoDBコレクションからポーリングされたイベントの強化と処理を試みています。ただし、外部データ ソースに接続する必要がある可能性があるイベント エンリッチャーを実装する最善の方法については、疑問があります。

mapFuture は適しているようですが、いくつか問題があります。

class EventEnricherActor extends Actor with ActorLogging {

  // ...

  def receive = {
    case e: Event =>
      sender ! augmentEvent(e)
  }

}

そして私のアプリ：

val enricherActor = actorSystem.actorOf(Props[EventEnricherActor])

Flow(mongodbConsumer).
      mapFuture(msg => enricherActor ? msg).
      onComplete(materializer) { _ => actorSystem.shutdown()}

しかし、私はこのエラーで立ち往生しています:

java.lang.ClassCastException: messages.Event cannot be cast to scala.runtime.Nothing$

mapFuture への呼び出しで。

私は何が欠けていますか？

これらのエンリッチメントに対処するためのより良いアイデアはありますか?

スタック トレースを更新します。

[ERROR] [08/12/2014 11:48:07.314] [actor-system-akka.actor.default-dispatcher-2] [akka://actor-system/user/flow-1-2-transform] failure during processing
java.lang.ClassCastException: messages.Event cannot be cast to scala.runtime.Nothing$
    at apps.Consume$$anonfun$1$$anonfun$apply$1.apply(Consume.scala:47)
    at akka.stream.impl.MapFutureProcessorImpl$$anonfun$1.apply$mcV$sp(MapFutureProcessorImpl.scala:125)
    at akka.stream.impl.Pump$$anonfun$pump$1.apply$mcV$sp(Transfer.scala:163)
    at akka.stream.impl.Pump$$anonfun$pump$1.apply(Transfer.scala:163)
    at akka.stream.impl.Pump$$anonfun$pump$1.apply(Transfer.scala:163)
    at akka.stream.impl.ActorBasedFlowMaterializer$.withCtx(ActorBasedFlowMaterializer.scala:133)
    at akka.stream.impl.Pump$class.pump(Transfer.scala:163)
    at akka.stream.impl.ActorProcessorImpl.pump(ActorProcessor.scala:238)
    at akka.stream.impl.BatchingInputBuffer.enqueueInputElement(ActorProcessor.scala:93)
    at akka.stream.impl.BatchingInputBuffer$$anonfun$upstreamRunning$1.applyOrElse(ActorProcessor.scala:140)
    at scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
    at scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
    at scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
    at akka.stream.impl.SubReceive.apply(Transfer.scala:18)
    at akka.stream.impl.SubReceive.apply(Transfer.scala:14)
    at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
    at akka.stream.impl.SubReceive.applyOrElse(Transfer.scala:14)
    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:165)
    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:166)
    at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
    at akka.stream.impl.ActorProcessorImpl.aroundReceive(ActorProcessor.scala:238)
    at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
    at akka.actor.ActorCell.invoke(ActorCell.scala:487)
    at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238)
    at akka.dispatch.Mailbox.run(Mailbox.scala:220)
    at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:393)
    at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

ありがとう

4

1 に答える 1

1

私の間違い、mongodbConsumer (ActorProducer) は型指定されていません!

したがって、この：

val mongodbConsumer = ActorProducer[Event](system.actorOf(MongodbConsumerActor.props(db)))

それ以外の：

val mongodbConsumer = ActorProducer(system.actorOf(MongodbConsumerActor.props(db)))
于 2014-08-13T11:15:20.783 に答える