Concurrent.patchPanel 不从多个枚举器发送数据 - 仅从添加的最后一个枚举器发送

Question

我希望其他人使用 patchPanel 将多个枚举器组合在一起，通过 websocket 向下传递到客户端。我遇到的问题是patchPanel 仅发送来自添加到其中的最后一个枚举器的数据馈送。

我遵循了以下示例； http://lambdaz.blogspot.ca/2012/12/play-21-multiplexing-enumerators-into.html 这是我能找到的关于 patchPanel 的唯一参考。

版本；玩！ 2.1.1（使用 Java 1.7.0_11 和 Scala 2.10.0）

网络套接字方法；

def monitorStream = WebSocket.async[JsValue] { request =>
  val promiseIn = promise[Iteratee[JsValue, Unit]]
  val out = Concurrent.patchPanel[JsValue] { patcher =>
    val in = Iteratee.foreach[JsValue] { json =>
      val event:Option[String] = (json \ "event").asOpt[String]
      val systemId = (json \ "systemId").as[Long]
      event.getOrElse("") match {
        case "join" => 
          val physicalSystem = SystemIdHandler.getById(systemId)
          val monitorOut = (MonitorStreamActor.joinMonitor(physicalSystem)) 
          monitorOut map { enum =>
            val success = patcher.patchIn(enum)
        }
      }
    }.mapDone { _ => Logger.info("Disconnected") }
    promiseIn.success(in)
  }
  future(Iteratee.flatten(promiseIn.future),out)
}

MonitorStreamActor 调用；

  def joinMonitor(physicalSystem: PhysicalSystem):
    scala.concurrent.Future[Enumerator[JsValue]]
     = {
    val monitorActor = ActorBase.akkaSystem.actorFor("/user/system-" + physicalSystem.name +"/stream")
    (monitorActor ? MonitorJoin()).map {
      case MonitorConnected(enumerator) =>
        enumerator
      }

  }

枚举器返回正常，输入它的数据来自调用actor的计时器。 Actor定义，定时器命中UpdatedTranStates情况；

class MonitorStreamActor() extends Actor {
  val (monitorEnumerator, monitorChannel) = Concurrent.broadcast[JsValue]
  import play.api.Play.current
  def receive = {
    case MonitorJoin() => {
      Logger.debug ("Actor monitor join")
      sender ! MonitorConnected(monitorEnumerator)
    }
    case UpdatedTranStates(systemName,tranStates) => {
      //println("Got updated Tran States")
      val json = Json.toJson(tranStates.map(m => Map("State" -> m._1, "Count" -> m._2) ))
      //println("Pushing updates to monitorChannel")
      sendUpdateToClients(systemName, "states", json)
    }
  def sendUpdateToClients(systemName:String, updateType:String, json:JsValue) {
    monitorChannel.push(Json.toJson(
      Map(
        "dataType"->Json.toJson(updateType),
        "systemName" -> Json.toJson(systemName),
        "data"->json)))
  }
}
}

我已经对此进行了一段时间的探索，但没有找到为什么只有添加到 patchPanel 中的最后一个枚举器才会发送数据的原因。 API 文档没有太大帮助，听起来您所要做的就是调用 patchIn 并且它应该将所有枚举器组合到一个迭代器，但似乎并非如此。

Answer 1

PatchPanel 按设计将当前枚举器替换为 patchIn 方法提供的新枚举器。

为了将多个枚举器组合在一起，您需要使用 interleave 或 andThen 方法将枚举器组合在一起。对于这种情况，最好使用交错，因为它会从每个 Enumerator 获取可用的事件，而不是清空一个然后移动到下一个（与 andThen 运算符一样）。

即，在monitorStream中；

val monitorOut = (MonitorStreamActor.joinMonitor(physicalSystem)) 
monitorOut map { enum =>
  mappedEnums += ((physicalSystem.name, enum))
  patcher.patchIn(Enumerator.interleave[JsValue]( mappedEnums.values.toSeq))
}

patcher 是补丁面板， mappedEnums 是 HashMap[String,Enumerator[JsValue]] - 每次 Enumerators 更改（添加或删除）时重新添加补丁程序 - 它可以工作，不确定它是否是最好的方法，但现在可以了:)