updateLiveStats

updateLiveStats.scala

def updateLiveStats(
                       time: Time,
                       liveId: String,
                           event: Option[Iterable[LiveEvent]],
                           state: State[LiveStatistics]
                          ): Option[LiveStatistics] = {

    event.flatMap { es =>
      val (creates, notCreates) = es.partition(_.isCreate)

      val (finishes, others) = notCreates.partition(_.isFinish)

      //var currentStats: Option[LiveStatistics] = None

      /**
        * 如果有直播创建，先初始化该直播
        * 然后处理进出，打赏等
        * 如果有直播结束，最后处理
        */
      if (!creates.isEmpty) {
        val createdEvent = creates.head.asInstanceOf[LiveCreated]
        val latestStats = LiveStatistics(
          IdGenerator.generateUuid,
          liveId,
          createdEvent.liveName,
          createdEvent.anchor,
          createdEvent.anchorName,
          createdEvent.startTime
        )

        val updateStats = others.foldLeft(latestStats)((s, e) => e.updateStatistics(s))

        state.update(updateStats)

        //currentStats = Some(updateStats)

      }

      if (state.exists()) {
        val updateStats = others.foldLeft(state.get())((s, e) => e.updateStatistics(s))

        state.update(updateStats)

        //currentStats = Some(updateStats)

      }

      // 设置直播结束时间
      for {
        f <- finishes.headOption
        s <- state.getOption()
      } yield {
        state.update(f.updateStatistics(s))
      }

      state.getOption()
    }
}