From 5748b8272acbf67356cd7466037c48df6bc95f6b Mon Sep 17 00:00:00 2001
From: Pascal <pascal@pdev.dev>
Date: Wed, 29 Jan 2020 01:19:10 -0600
Subject: [PATCH] [mod] modified broadcastobject to mirror a real live
 broadcast better  - allow to watch until a specific or later frame has been
 cast  - allow to return from watching if the broadcast is ended  - allow to
 reset the broadcast to be reset  - in retrospect this wasnt really needed for
 the project...

---
 src/main/scala/scim/lib/Concurrency.scala | 49 +++++++++++++++++++++--
 1 file changed, 46 insertions(+), 3 deletions(-)

diff --git a/src/main/scala/scim/lib/Concurrency.scala b/src/main/scala/scim/lib/Concurrency.scala
index caa90a4..da7bc13 100644
--- a/src/main/scala/scim/lib/Concurrency.scala
+++ b/src/main/scala/scim/lib/Concurrency.scala
@@ -35,6 +35,7 @@ object ScalaLocks {
     }
   }
 
+  // The conditional readlock waits for a condition to be met, before executing the provided read function
   class ScalaConditionalReadLock(readLock: ReentrantReadWriteLock.ReadLock, writeLock: ReentrantReadWriteLock.WriteLock, signal: Condition) {
     private val scalaWriteLock = new ScalaWriteLock(writeLock)
 
@@ -62,7 +63,7 @@ object ScalaLocks {
 
 object ConcurrentBroadcast {
 
-  class BroadcastObject[A]() {
+  class BroadcastObject[A](default: Option[A] = None) {
 
     import ScalaLocks._
 
@@ -76,12 +77,15 @@ object ConcurrentBroadcast {
     private val writeLock = new ScalaWriteLock(javaWriteLock)
     private val readLock = new ScalaReadLock(javaReadLock)
 
-    private var value: Option[A] = None
+    private var frame: Int = 0 // should be a large enough buffer
+    private var broadcastEnded = false
+    private var value: Option[A] = default
 
     // NON-BLOCKING: check if value has been set, write new value and return if it was overwritten
     def broadcast(newValue: A): Unit = {
       writeLock(() => {
         value = Some(newValue)
+        frame += 1
         signal.signalAll()
       })
     }
@@ -91,23 +95,62 @@ object ConcurrentBroadcast {
       readWriteLock(() => {
         val result = value.isEmpty
         value = Some(newValue)
+        frame += 1
         signal.signalAll()
         result
       })
     }
 
+    def broadcastStop(): Unit = {
+      writeLock(() => {
+        value = None // this indicates the watcher that the broadcast ended
+        broadcastEnded = true
+        signal.signal()
+      })
+    }
+
+    def broadcastRestart(): Unit = {
+      writeLock(() => {
+        broadcastEnded = false
+      })
+    }
+
+    def broadcastReset(): Unit = {
+      broadcastEnded = false
+      frame = 0
+      value = default
+    }
+
     // BLOCKING: watch value, block until it is set, then return it
     def watch(): A = {
       conditionalReadLock(!value.isEmpty, value.get)
     }
 
+    // BLOCKING: watch value, block until it is set or the broadcast ends
+    // returns None for broadcast end and Some[A] otherwise
+    def watchUntilEnd(): Option[A] = {
+      conditionalReadLock(!value.isEmpty || broadcastEnded, value)
+    }
+
+    // BLOCKING: watch value from frame, block until it is set, then return it
+    // IMPORTANT: no guarantee for the frame, a later one may be returned
+    def watchFrame(requestedFrame: Int): (A, Int) = {
+      conditionalReadLock(!value.isEmpty && frame >= requestedFrame, (value.get, frame))
+    }
+
+    // BLOCKING: watch value from frame, block until it is set or the broadcast ends
+    // returns None for broadcast end and Some[A] otherwise
+    // IMPORTANT: no guarantee for the frame, a later one may be returned
+    def watchFrameUntilEnd(requestedFrame: Int): (Option[A], Int) = {
+      conditionalReadLock((!value.isEmpty && frame >= requestedFrame) || broadcastEnded, (value, frame))
+    }
+
     // NON-BLOCKING: returns value as Option
     def peak(): Option[A] = {
       readLock {
         value
       }
     }
-
   }
 
   abstract class Broadcast[A](stream: mutable.ArraySeq[BroadcastObject[A]]) extends Runnable {