Scala reactive programming FRP

1.generators.sc

object generator {

	// Coursera Reactive programming
		
	trait Generator[T] { self =>
		def generate: T
		def map[K](f: T => K) = new Generator[K] {
//			{val sample = generate ; println("map creates a generator [" + sample.getClass().getSimpleName + "]") }
			def generate: K = f(self.generate)
		}
		def flatMap[G](f: T => Generator[G]) = new Generator[G] {
//			{val sample = generate ; println("flatmap creates a generator [" + sample.getClass().getSimpleName + "]") }
			def generate: G = f(self.generate).generate
		}
		
		def sample(len: Int = 3) = (1 to len) map (_ => generate) mkString ", "
		def print(len: Int) = (1 to len) foreach (_ => println (generate))
	}
	
	val int = new Generator[Int] {
		val core = new java.util.Random
		def generate: Int = core.nextInt()
	}                                         //> int  : generator.Generator[Int]{val core: java.util.Random} = generator$$ano
                                                  //| nfun$main$1$$anon$3@1e2fe5d
	val bool = int map (_ > 0)                //> bool  : generator.Generator[Boolean] = generator$Generator$$anon$1@1590164
	bool sample()                             //> res0: String = true, false, false
	def pair[T,G](t: Generator[T], g: Generator[G]) = new Generator [(T,G)]{
		def generate = (t.generate, g.generate)
	}                                         //> pair: [T, G](t: generator.Generator[T], g: generator.Generator[G])generator.
                                                  //| Generator[(T, G)]
	pair(int, bool) sample ()                 //> res1: String = (988902526,false), (-1544701772,true), (911260777,false)
	def pair2[T,G](t: Generator[T], g: Generator[G]) = t.map(t => (t, g.generate))
                                                  //> pair2: [T, G](t: generator.Generator[T], g: generator.Generator[G])generato
                                                  //| r.Generator[(T, G)]
	
	def pair3[T,G](t: Generator[T], g: Generator[G]) = new Generator[(T, G)] {
		def generate = (t.generate, g.generate)
	}                                         //> pair3: [T, G](t: generator.Generator[T], g: generator.Generator[G])generato
                                                  //| r.Generator[(T, G)]
	
	pair(int, bool) sample ()                 //> res2: String = (322687537,false), (1246101362,false), (187067389,true)
	pair2(int, bool) sample ()                //> res3: String = (1844709653,true), (287530962,false), (627834111,false)
	pair3(int, bool) sample ()                //> res4: String = (1580499174,false), (721355692,false), (-1884957912,true)
	def pair4[T,G](t: Generator[T], g: Generator[G]) = t.map (t => g.map (g => (t,g)).generate)
                                                  //> pair4: [T, G](t: generator.Generator[T], g: generator.Generator[G])generato
                                                  //| r.Generator[(T, G)]
	pair4(int, bool) sample ()                //> res5: String = (-454320814,true), (367096520,false), (-1026375939,true)

	def pair44[T,G](t: Generator[T], g: Generator[G]): Generator[(T,G)] = t.flatMap{t => g.map (g => (t,g))}
                                                  //> pair44: [T, G](t: generator.Generator[T], g: generator.Generator[G])generat
                                                  //| or.Generator[(T, G)]
	
	def pair5[T,G](t: Generator[T], g: Generator[G]): Generator[(T,G)] = for (t <- t; g <- g) yield (t,g)
                                                  //> pair5: [T, G](t: generator.Generator[T], g: generator.Generator[G])generato
                                                  //| r.Generator[(T, G)]
	pair5(int, bool) sample ()                //> res6: String = (1511780349,false), (-1501618693,true), (377758466,true)
	
	def const[T](value: T) = new Generator[T]{def generate = value}
                                                  //> const: [T](value: T)generator.Generator[T]
	const(3) sample ()                        //> res7: String = 3, 3, 3
	
