scala - 如何将 `andThen` Akka `Receive` s？

Question

我有以下内容:

def become(r: ListMap[String, Receive]) {
  val undefined: PartialFunction[Any, Unit] = Map.empty

  val receive = r.foldLeft(undefined) { (accum: Receive, rhs: Pair[String, Receive]) =>
    accum orElse rhs._2
  }

  val echo: Receive = {
    case message => println(s"*************************************************** ${sender}: processing ${message}")
  }

  context.become(echo andThen receive)
}

receive 本身工作正常，但是当使用 echo andThen receive 时，系统会吐出一堆:

scala.MatchError: () (of class scala.runtime.BoxedUnit)
    at scala.PartialFunction$$anon$1.apply(PartialFunction.scala:248)
    at scala.PartialFunction$$anon$1.apply(PartialFunction.scala:246)
    at kvstore.Replica$$anonfun$secondary$1.applyOrElse(Replica.scala:234)
    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 kvstore.Replica$$anonfun$primary$1.applyOrElse(Replica.scala:148)
    at scala.PartialFunction$OrElse.apply(PartialFunction.scala:162)
    at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
    at scala.collection.AbstractMap.applyOrElse(Map.scala:58)
    at scala.PartialFunction$OrElse.apply(PartialFunction.scala:162)
    at scala.PartialFunction$AndThen.applyOrElse(PartialFunction.scala:185)
    at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498)
    at akka.actor.ActorCell.invoke(ActorCell.scala:456)
    at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:237)
    at akka.dispatch.Mailbox.run(Mailbox.scala:219)
    at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386)
    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)

堆栈跟踪指向:

def secondary(
                // expected sequence id
                expectedSeq: Long,
                // a map from sequence id to replication requester
                replicationRequester: Map[Long, ActorRef],
                // a map from key to triple of sequence id, persistence repeat, and persistence timeout
                persistenceSchedulers: Map[String, Triple[Long, Cancellable, Cancellable]]): Receive = {

和:

def primary(
              // a map from secondary replicas to replicators
              replicators: Map[ActorRef, ActorRef],
              // a map from pair of key and replicator to message id, replication repeat, and replication timeout
              replicationSchedulers: Map[Pair[String, ActorRef], Triple[Long, Cancellable, Cancellable]],
              // map from message id to pair of replication requester and set of replicas
              awaitingReplication: Map[Long, Pair[ActorRef, Set[ActorRef]]]): Receive = {

两者都通过类似的方式使用:

def receive = {
  case JoinedSecondary => {
    receivePartialFunctions = ListMap(secondaryReceiveName -> secondary(0, Map.empty, Map.empty))
    become(receivePartialFunctions)
  }

  case JoinedPrimary => {
    val replicator = context.actorOf(Props(classOf[Replicator], self), s"replicator-${System.nanoTime()}")

    receivePartialFunctions = ListMap(
      primaryReceiveName -> primary(Map(self -> replicator), Map.empty, Map.empty),
      secondaryReceiveName -> secondary(0, Map.empty, Map.empty)
    )
    become(receivePartialFunctions)
  }
}

这个异常是什么意思，需要做什么来修复它？

Answer 1

我认为你的回声一定不是来自 Receive 类型，而是

PartialFunction[Any,Any]

例如:

scala> val echo :PartialFunction[Any,Unit] = { case message => println(message) }
echo: PartialFunction[Any,Unit] = <function1>

scala> val echo2 :PartialFunction[Any,Unit] = { case message => println(message+"2") }
echo2: PartialFunction[Any,Unit] = <function1>

scala> (echo andThen echo2)("hi")
hi
()2

对比

scala> val echo :PartialFunction[Any,Any] = { case message => println(message); message }
echo: PartialFunction[Any,Any] = <function1>


scala> (echo andThen echo2)("hi")
hi
hi2

您的第一个 PartialFunction 将结果发送给第二个。您的“echo”服务发送 Unit，因此第二个无法执行任何操作。