Add AsyncCallback.readWriteMutex

Closes #816

Author: japgolly

core/src/main/scala/japgolly/scalajs/react/AsyncCallback.scala

@@ -365,6 +365,88 @@ object AsyncCallback {
   /** Creates a new (non-reentrant) mutex. */
   def mutex: CallbackTo[Mutex] =
     CallbackTo(new Mutex)
+
+  // ===================================================================================================================
+
+  final class ReadWriteMutex private[AsyncCallback]() {
+
+    // Whether it's a read or write mutex is determined by readers being > 0 or not
+    private var mutex: Option[AsyncCallback.Barrier] =
+      None
+
+    private var readers =
+      0
+
+    private val releaseMutex: Callback =
+      CallbackTo {
+        if (readers == 0) {
+          val old = mutex
+          mutex = None
+          old
+        } else
+          None
+      }.flatMap(Callback.traverseOption(_)(_.complete))
+
+    private val releaseReader: Callback =
+      CallbackTo {
+        readers -= 1
+      } >> releaseMutex
+
+    /** Wrap a [[AsyncCallback]] so that it executes in the write-mutex.
+      * There can only be one writer active at one time.
+      *
+      * Note: THIS IS NOT RE-ENTRANT. Calling this from within the read or write mutex will block.
+      */
+    def write[A](ac: AsyncCallback[A]): AsyncCallback[A] =
+      AsyncCallback.byName {
+
+        mutex match {
+          case None =>
+            // Mutex empty
+            val b = AsyncCallback.barrier.runNow()
+            mutex = Some(b)
+            ac.finallyRunSync(releaseMutex)
+
+          case Some(b) =>
+            // Mutex in use
+            b.await >> write(ac)
+        }
+      }
+
+    /** Wrap a [[AsyncCallback]] so that it executes in the read-mutex.
+      * There can be many readers active at one time.
+      *
+      * Note: Calling this from within the write-mutex will block.
+      */
+    def read[A](ac: AsyncCallback[A]): AsyncCallback[A] =
+      AsyncCallback.byName {
+
+        mutex match {
+          case None =>
+            // Mutex empty
+            val b = AsyncCallback.barrier.runNow()
+            mutex = Some(b)
+            assert(readers == 0)
+            readers = 1
+            ac.finallyRunSync(releaseReader)
+
+          case Some(b) =>
+            if (readers > 0) {
+              // Read-mutex in use
+              readers += 1
+              ac.finallyRunSync(releaseReader)
+
+            } else {
+              // Write-mutex in use
+              b.await >> read(ac)
+            }
+        }
+      }
+  }
+
+  /** Creates a new (non-reentrant) read/write mutex. */
+  def readWriteMutex: CallbackTo[ReadWriteMutex] =
+    CallbackTo(new ReadWriteMutex)
 }
 
 // █████████████████████████████████████████████████████████████████████████████████████████████████████████████████████

doc/changelog/1.7.7.md

@@ -48,6 +48,27 @@ This entire release is focused on `AsyncCallback`.
     }
     ```
 
+  * Added `readWriteMutex` which returns a `AsyncCallback.ReadWriteMutex`:
+
+    ```scala
+    AsyncCallback.ReadWriteMutex {
+
+      /** Wrap a AsyncCallback so that it executes in the write-mutex.
+        * There can only be one writer active at one time.
+        *
+        * Note: THIS IS NOT RE-ENTRANT. Calling this from within the read or write mutex will block.
+        */
+      def write[A](ac: AsyncCallback[A]): AsyncCallback[A]
+
+      /** Wrap a AsyncCallback so that it executes in the read-mutex.
+        * There can be many readers active at one time.
+        *
+        * Note: Calling this from within the write-mutex will block.
+        */
+      def read[A](ac: AsyncCallback[A]): AsyncCallback[A]
+    }
+    ```
+
   * You can now add a `_` suffix to the following to return an `AsyncCallback[Unit]` and be more efficient under-the-hood:
 
     * `traverse`

test/src/test/scala/japgolly/scalajs/react/core/AsyncCallbackTest.scala

@@ -6,6 +6,7 @@ import japgolly.scalajs.react.{AsyncCallback, Callback}
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.Future
 import scala.util.Success
+import sourcecode.Line
 import utest._
 
 object AsyncCallbackTest extends TestSuite {
@@ -26,6 +27,22 @@ object AsyncCallbackTest extends TestSuite {
     }.unsafeToFuture()
   }
 
+  private def assertMutexRejection(mutex: AsyncCallback[Unit] => AsyncCallback[Unit])(implicit l: Line): AsyncCallback[Unit] =
+    mutex(AsyncCallback.unit).timeoutMs(500).delayMs(1).map(o => assertEq(o.isDefined, false))
+
+  private def testInMutex(mutex: AsyncCallback[Unit] => AsyncCallback[Unit])
+                         (test: AsyncCallback[Any]): AsyncCallback[Unit] =
+    for {
+      b <- AsyncCallback.barrier.asAsyncCallback
+      _ <- mutex(b.await).fork_.asAsyncCallback // mutex start
+      _ <- test
+      _ <- b.complete.asAsyncCallback // mutex end
+      _ <- mutex(AsyncCallback.unit)
+    } yield ()
+
+  private def testWriteMutex(mutex: AsyncCallback[Unit] => AsyncCallback[Unit]): AsyncCallback[Unit] =
+    testInMutex(mutex)(assertMutexRejection(mutex))
+
   override def tests = Tests {
 
     "fromCallback" - {
@@ -239,12 +256,29 @@ object AsyncCallbackTest extends TestSuite {
     "mutex" - asyncTest {
       for {
         mutex <- AsyncCallback.mutex.asAsyncCallback
-        b     <- AsyncCallback.barrier.asAsyncCallback
-        _     <- mutex(b.await).fork_.asAsyncCallback // mutex start
-        t     <- mutex(AsyncCallback.unit).timeoutMs(500).delayMs(1)
-        _     <- b.complete.asAsyncCallback // mutex end
-        _     <- mutex(AsyncCallback.unit)
-      } yield assert(t.isEmpty)
+        _     <- testWriteMutex(mutex(_))
+      } yield ()
+    }
+
+    "readWriteMutex" - asyncTest {
+      for {
+        mutex <- AsyncCallback.readWriteMutex.asAsyncCallback
+
+        _     <- testWriteMutex(mutex.write) // confirm write blocks writes
+
+        b1    <- AsyncCallback.barrier.asAsyncCallback
+        b2    <- AsyncCallback.barrier.asAsyncCallback
+        l     <- AsyncCallback.countDownLatch(2).asAsyncCallback
+        _     <- mutex.read(l.countDown.asAsyncCallback >> b1.await).fork_.asAsyncCallback // read mutex enter
+        _     <- mutex.read(l.countDown.asAsyncCallback >> b2.await).fork_.asAsyncCallback // read mutex enter
+        _     <- l.await // confirm we've got two readers in the mutex at the same time
+        _     <- assertMutexRejection(mutex.write) // confirm we can't use the write mutex
+        _     <- b1.complete.asAsyncCallback // read mutex exit
+        _     <- b2.complete.asAsyncCallback // read mutex exit
+
+        _     <- testInMutex(mutex.write)(assertMutexRejection(mutex.read)) // confirm write blocks reads
+
+      } yield ()
     }
 
   }