	def range(from: Int, to: Int) = int map {i => from + i.abs % (to - from)}
                                                  //> range: (from: Int, to: Int)generator.Generator[Int]
	range(1,5) sample 100                     //> res8: String = 2, 1, 1, 4, 2, 4, 4, 1, 4, 3, 3, 4, 3, 2, 1, 1, 1, 4, 1, 1, 
                                                  //| 3, 2, 1, 3, 4, 4, 3, 2, 2, 2, 4, 1, 4, 1, 1, 1, 1, 3, 3, 4, 1, 1, 1, 1, 2, 
                                                  //| 2, 3, 4, 2, 3, 4, 1, 1, 2, 1, 4, 4, 3, 3, 1, 4, 3, 2, 4, 3, 4, 2, 3, 1, 4, 
                                                  //| 3, 4, 1, 3, 4, 4, 4, 4, 3, 1, 1, 2, 2, 3, 2, 3, 3, 4, 2, 4, 1, 4, 2, 4, 3, 
                                                  //| 2, 3, 1, 2, 4
	
	def oneOf[T](values: T*) = range (0, values.length) map {values(_)}
                                                  //> oneOf: [T](values: T*)generator.Generator[T]
	oneOf(1,2,3) sample 10                    //> res9: String = 1, 1, 3, 3, 3, 3, 2, 2, 2, 1
	
	
	def list1[T](g: Generator[T]): Generator[List[T]] = bool map {stop => if (stop) Nil else g.generate :: (list1(g).generate)}
                                                  //> list1: [T](g: generator.Generator[T])generator.Generator[List[T]]

	def list2[T](g: Generator[T]): Generator[List[T]] = for (stop <- bool) yield if (stop) Nil else g.generate :: list2(g).generate
                                                  //> list2: [T](g: generator.Generator[T])generator.Generator[List[T]]

	def list3[T](g: Generator[T]): Generator[List[T]] = bool flatMap{stop => if (stop) const(Nil) else for (head <- g ; tail <- list3(g)) yield head :: tail}
                                                  //> list3: [T](g: generator.Generator[T])generator.Generator[List[T]]
	def list4[T](g: Generator[T]): Generator[List[T]] = bool flatMap{stop => if (stop) const(Nil) else g flatMap {head => list4(g) map {tail => head :: tail}}}
                                                  //> list4: [T](g: generator.Generator[T])generator.Generator[List[T]]

	list1(int).sample(10)                     //> res10: String = List(-400691630), List(-1539958630, 735132745), List(-12467
                                                  //| 94552), List(1798394896, 1859557281), List(-1863537047, 638280036, 41155254
                                                  //| 7, 1974048976), List(), List(), List(-2091385359, 1184949998, 1824524305), 
                                                  //| List(248717654, -2000541436), List(-139412964)
	list4(int) sample 10                      //> res11: String = List(), List(), List(), List(874658464, 834967309), List(),
                                                  //|  List(), List(1715631146), List(), List(), List(1427562412, 1526008939, 118
                                                  //| 2320200)
	
	trait Tree[T] {
		def and(that: Tree[T]) = Inner(this, that)
	}
	case class Inner[T](left: Tree[T], right: Tree[T]) extends Tree[T]
	case class Leaf[T](value: T) extends Tree[T]
	
	def tree[T](t: Generator[T]): Generator[Tree[T]] = bool map {stop => if (stop) Leaf(t.generate) else Inner(tree(t).generate, tree(t).generate)}
                                                  //> tree: [T](t: generator.Generator[T])generator.Generator[generator.Tree[T]]
                                                  //| 

	implicit def leaf[T](t: T) = Leaf(t)      //> leaf: [T](t: T)generator.Leaf[T]
	
	val t1 = Leaf(1) and Leaf(2)              //> java.lang.ClassFormatError: Duplicate field name&signature in class file ge
                                                  //| nerator$$anonfun$main$1$Leaf$3
                                                  //| 	at java.lang.ClassLoader.defineClass1(Native Method)
                                                  //| 	at java.lang.ClassLoader.defineClass(Unknown Source)
                                                  //| 	at java.security.SecureClassLoader.defineClass(Unknown Source)
                                                  //| 	at java.net.URLClassLoader.defineClass(Unknown Source)
                                                  //| 	at java.net.URLClassLoader.access$100(Unknown Source)
                                                  //| 	at java.net.URLClassLoader$1.run(Unknown Source)
                                                  //| 	at java.net.URLClassLoader$1.run(Unknown Source)
                                                  //| 	at java.security.AccessController.doPrivileged(Native Method)
                                                  //| 	at java.net.URLClassLoader.findClass(Unknown Source)
                                                  //| 	at java.lang.ClassLoader.loadClass(Unknown Source)
                                                  //| 	at sun.misc.Launcher$AppClassLoader.loadClass(Unknown Source)
                                                  //| 	at java.lang.ClassLoader.loadClass(Unknown Source)
                                                  //| 	at generator$$anonfun$main$1$Leaf$4$.apply(generator.scala:71)
                                                  //| 	at generator$$anonfun$main$1.apply$mcV$sp(generator.scala:77)
	Leaf(1) and 2

