score:1
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)
}
Source: stackoverflow.com
Related Query
- How do I replace a program written as a sequenced stream of state transitions with scalaz-stream?
- How to access state during transitions in Akka FSM
- How to write program in Spark to replace word
- How to aggregate events in flink stream before merging with current state by reduce function?
- Scala View + Stream combo causing OutOfMemory Error. How do I replace it with a View?
- How do I read a large CSV file with Scala Stream class?
- Scala: How can I replace value in Dataframes using scala
- How do I read a parquet in PySpark written from Spark?
- How to convert a Java Stream to a Scala Stream?
- How to program in Scala to be forward-compatible with Dotty
- Akka stream 2.6. How to create ActorMaterializer?
- Spark Standalone Mode: How to compress spark output written to HDFS
- How to replace a JSON value in Play
- How to skip invalid characters in stream in Java/Scala?
- How do I replace the fork join pool for a Scala 2.9 parallel collection?
- How to check existence of a program in the path
- How do I parse an xml document as a stream using Scala?
- reader writer state monad - how to run this scala code
- How to debug a scala based Spark program on Intellij IDEA
- How to stop program execution in worksheet?
- How do you implement @BeforeClass semantics in a JUnit 4 test written in scala?
- How to instantiate and populate a Scala Stream in Java?
- How to split an inbound stream on a delimiter character using Akka Streams
- How to call main method of a Scala program from the main method of a java program?
- How can I configure IDEA to automatically replace => with ⇒ and -> with →?
- How can I publish or subscribe to a materialized Akka Stream flow graph?
- How to set breakpoints in vs code in a scala program
- How to log to an explicit AWS CloudWatch log stream and change it programmatically (Java/Scala/log4j)
- How to replace a given item in a list?
- How to run a spark example program in Intellij IDEA
More Query from same tag
- How to efficiently perform this column operation on a Spark Dataframe?
- Recursively Generate Hierarchy Data set using spark scala functional programming
- Orika class com.sun.proxy is not accessible
- Accessing class-level values of type parameters in Scala
- Drawing works fine with glDrawArrays, but is failing using glDrawElements
- Scala - how to pass implicit parameter into function (HOF)?
- Retaining mixin traits with returned value in Scala
- error when trying to download the Newman REST libraries for Eclipse via sbt
- Overloaded method value [subscribe] cannot be applied
- Spark JDBC to DashDB (DB2) with CLOB errors
- How to handle a failed future in a for-comprehension
- How to tell to the compiler a function can only return subtype of a trait and not the trait itself?
- How to write Aggregate pattern in Scala?
- Json4s casting error during string parsing
- Creating a type-sensitive function without changing the parent trait or case classes
- JSON (String?) value prints with quotes (" ")
- How can I match constructor types with Java reflection with Scala primitive types?
- How to parse this "pretty" BSON in Scala?
- What's a shebang line for Scala that doesn't corrupt mimetype?
- Scala Swing image
- Is there a sequence function for Try objects in scalaz?
- Spark 1.5.2 Shuffle/Serialization - running out of memory
- Parsing scalaz.Validation in Java
- Conditional slicing in Scala Breeze
- How to Enable HSTS in Play framework 2.3.x using scala code?
- Scala java.io toArray error when zipping feature importance vector to column names array
- What is the best way to use multiple akka scheduler each calling different methods?
- Seq[AnyVal] returned as type instead of Seq[Int]
- Why does this wildcarded function tells me it has the wrong number of parameters?
- Scala fails to run compiled class, but runs it directly