How do I replace a program written as a sequenced stream of state transitions with scalaz-stream?

I'm trying to understand how to reorganize a program which I would previously have written as a sequence of state transitions:

I have some business logic:

type In = Long
type Count = Int
type Out = Count
type S = Map[Int, Count]


val inputToIn: String => Option[In]
= s => try Some(s.toLong) catch { case _ : Throwable => None }


def transition(in: In): S => (S, Out)
= s => { val n = s.getOrElse(in, 0); (s + (in -> n+1), n+1) }


val ZeroOut: Out = 0
val InitialState: S = Map.empty

With these I wish to construct a program to pass in some initial State (an empty Map), read input from stdin, convert it to In, run the state transition and output the current state S and the output Out to stdout.

Previously, I would have done something like this:

val runOnce = StateT[IO, S, Out](s => IO.readLn.map(inputToIn) flatMap {
case None     => IO((s, ZeroOut))
case Some(in) => val (t, o) = transition(in)(s)
IO.putStrLn(t.toString) |+| IO.putStrLn(o.toString) >| IO((t, o))
})


Stream.continually(runOnce).sequenceU.eval(InitialState)

However, I'm really struggling to see how to connect this approach (a stream of state transitions) with scalaz-stream. I started with this:

type Transition = S => (S, Out)
val NoTransition: Transition = s => (s, 0)


io.stdInLines.map(inputToIn).map(_.fold(NoTransition)(transition))

This is of type: Process[Task, Transition]. I don't really know where to go from there.

  1. How do I "pass in" my InitialState and run the program, threading in the output S at each step as the input S to the next one?
  2. How do I get the values of S and Out at each step and print them to stdout (assuming I can convert them to strings)?

In trying to use a single for-comprehension, I get similarly stuck:

for {
i <- Process.eval(Task.now(InitialState))
l <- io.stdInLines.map(inputToIn)
...

Any help is greatly appreciated!

I've got a bit further now. 

type In_ = (S, Option[In])
type Out_ = (S, Out)


val input: Process[Task, In_]
= for  {
i <- Process.emit(InitialState)
o <- io.stdInLines.map(inputToIn)
} yield (i, o)


val prog =
input.pipe(process1.collect[In_, Out_]) {
case (s, Some(in)) => transition(in)(s)
}).to(io.stdOutLines.contramap[Out_](_.toString))

Then

prog.run.run

It doesn't work: It seems like the state is not being threaded through the stream. Rather, at each stage, the initial state is being passed in.

Paul Chiusano suggested using the approach of process1.scan. So now I do this:

type In_  = In
type Out_ = (S, Out)


val InitialOut_ = (InitialState, ZeroOut)


val program =
io.stdInLines.collect(Function.unlift(inputToIn)).pipe(
process1.scan[In_, Out_](InitialOut_) {
case ((s, _), in) => transition(in)(s)
}).to(io.stdOutLines.contramap[Out_](_.shows))

There's a problem here: In this specific example, my Out type is a monoid, so my initial state can be created using its identity but this may not generally be the case. What would I do then? (I guess I could use Option but this seems like it's unnecessary.)

1447 次浏览
小开 
import io.FilePath


import scalaz.stream._
import Process._
import scalaz.concurrent.Task
import Task._
import scalaz.{Show, Reducer, Monoid}
import scalaz.std.list._
import scalaz.syntax.foldable._
import scalaz.syntax.bind._
import scalaz.stream._
import io._
import scalaz.stream.text._
import Processes._
import process1.lift
import control.Functions._


/**
* A Fold[T] can be used to pass over a Process[Task, T].
*
* It has:
*
*  - accumulation, with an initial state, of type S, a fold action and an action to perform with the last state
*
*  - side-effects with a Sink[Task, (T, S)] to write to a file for example, using the current element in the Process
*    and the current accumulated state
*
* This covers many of the needs of iterating over a Scalaz stream and is composable because there is a Monoid
* instance for Folds
*
*/
trait Fold[T] {
type S


def prepare: Task[Unit]
def sink: Sink[Task, (T, S)]
def fold: (T, S) => S
def init: S
def last(s: S): Task[Unit]


/** create a Process1 returning the state values */
def foldState1: Process1[T, S] =
Processes.foldState1(fold)(init)


/** create a Process1 returning the folded elements and the state values */
def zipWithState1: Process1[T, (T, S)] =
Processes.zipWithState1(fold)(init)


}


