Akka Streams 中的“conditionalZip”运算符

Question

假设我有两个来源：

val first = Source(1 :: 2 :: 4 :: 6 :: Nil)
val second = Source(1 :: 2 :: 3 :: 4 :: 5 :: 6 :: 7 :: Nil)

是否可以创建一个仅根据条件配对元素的 zip？我的意思是：

first.conditionalZip(second, _ == _) // if that method exited

该代码将从first 源中获取元素并从second 中删除元素，直到有一个满足条件的元素，然后输出一个元组。上述调用的结果将是(1, 1), (2, 2), (4, 4), (6, 6)。

Answer 1

考虑对两个Sources进行压缩，然后使用statefulMapConcat按照条件函数对压缩后的元素进行变换，如下图：

import akka.stream.scaladsl._
import akka.NotUsed

def popFirstMatch(ls: List[Int], condF: Int => Boolean): (Option[Int], List[Int]) = {
  ls.find(condF) match {
    case None =>
      (None, ls)
    case Some(e) => 
      val idx = ls.indexOf(e)
      if (idx < 0)
        (None, ls)
      else {
        val (l, r) = ls.splitAt(idx)
        (r.headOption, l ++ r.tail)
      }
  }
}

def conditionalZip( first: Source[Int, NotUsed],
                    second: Source[Int, NotUsed],
                    filler: Int,
                    condFcn: (Int, Int) => Boolean ): Source[(Int, Int), NotUsed] = {
    first.zipAll(second, filler, filler).statefulMapConcat{ () =>
      var prevList = List.empty[Int]
      tuple => tuple match { case (e1, e2) =>
        if (e2 != filler) {
          if (e1 != filler && condFcn(e1, e2))
            (e1, e2) :: Nil
          else {
            if (e1 != filler)
              prevList :+= e1
            val (opElem, rest) = popFirstMatch(prevList, condFcn(_, e2))
            prevList = rest
            opElem match {
              case None => Nil
              case Some(e) => (e, e2) :: Nil
            }
          }
        }
        else
          Nil
      }
    }
}

试运行：

import akka.actor.ActorSystem
implicit val system = ActorSystem("system")
implicit val ec = system.dispatcher

// Example 1:
val first = Source(1 :: 2 :: 4 :: 6 :: Nil)
val second = Source(1 :: 2 :: 3 :: 4 :: 5 :: 6 :: 7 :: Nil)

conditionalZip(first, second, Int.MinValue, _ == _).runForeach(println) 
// (1,1)
// (2,2)
// (4,4)
// (6,6)

conditionalZip(first, second, Int.MinValue, _ > _).runForeach(println) 
// (4,3)
// (6,4)

conditionalZip(first, second, Int.MinValue, _ < _).runForeach(println) 
// (1,2)
// (2,3)
// (4,5)
// (6,7)

// Example 2:
val first = Source(3 :: 9 :: 5 :: 5 :: 6 :: Nil)
val second = Source(1 :: 3 :: 5 :: 2 :: 5 :: 6 :: Nil)

conditionalZip(first, second, Int.MinValue, _ == _).runForeach(println)
// (3,3)
// (5,5)
// (5,5)
// (6,6)

conditionalZip(first, second, Int.MinValue, _ > _).runForeach(println)
// (3,1)
// (9,3)
// (5,2)
// (6,5)

conditionalZip(first, second, Int.MinValue, _ < _).runForeach(println)
// (3,5)
// (5,6)

几点说明：

1. 方法 zipAll（在 Akka Stream 2.6+ 上可用）压缩两个源，同时使用提供的“填充”值填充更少的元素。在这种情况下，这些填充物是没有意义的，因此应该分配一个与实际元素不同的值。

2. 一个内部列表prevList 在statefulMapConcat 中用于存储来自第一个来源的元素，以便在后续迭代中与来自第二个来源的元素进行比较。如果源中的元素不同，List 可以替换为 Set 以获得更好的查找性能。

3. 方法popFirstMatch用于提取prevList中与提供的部分条件函数匹配的第一个元素，返回Option类型元素的Tuple和剩余的List。

4. 请注意，这只是说明statefulMapConcat 如何解决所描述的问题。如果没有详细实现以覆盖所有情况或细化相当广泛的条件函数(Int, Int) => Boolean 的范围，示例代码的行为可能不一定符合确切要求。