	val t2 = (Leaf(1) and 2) and (Leaf(3) and (Leaf(8) and 9))
	
	tree(int) sample (10)
	
}

2.DiscreetEventSimulation_and_Signals.sc

// Please note how smartly I both stimulate and observe a wire using
//
//	val a = Wire() probe "a" waveform (1 -> true, 10 -> false, 11 -> true) ;
//	inv_ (a) probe ("not a")
//
// and compose gates
//	inv_(or_(inv_(a), inv_(b))) probe "composed a and b"

object FRP_Signal extends App {

	//Course era reactive programming 002
  println("Part 1: Discrete Event Simulation")
  
	import DiscreteEventSimulation._
	import Netlist._
	import simulation._
	val a = Wire() probe "a" waveform (1 -> true, 10 -> false, 11 -> true) ;
	inv (a, Wire() probe ("not a"))
	inv_(a ) probe ("not_a")
	run()
	
	
	val b = (Wire() probe "b"). waveform (1 -> true, 2 -> false)
	and (a, b, Wire() probe "a and  b")
	and_(a, b) probe "a and_ b"
	inv_(or_(inv_(a), inv_(b))) probe "composed a and b"
	
	run()
	
  println("Part 2: Signals")
	import signals._
	val s1 = Var(1) ; val s2 = Var(s1() + 1) ; val s3 = Var(s2() + 1)
	s3()
  //s1() = s1() // can be captured by mine caller.value != this
	//s1() = s2() // can be captured by oderski's caller.value.observers.contains(this)
	//s1() = s3() // stack overflow because nobody can capture this


	
	
}

//////      P A R T    I   --- Discrete event simulation
object DiscreteEventSimulation {
	type Action = () => Unit

	val simulation = new Simulation{}
	object Netlist {
		
		case class Wire() {
			var value = false
			var actions: List[Action] = Nil
			def update(newVal: Boolean) = if (value != newVal) {
				value = newVal ; actions foreach {_()}
			}
			def register(a: Action) {actions ::= a ; a()}
			def reg(time: Int, expr: => Boolean, inputs: Wire*) {
				inputs foreach {_.register(() => {
					val value = expr
					simulation.after(time, () => update(value))
				})}
				
			}
			
			//additional methods
			def probe(name: String): Wire = {register(() => println(simulation.curTime + ": " + name + " => " + value)) ; this }
			def waveform(events: (Int, Boolean)*): Wire = {events foreach {case (time, value) => simulation.after(time, () => update(value))} ; this}
		}
		
		def inv(i: Wire, o: Wire) = o.reg(1, !i.value, i)
		def and(a: Wire, b: Wire, o: Wire) = o.reg(2, a.value && b.value, a, b)
		def  or(a: Wire, b: Wire, o: Wire) = o.reg(2, a.value || b.value, a, b)
		def HA(a: Wire, b: Wire, s: Wire, c: Wire) = {
			//val d, e, c = Wire() ; inv(c, e)
			//and(a,b, c) ; and(d,e, s)
			val d, e, c = Wire() ; inv(c, e)
			and(a,b, c) ; and(d,inv_(c), s)
		}
		def fa(a: Wire, b: Wire, cin: Wire, sum: Wire, cout: Wire) {
			val s, c1, c2 = Wire()
			HA(b, cin, s, c1) ; HA(a, s, sum, c2) ; or(c1, c2, cout)
		}

	
		// convinice method that creates output wire itself
		def gate(time: Int, func: Seq[Boolean] => () => Boolean, inputs: Wire*): Wire = {
			def expr = func{inputs map (_.value)}()
			val o = Wire() ; o.reg(time, expr, inputs.toList: _*) ;
			o
		}
		