/**
* Fold functions and typeclasses
*/
object Fold {


/**
* Create a Fold from a Sink with no accumulation
*/
def fromSink[T](aSink: Sink[Task, T]) =  new Fold[T] {
type S = Unit
lazy val sink: Sink[Task, (T, S)] = aSink.extend[S]


def prepare = Task.now(())
def fold = (t: T, u: Unit) => u
def init = ()
def last(u: Unit) = Task.now(u)
}


/**
* Transform a simple sink where the written value doesn't depend on the
* current state into a sink where the current state is passed all the time
* (and actually ignored)
* Create a Fold a State function
*/
def fromState[T, S1](state: (T, S1) => S1)(initial: S1) = new Fold[T] {
type S = S1
lazy val sink: Sink[Task, (T, S)] = unitSink[T, S]


def prepare = Task.now(())
def fold = state
def init = initial
def last(s: S) = Task.now(())
}


/**
* Create a Fold from a side-effecting function
*/
def fromFunction[T](f: T => Task[Unit]): Fold[T] =
fromSink(Process.constant(f))


/**
* Create a Fold from a Reducer
*/
def fromReducer[T, S1](reducer: Reducer[T, S1]): Fold[T] = new Fold[T] {
type S = S1
lazy val sink: Sink[Task, (T, S)] = unitSink[T, S]


def prepare = Task.now(())
def fold = reducer.cons
def init = reducer.monoid.zero
def last(s: S) = Task.now(())
}


/**
* Create a Fold from a Reducer and a last action
*/
def fromReducerAndLast[T, S1](reducer: Reducer[T, S1], lastTask: S1 => Task[Unit]): Fold[T] = new Fold[T] {
type S = S1
lazy val sink: Sink[Task, (T, S)] = unitSink[T, S]


def prepare = Task.now(())
def fold = reducer.cons
def init = reducer.monoid.zero
def last(s: S) = lastTask(s)
}


/**
* This Sink doesn't do anything
* It can be used to build a Fold that does accumulation only
*/
def unitSink[T, S]: Sink[Task, (T, S)] =
channel((tu: (T, S)) => Task.now(()))


/**
* Unit Fold with no side-effect or accumulation
*/
def unit[T] = fromSink(channel((t: T) => Task.now(())))


/**
* Unit fold function
*/
def unitFoldFunction[T]: (T, Unit) => Unit = (t: T, u: Unit) => u


/** create a fold sink to output lines to a file */
def showToFilePath[T : Show, S](path: FilePath): Sink[Task, (T, S)] =
io.fileChunkW(path.path).pipeIn(lift(Show[T].shows) |> utf8Encode).extend[S]


implicit class FoldOps[T](val fold: Fold[T]) {
}


/**
* Monoid for Folds, where effects are sequenced
*/
implicit def foldMonoid[T]: Monoid[Fold[T]] = new Monoid[Fold[T]] {
def append(f1: Fold[T], f2: =>Fold[T]): Fold[T] = f1 >> f2
lazy val zero = Fold.unit[T]
}


/**
* create a new Fold sequencing the effects of 2 Folds
*/
implicit class sequenceFolds[T](val fold1: Fold[T]) {
def >>(fold2: Fold[T]) = new Fold[T] {
type S = (fold1.S, fold2.S)


def prepare = fold1.prepare >> fold2.prepare


def sink = fold1.sink.zipWith(fold2.sink) { (f1: ((T, fold1.S)) => Task[Unit], f2: ((T, fold2.S)) => Task[Unit]) =>
(ts: (T, S)) => {
val (t, (s1, s2)) = ts
(f1((t, s1)) |@| f2((t, s2)))((_,_))
}
}


def fold = (t : T, s12: (fold1.S, fold2.S)) => (fold1.fold(t, s12._1), fold2.fold(t, s12._2))
def last(s12: (fold1.S, fold2.S)) = (fold1.last(s12._1) |@| fold2.last(s12._2))((_,_))
def init = (fold1.init, fold2.init)
}
}


/**
* Run a fold an return the last value
*/
def runFoldLast[T](process: Process[Task, T], fold: Fold[T]): Task[fold.S] =
fold.prepare >>
logged(process |> fold.zipWithState1).drainW(fold.sink).map(_._2).runLastOr(fold.init)


/**
* Run a Fold an let it perform a last action with the accumulated state
*/
def runFold[T](process: Process[Task, T], fold: Fold[T]): Task[Unit] =
runFoldLast(process, fold).flatMap(fold.last)


/**
* Run a list of Folds, sequenced with the Fold Monoid
*/
def runFolds[T](process: Process[Task, T], folds: List[Fold[T]]): Task[Unit] =
runFold(process, folds.suml)


}

提交答案