add answers to exercise 1-4 chapter 6

Author: ryblovAV

src/main/scala/org/learningconcurrency/exercises/ch6/Ex1.scala

@@ -0,0 +1,31 @@
+package org.learningconcurrency
+package exercises
+package ch6
+
+import rx.lang.scala._
+
+import org.learningconcurrency.ch2._
+
+/**
+  * Implement a custom Observable[Thread] object that emits an event when it detects that a thread was started.
+  * The implementation is allowed to miss some of the events.
+  */
+object Ex1 extends App {
+
+  val observable = Observable.apply[Thread] {
+    obs =>
+
+      val t = new Thread {
+        override def start(): Unit = {
+          super.start()
+          obs.onNext(this)
+        }
+      }
+
+      obs.onNext(t)
+      Subscription()
+  }
+
+  observable.subscribe((t) => log(t.toString()))
+
+}

src/main/scala/org/learningconcurrency/exercises/ch6/Ex2.scala

@@ -0,0 +1,34 @@
+package org.learningconcurrency
+package exercises
+package ch6
+
+import rx.lang.scala.Observable
+
+/**
+  * Implement an Observable object that emits an event every 5 seconds and every 12 seconds,
+  * but not if the elapsed time is a multiple of 30 seconds.
+  * Use functional combinators on Observable objects.
+  */
+
+import scala.concurrent.duration._
+
+object Ex2A extends App {
+
+    val a = Observable.interval(5 seconds).map(_ * 5)
+    val b = Observable.interval(12 seconds).map(_ * 12)
+
+    val c = (a merge b distinct) filter (_ % 30 != 0)
+
+    c.subscribe((s) => log(s.toString))
+
+  Thread.sleep(70000)
+}
+
+object Ex2B extends App {
+
+  val d = Observable.interval(1 seconds).filter((l) => (l % 30 != 0) && ((l % 5 == 0) || (l % 12 == 0)))
+  d.subscribe((s) => log(s.toString))
+
+  Thread.sleep(70000)
+
+}

src/main/scala/org/learningconcurrency/exercises/ch6/Ex3.scala

@@ -0,0 +1,38 @@
+package org.learningconcurrency
+package exercises
+package ch6
+
+import rx.lang.scala._
+import rx.lang.scala.Observable._
+
+import scala.annotation.tailrec
+import scala.io.Source
+import scala.util.Random
+
+/**
+  * Use the randomQuote method from this section in order to create an Observable object
+  * with the moving average of the quote lengths.
+  * Each time a new quote arrives, a new average value should be emitted.
+  */
+object Ex3 extends App {
+
+  import scala.concurrent.duration._
+
+  @tailrec
+  def randomString(length: Int, l: List[Char] = List.empty[Char]):List[Char] = {
+    if (length == 1) util.Random.nextPrintableChar :: l
+    else randomString(length-1,util.Random.nextPrintableChar :: l)
+
+  }
+
+  def randomQuoteMock = Observable.interval(1 seconds).map((l) => randomString(Random.nextInt(10)+1))
+
+  randomQuoteMock.scan((0D,0)){
+    (n,q) => n match {
+      case (s,c) => (s + q.length, c + 1)
+    }
+  }.filter(_._2 != 0).map((e) => e._1/e._2).subscribe((e) => log(s"avg = $e"))
+
+  Thread.sleep(10000)
+
+}

src/main/scala/org/learningconcurrency/exercises/ch6/Ex4.scala

@@ -0,0 +1,77 @@
+package org.learningconcurrency
+package exercises
+package ch6
+
+/**
+  Implement the reactive signal abstraction, represented with the Signal[T] type.
+
+  The Signal[T] type comes with the method apply, used to query the last event emitted by this signal,
+  and several combinators with the same semantics as the corresponding Observable methods:
+
+  class Signal[T] {
+     def apply(): T = ???
+     def map(f: T => S): Signal[S] = ???
+     def zip[S](that: Signal[S]): Signal[(T, S)] = ???
+     def scan[S](z: S)(f: (S, T) => S) = ???
+  }
+
+  Then, add the method toSignal to the Observable[T] type, which converts
+  an Observable object to a reactive signal: def toSignal: Signal[T] = ???
+
+  Consider using Rx subjects for this task.
+  */
+
+object Ex4 extends App {
+
+  import rx.lang.scala._
+
+  implicit class ObserverableAdditional[T](val self:Observable[T]) extends AnyVal {
+
+    def toSignal:Signal[T] = {
+      val s = new Signal[T]
+      self.last.subscribe(s.subject)
+      s
+    }
+
+  }
+
+  class Signal[T] {
+
+    def this(t:T) {
+      this()
+      a = t
+    }
+
+    var a:T = _
+
+    val subject = Subject[T]()
+
+    subject.subscribe(a = _)
+
+    def apply(): T = a
+
+    def map[S](f: T => S): Signal[S] = new Signal[S](f(a))
+
+    def zip[S](that: Signal[S]): Signal[(T, S)] = new Signal[(T,S)]((a,that.a))
+
+    def scan[S](z: S)(f: (S, T) => S):Signal[S] = new Signal[S](f(z,a))
+  }
+
+  //test
+  val s1 = Observable.items[String]("A","B","C").toSignal
+  log(s"element = ${s1()}")
+
+  val s2 = Observable.items[Int](1,2,3).toSignal
+
+  val sMap = s1.map(_+"~")
+  log(s"sMap: element = ${sMap()}")
+
+  val sZip = s1.zip(s2)
+  log(s"sZip: element = ${sZip()}")
+
+  val sScan = s2.scan(10)((s,t)=>s+t)
+  log(s"sScan: element = ${sScan()}")
+
+
+
+}