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 回答