Как создать Iteratee, который проходит через значение на внутреннюю Iteratee, если конкретное значение не найдено

Question

У меня есть ADT, что по сути помесь Option и Try:

sealed trait Result[+T] 
case object Empty extends Result[Nothing] 
case class Error(cause: Throwable) extends Result[Nothing] 
case class Success[T](value: T) extends Result[T] 

(предположит, общие комбинатор как map , flatMap и т.д. определяются на результат)

Дано Iteratee[A, Result[B] называется inner, я хочу, чтобы создать новый Iteratee[Result[A], Result[B]] со следующим поведением:

• Если вход является Success(a), кормить a к inner
• Если вход является Empty, не оп
• Если вход является Error(err), я хочу inner быть полностью проигнорированы, а не возвращать Done итерация с результатом Error(err).

Пример Поведение:

// inner: Iteratee[Int, Result[List[Int]]] 
// inputs: 
1 
2 
3 
// output: 
Success(List(1,2,3)) 

// wrapForResultInput(inner): Iteratee[Result[Int], Result[List[Int]]] 
// inputs: 
Success(1) 
Success(2) 
Error(Exception("uh oh")) 
Success(3) 

// output: 
Error(Exception("uh oh")) 

Это звучит для меня как работа для Enumeratee, но я не смог найти что-нибудь в the docs, который выглядит, как он будет делать то, что я хочу, и внутренние реализации по-прежнему остаются вуду для меня.

Как я могу реализовать wrapForResultInput для создания описанного выше поведения?

---

Добавление еще некоторые детали, которые не будут действительно соответствовать в комментарии:

Да, похоже, я ошибся в моем вопросе. Я описал это с точки зрения Iteratees, но, похоже, я действительно искал Enumeratees.

В какой-то момент в API, который я создаю, есть класс Transformer[A], который по существу является Enumeratee[Event, Result[A]]. Я хотел бы разрешить клиентам преобразовать этот объект, предоставив Enumeratee[Result[A], Result[B]], что приведет к Transformer[B] aka Enumeratee[Event, Result[B]].

Для более сложного примера, предположим, что у меня есть Transformer[AorB] и хотите, чтобы превратить это в Transformer[(A, List[B])]:

// the Transformer[AorB] would give 
a, b, a, b, b, b, a, a, b 

// but the client wants to have 
a -> List(b), 
a -> List(b, b, b), 
a -> Nil 
a -> List(b) 

Клиент может осуществить Enumeratee[AorB, Result[(A, List[B])]] без особых проблем с помощью Enumeratee.grouped, но они необходимы, чтобы обеспечить Enumeratee[Result[AorB], Result[(A, List[B])], который, кажется, представляет собой много осложнений, которые я хотел бы скрыть от них, если это возможно.

val easyClientEnumeratee = Enumeratee.grouped[AorB]{ 
    for { 
    _ <- Enumeratee.dropWhile(_ != a) ><> Iteratee.ignore 
    headResult <- Iteratee.head.map{ Result.fromOption } 
    bs <- Enumeratee.takeWhile(_ == b) ><> Iteratee.getChunks 
} yield headResult.map{_ -> bs} 

val harderEnumeratee = ??? ><> easyClientEnumeratee 

val oldTransformer: Transformer[AorB] = ... // assume it already exists 
val newTransformer: Transformer[(A, List[B])] = oldTransformer.andThen(harderEnumeratee) 

Так что я ищу является ??? определить harderEnumeratee для того, чтобы облегчить нагрузку на пользователя, который уже реализован easyClientEnumeratee.

Я думаю ??? должен быть Enumeratee[Result[AorB], AorB], но если я пытаюсь что-то вроде

Enumeratee.collect[Result[AorB]] { 
    case Success(ab) => ab 
    case Error(err) => throw err 
} 

ошибка будет фактически брошен; Я действительно хочу, чтобы ошибка возвращалась как Error(err).

Answer 1

Простейшая реализация такого метода будет Iteratee.fold2, который мог бы собирать элементы, пока что-то не произошло.

Поскольку вы возвращаете один результат и не могу ничего возвращать, пока не убедитесь, что нет ошибок, Iteratee было бы достаточно для такой задачи

def listResults[E] = Iteratee.fold2[Result[E], Either[Throwable, List[E]]](Right(Nil)) { (state, elem) => 
    val Right(list) = state 
    val next = elem match { 
    case Empty => (Right(list), false) 
    case Success(x) => (Right(x :: list), false) 
    case Error(t) => (Left(t), true) 
    } 
    Future(next) 
} map { 
    case Right(list) => Success(list.reverse) 
    case Left(th) => Error(th) 
} 

Теперь, если мы подготовим небольшую площадку

import scala.concurrent.ExecutionContext.Implicits._ 
import scala.concurrent.{Await, Future} 
import scala.concurrent.duration._ 

val good = Enumerator.enumerate[Result[Int]](
    Seq(Success(1), Empty, Success(2), Success(3))) 

val bad = Enumerator.enumerate[Result[Int]](
    Seq(Success(1), Success(2), Error(new Exception("uh oh")), Success(3))) 

def runRes[X](e: Enumerator[Result[X]]) : Result[List[X]] = Await.result(e.run(listResults), 3 seconds) 

мы можем проверить эти результаты

runRes(good) //res0: Result[List[Int]] = Success(List(1, 2, 3)) 
runRes(bad) //res1: Result[List[Int]] = Error(java.lang.Exception: uh oh)