		def inv_(i: Wire): Wire = gate(1, (value) => {() => !value(0)}, i)
		
		def gate2(op: (Boolean, Boolean) => Boolean, identity: Boolean, inputs: Wire*): Wire =
			gate(2, (values) => {() => values.foldLeft(identity){op}}, inputs: _*)
		def and_(inputs: Wire*): Wire = gate2((a,b) => a && b, true, inputs: _*)
		def or_(inputs: Wire*): Wire = gate2((a,b) => a || b, false, inputs: _*)
		
	}
	                                                  
	trait Simulation {
	
		case class Event(time: Int, action: Action)
		type Agenda = List[Event]
		
		var curTime: Int = 0
		var agenda: Agenda = Nil
		def run() {
			after(0, () => println(" * * * * *     Simulation started at " + curTime + "     * * *  *"))
			loop();
		}
		
	def after(time: Int, action: Action) {
		def insert(event: Event, agenda: Agenda): Agenda = agenda match {
			case head :: tail if event.time > head.time => head :: insert(event, tail)
			case _ =>  event :: agenda
		}
		agenda = insert(Event(time + curTime, action), agenda)
	}
	
	protected def loop(): Unit = agenda match {
			case Event(t, action) :: tail =>
				curTime = t ;	agenda = tail ;
				action() ; loop()
			case Nil =>
		}
		
	}
}



////////       P A R T     II
class Stack[T](init: T) {
	private var stack = List(init)
	def value = stack.head
	def withValue[R](newValue: T)(op: => R): R = {
		stack ::= newValue
		try op finally stack = stack.tail
	}
}

object signals {
	
	class Signal[T](initExpr : => T) {
		import Signal._
		
		var expr: () => T = _
		var value: T = _
		var observers: Set[Signal[_]] = Set()
		protected def update(newExpr: => T) = {
			expr = () => newExpr;
			computeValue()
		}
		def computeValue() {
			val newValue = caller.withValue(this)(expr())
			if (value != newValue) {
				value = newValue ;
				val leaving = observers ; observers = Set()
				leaving foreach {_.computeValue()}
			}
		}
		update(initExpr)
		def apply() = {
			observers += caller.value
			
			assert(
				//caller.value != this
				!caller.value.observers.contains(this) // Odersky orignal. I think that neither is good because there can be indirect circle.
				, "signal is going to observe itself!")
			value
		}
	}
	
	object NoSignal extends Signal[Nothing](???) {
		override def computeValue() {}
	}
	
	object Signal {
		val caller = new Stack[Signal[_]](NoSignal)
		def apply[T](expr: => T) = new Signal[T](expr)
	}
	
	
	class Var[T](expr: => T) extends Signal[T](expr) {
		override def update(expr: => T) = super.update(expr)
	}
	object Var {
		def apply[T](expr: => T) = new Var[T](expr)
	}
	
}

5.actor_based_tree.sc

// This is optimized (more concise) version of http://alexminnaar.com/building-a-distributed-binary-search-tree-with-akka.html

package actorbintree

import akka.actor._
import scala.collection.immutable.Queue

object BinaryTreeSet {

  trait Operation { def requester: ActorRef; def id: Int; def elem: Int}

  trait OperationReply { def id: Int }

  /** Request with identifier `id` to insert an element `elem` into the tree.
    * The actor at reference `requester` should be notified when this operation
    * is completed.
    */
  case class Insert(requester: ActorRef, id: Int, elem: Int) extends Operation

  /** Request with identifier `id` to check whether an element `elem` is present
    * in the tree. The actor at reference `requester` should be notified when
    * this operation is completed.
    */
  case class Contains(requester: ActorRef, id: Int, elem: Int) extends Operation

  /** Request with identifier `id` to remove the element `elem` from the tree.
    * The actor at reference `requester` should be notified when this operation
    * is completed.
    */
  case class Remove(requester: ActorRef, id: Int, elem: Int) extends Operation

  /** Request to perform garbage collection*/
  case object GC

