5

我想将两个 scalaz 流与一个谓词结合起来,该谓词从任一流中选择下一个元素。例如,我希望这个测试通过:

val a = Process(1, 2, 5, 8)
val b = Process(3, 4, 5, 7)

choose(a, b)(_ < _).toList shouldEqual List(1, 2, 3, 4, 5, 5, 7, 8)

正如你所看到的,我们不能做一些聪明的事情,比如zip对这两个元素进行排序,因为有时可能会连续选择其中一个过程。

我尝试了一个我认为可行的解决方案。它编译!但该死的,如果它什么都不做。JVM只是挂起:(

import scalaz.stream.Process._
import scalaz.stream._

object StreamStuff {
  def choose[F[_], I](a:Process[F, I], b:Process[F, I])(p: (I, I) => Boolean): Process[F, I] =
    (a.awaitOption zip b.awaitOption).flatMap {
      case (Some(ai), Some(bi)) =>
        if(p(ai, bi)) emit(ai) ++ choose(a, emit(bi) ++ b)(p)
        else emit(bi) ++ choose(emit(ai) ++ a, b)(p)
      case (None, Some(bi)) => emit(bi) ++ b
      case (Some(ai), None) => emit(ai) ++ a
      case _ => halt
    }
}

请注意,以上是我的第二次尝试。在我的第一次尝试中,我尝试创建一个Tee但我无法弄清楚如何取消消费失败者元素。我觉得我需要像这里一样的递归。

我正在使用流版本0.7.3a

非常感谢任何提示(包括增量提示,因为我想简单地学习如何自己解决这些问题)!

4

2 回答 2

5

我将在下面给出一些提示和实现,因此如果您想自己制定解决方案,您可能需要覆盖屏幕。

免责声明:这只是我想到的第一种方法,而且我对 scalaz-stream API 的熟悉程度有点生疏,所以可能有更好的方法来实现这个操作,这个方法可能在某些可怕的方面完全错误,等等.

提示 1

您可以在下一次递归调用中传递它们,而不是尝试“取消使用”丢失的元素。

提示 2

您可以通过指示哪一方最后输掉来避免必须累积多个输掉的元素。

提示 3

当我使用 Scalaz 流时,我经常发现首先使用普通集合来勾勒出一个实现更容易。这是列表所需的辅助方法:

/**
 * @param p if true, the first of the pair wins
 */
def mergeListsWithHeld[A](p: (A, A) => Boolean)(held: Either[A, A])(
  ls: List[A],
  rs: List[A]
): List[A] = held match {
  // Right is the current winner.
  case Left(l) => rs match {
    // ...but it's empty.
    case Nil => l :: ls
    // ...and it's still winning.
    case r :: rt if p(r, l) => r :: mergeListsWithHeld(p)(held)(ls, rt)
    // ...upset!
    case r :: rt => l :: mergeListsWithHeld(p)(Right(r))(ls, rt)
  }
  // Left is the current winner.
  case Right(r) => ls match {
    case Nil => r :: rs
    case l :: lt if p(l, r) => l :: mergeListsWithHeld(p)(held)(lt, rs)
    case l :: lt => r :: mergeListsWithHeld(p)(Left(l))(lt, rs)
  }
}

假设我们手头已经有一个失败的元素,但是现在我们可以编写我们真正想要使用的方法:

def mergeListsWith[A](p: (A, A) => Boolean)(ls: List[A], rs: List[A]): List[A] =
  ls match {
    case Nil => rs
    case l :: lt => rs match {
      case Nil => ls
      case r :: rt if p(l, r) => l :: mergeListsWithHeld(p)(Right(r))(lt, rt)
      case r :: rt            => r :: mergeListsWithHeld(p)(Left(l))(lt, rt)
    }
  }

接着:

scala> org.scalacheck.Prop.forAll { (ls: List[Int], rs: List[Int]) =>
     |   mergeListsWith[Int](_ < _)(ls.sorted, rs.sorted) == (ls ++ rs).sorted
     | }.check
+ OK, passed 100 tests.

好的,看起来不错。我们可以为列表编写更好的方法,但是这个实现与我们需要为Process.

执行

这里或多或少与 scalaz-stream 等效:

import scalaz.{ -\/, \/, \/- }
import scalaz.stream.Process.{ awaitL, awaitR, emit }
import scalaz.stream.{ Process, Tee, tee }

def mergeWithHeld[A](p: (A, A) => Boolean)(held: A \/ A): Tee[A, A, A] =
  held.fold(_ => awaitR[A], _ => awaitL[A]).awaitOption.flatMap {
    case None =>
      emit(held.merge) ++ held.fold(_ => tee.passL, _ => tee.passR)
    case Some(next) if p(next, held.merge) =>
      emit(next) ++ mergeWithHeld(p)(held)
    case Some(next) =>
      emit(held.merge) ++ mergeWithHeld(p)(
        held.fold(_ => \/-(next), _ => -\/(next))
      )
  }

def mergeWith[A](p: (A, A) => Boolean): Tee[A, A, A] =
  awaitL[A].awaitOption.flatMap {
    case None => tee.passR
    case Some(l) => awaitR[A].awaitOption.flatMap {
      case None =>               emit(l) ++ tee.passL
      case Some(r) if p(l, r) => emit(l) ++ mergeWithHeld(p)(\/-(r))
      case Some(r)            => emit(r) ++ mergeWithHeld(p)(-\/(l))
    }
  }

让我们再次检查一下:

scala> org.scalacheck.Prop.forAll { (ls: List[Int], rs: List[Int]) =>
     |   Process.emitAll(ls.sorted).tee(Process.emitAll(rs.sorted))(
     |     mergeWith(_ < _)
     |   ).toList == (ls ++ rs).sorted
     | }.check
+ OK, passed 100 tests.

如果没有更多的测试,我不会将它投入生产,但它看起来很有效。

于 2016-07-13T17:29:46.930 回答
0

正如 Travis Brown 建议的那样,您必须实现自定义 T 恤。这是我对 tee 的实现:

/*
  A tee which sequentially compares elements from left and right
  and passes an element from left if predicate returns true, otherwise
  passes an element from right.
 */
def predicateTee[A](predicate: (A, A) => Boolean): Tee[A, A, A] = {

  def go(stack: Option[A \/ A]): Tee[A, A, A] = {
    def stackEither(l: A, r: A) =
      if (predicate(l, r)) emit(l) ++ go(\/-(r).some) else emit(r) ++ go(-\/(l).some)

    stack match {
      case None =>
        awaitL[A].awaitOption.flatMap { lo =>
          awaitR[A].awaitOption.flatMap { ro =>
            (lo, ro) match {
              case (Some(l), Some(r)) => stackEither(l, r)
              case (Some(l), None) => emit(l) ++ passL
              case (None, Some(r)) => emit(r) ++ passR
              case _ => halt
            }
          }
        }
      case Some(-\/(l)) => awaitR[A].awaitOption.flatMap {
        case Some(r) => stackEither(l, r)
        case None => emit(l) ++ passL
      }
      case Some(\/-(r)) => awaitL[A].awaitOption.flatMap {
        case Some(l) => stackEither(l, r)
        case None => emit(r) ++ passR
      }
    }
  }

  go(None)
}

val p1: Process[Task, Int] = Process(1, 2, 4, 5, 9, 10, 11)
val p2: Process[Task, Int] = Process(0, 3, 7, 8, 6)

p1.tee(p2)(predicateTee(_ < _)).runLog.run
//res0: IndexedSeq[Int] = Vector(0, 1, 2, 3, 4, 5, 7, 8, 6, 9, 10, 11)
于 2017-04-30T05:09:28.117 回答