A purely functional dataflow graph using Arrows

Dataflow.scala

/**
 * A purely functional dataflow graph using Arrows.
 * Note that the groupBy compbinator is built from the accum primitive. 
 * @author willtim
 */
object Dataflow {
  
  case class Auto[A,B] (run: A => (B, Auto[A,B])) { f =>
    
    def >>> [C] (g: Auto[B,C]): Auto[A,C] = 
      Auto[A,C]( a => {
        val (b, f2) = f.run(a)
        val (c, g2) = g.run(b)
        (c, f2 >>> g2)
      })
  }
  
  def unit[A,B] (f: A => B): Auto[A,B] = Auto( a => (f(a), unit(f)) )
  
  def run[A,B] (a: Auto[A,B], src: Iterable[A]) =
    new Iterable[B] {
      def iterator = new Iterator[B] {
        val i = src.iterator
        var s = a
        def next: B =  {
          var (b, s2) = s.run(i.next())
          s = s2
          b
        }
        def hasNext: Boolean = i.hasNext
      }
    }
  
  def accum[A,B] (acc: B, f: (A, B) => B): Auto[A,B] =
      Auto[A,B]( a => {
        val acc2 = f(a, acc)
        (acc2, accum(acc2, f))
      })
      
  def groupBy[K,V] (f: (V,V)=>V): Auto[(K,V), Map[K,V]] = accum(Map(), {
    case ( (k, v), m ) =>
      m get k match {
        case None => m + (k -> v)
        case Some(v0) => m + (k -> f(v0,v))
      }})

  def main(args: Array[String]) {
    var g : Auto[(String,Int),Map[String,Int]] = groupBy(_+_)
    println(run(g, List(("a",1),("b",2),("c",3),("a",4),("b",5))))
  }
}