  /** Holds the answer to the Contains request with identifier `id`.
    * `result` is true if and only if the element is present in the tree.
    */
  case class ContainsResult(id: Int, result: Boolean) extends OperationReply
  
  /** Message to signal successful completion of an insert or remove operation. */
  case class OperationFinished(id: Int) extends OperationReply

}

abstract class InsertingActor extends Actor {
  def props(elem: Int, initiallyRemoved: Boolean = false) = {
    val result = context.actorOf(Props(classOf[BinaryTreeNode],  elem, initiallyRemoved))
    println("creating " + result + " => " + elem) ; result
  }
  
}
class BinaryTreeSet extends InsertingActor {
  import BinaryTreeSet._, BinaryTreeNode._

  def createRoot = props(0, initiallyRemoved = true)

  var root: ActorRef = createRoot

  // optional
  var pendingQueue = Queue.empty[Operation]

  // optional
  /** Accepts `Operation` and `GC` messages. */
  def receive: Receive = {
    case GC => 
      val newRoot = createRoot ; root ! CopyTo(newRoot)
      context.become(garbageCollecting(newRoot))
    case o: Operation => root.tell(o, sender)
  }
  
  // optional
  /** Handles messages while garbage collection is performed.
    * `newRoot` is the root of the new binary tree where we want to copy
    * all non-removed elements into.
    */
  def garbageCollecting(newRoot: ActorRef): Receive = {
    case o: Operation => pendingQueue = pendingQueue.enqueue(o)
    case CopyFinished => 
      root = newRoot ; pendingQueue foreach {op => println("replying " + op + " to new root") ; root ! op} 
      context.become(receive) ; pendingQueue = Queue.empty
  }

}

object BinaryTreeNode {
  trait Position

  case object Left extends Position
  case object Right extends Position

  case class CopyTo(treeNode: ActorRef)
  case object CopyFinished

}

class BinaryTreeNode(val elem: Int, initiallyRemoved: Boolean) extends InsertingActor {
  import BinaryTreeNode._
  import BinaryTreeSet._
  
  def pos(value: Int) = if (value < elem) Left else Right
  def child(value: Int) = children.get(pos(value))
  var children = scala.collection.mutable.Map[Position, ActorRef]()
  var removed = initiallyRemoved

  def treesome(msg: Operation, me: => Unit, none: => Unit) {
    if (elem == msg.elem) me else child(msg.elem) match {
      case Some(child) => println(elem + " forwarding " + msg + " to " + child) ; child ! msg
      case None => none
    }
  }
  
  // optional
  /** Handles `Operation` messages and `CopyTo` requests. */
  def receive: Receive = {
    case msg @ Contains(req, id, value) => 
      def reply(result: Boolean) = {println(elem + " says that " + msg + " = " + result); req ! ContainsResult(id, result)}
      treesome(msg, reply(!removed), reply(false))
      
    case msg @ Insert(req, id, value) => 
      def reply = {println(elem + " finished " + msg); req ! OperationFinished(id)}
      treesome(msg, {removed = false ; reply}, {children(pos(value)) = props(value, false) ; reply})
      
    case msg @ Remove(req, id, value) => 
      def reply = {println(elem + " finished " + msg); req ! OperationFinished(id)}
      treesome(msg, {removed = true ; reply}, reply)
    
    
    case CopyTo(newRoot) => 
      println(elem + " copy")
      if (!removed) newRoot ! Insert(self, -1, elem)
      children.values foreach {_ ! CopyTo(newRoot)}
      if (children.isEmpty) sender ! CopyFinished 
      else context.become(copying(children.values.toSet, sender))
      
  }

  // optional
  /** `expected` is the set of ActorRefs whose replies we are waiting for,
    * `insertConfirmed` tracks whether the copy of this node to the new tree has been confirmed.
    */
  def copying(expected: Set[ActorRef], originator: ActorRef): Receive = {
    case CopyFinished => 
      val subtrees = expected - sender
      if (subtrees.isEmpty)  {originator ! CopyFinished ; context.become(receive)} 
      else context.become(copying(subtrees, originator))
    case Insert => println(self + " duplicate complete")
  }


}