#15 Add support for priority prioritySelect() for channels.

Author: yruslan

src/main/scala/com/github/yruslan/channel/Channel.scala

@@ -101,7 +101,7 @@ abstract class Channel[T] extends ReadChannel[T] with WriteChannel[T] {
 
   protected def fetchValueOpt(): Option[T]
 
-  final override def sender(value: T) (action: => Unit = {}): Selector = {
+  final override def sender(value: T)(action: => Unit = {}): Selector = {
     new Selector(true, this) {
       override def sendRecv(): Boolean = trySend(value)
 
@@ -250,23 +250,23 @@ object Channel {
   val CLOSED = 2
 
   /**
-   * Create a synchronous channel.
-   *
-   * @tparam T The type of the channel.
-   * @return A new channel
-   */
+    * Create a synchronous channel.
+    *
+    * @tparam T The type of the channel.
+    * @return A new channel
+    */
   def make[T]: Channel[T] = {
     new SyncChannel[T]
   }
 
   /**
-   * Create a channel. By default a synchronous channel will be created.
-   * If bufferSize is greater then zero, a buffered channel will be created.
-   *
-   * @param bufferSize Asynchronous buffer size.
-   * @tparam T The type of the channel.
-   * @return A new channel
-   */
+    * Create a channel. By default a synchronous channel will be created.
+    * If bufferSize is greater then zero, a buffered channel will be created.
+    *
+    * @param bufferSize Asynchronous buffer size.
+    * @tparam T The type of the channel.
+    * @return A new channel
+    */
   def make[T](bufferSize: Int): Channel[T] = {
     require(bufferSize >= 0)
 
@@ -288,41 +288,86 @@ object Channel {
   }
 
   /**
-   * Waits for a non-blocking operation to be available on the list of channels.
-   *
-   * @param selector  A first channel to wait for (mandatory).
-   * @param selectors Other channels to wait for.
-   * @return true is none of the channels are closed and select() can be invoked again, false if at least one of channels is closed.
-   */
+    * Waits for a non-blocking operation to be available on the list of channels.
+    * If more than one channel is ready to perform its operation, the channel to perform the operation on will be chosen
+    * at random.
+    *
+    * @param selector  A first channel to wait for (mandatory).
+    * @param selectors Other channels to wait for.
+    * @return true is none of the channels are closed and select() can be invoked again, false if at least one of channels is closed.
+    */
   def select(selector: Selector, selectors: Selector*): Boolean = {
-    trySelect(Duration.Inf, selector, selectors: _*)
+    trySelect(Duration.Inf, false, selector, selectors: _*)
+  }
+
+  /**
+    * Waits for a non-blocking operation to be available on the list of channels.
+    * If more than one channel is ready to perform its operation, the first one in the list takes precedence.
+    *
+    * @param selector  A first channel to wait for (mandatory).
+    * @param selectors Other channels to wait for.
+    * @return true is none of the channels are closed and select() can be invoked again, false if at least one of channels is closed.
+    */
+  def prioritySelect(selector: Selector, selectors: Selector*): Boolean = {
+    trySelect(Duration.Inf, true, selector, selectors: _*)
   }
 
   /**
-   * Non-blocking check for a possibility of a non-blocking operation on several channels.
-   *
-   * @param selector  A first channel to wait for (mandatory).
-   * @param selectors Other channels to wait for.
-   * @return true if one of pending operations wasn't blocking.
-   */
+    * Non-blocking check for a possibility of a non-blocking operation on several channels.
+    * If more than one channel is ready to perform its operation, the channel to perform the operation on will be chosen
+    * at random.
+    *
+    * @param selector  A first channel to wait for (mandatory).
+    * @param selectors Other channels to wait for.
+    * @return true if one of pending operations wasn't blocking.
+    */
   def trySelect(selector: Selector, selectors: Selector*): Boolean = {
-    trySelect(Duration.Zero, selector, selectors: _*)
+    trySelect(Duration.Zero, false, selector, selectors: _*)
   }
 
   /**
-   * Waits for a non-bloaking action to be available.
-   *
-   * @param timout    A timeout to wait for a non-blocking action to be available.
-   * @param selector  A first channel to wait for (mandatory).
-   * @param selectors Other channels to wait for.
-   * @return true if one of pending operations wasn't blockingю
-   */
-  @throws[InterruptedException]
+    * Non-blocking check for a possibility of a non-blocking operation on several channels.
+    *
+    * @param selector  A first channel to wait for (mandatory).
+    * @param selectors Other channels to wait for.
+    * @return true if one of pending operations wasn't blocking.
+    */
   def trySelect(timout: Duration, selector: Selector, selectors: Selector*): Boolean = {
+    trySelect(timout, false, selector, selectors: _*)
+  }
+
+  /**
+    * Non-blocking check for a possibility of a non-blocking operation on several channels.
+    * If more than one channel is ready to perform its operation, the first one in the list takes precedence.
+    *
+    * @param selector  A first channel to wait for (mandatory).
+    * @param selectors Other channels to wait for.
+    * @return true if one of pending operations wasn't blocking.
+    */
+  def tryPrioritySelect(selector: Selector, selectors: Selector*): Boolean = {
+    trySelect(Duration.Zero, true, selector, selectors: _*)
+  }
+
+  /**
+    * Waits for a non-bloaking action to be available.
+    *
+    * @param timout            A timeout to wait for a non-blocking action to be available.
+    * @param isPriorityOrdered If true, when more then one selectors is ready, the first one in the list will be selected.
+    * @param selector          A first channel to wait for (mandatory).
+    * @param selectors         Other channels to wait for.
+    * @return true if one of pending operations wasn't blockingю
+    */
+  @throws[InterruptedException]
+  def trySelect(timout: Duration, isPriorityOrdered: Boolean, selector: Selector, selectors: Selector*): Boolean = {
     val sem = new Semaphore(0)
 
-    // If several channels have pending messages, select randomly the channel to return
-    val sel = scala.util.Random.shuffle(selector :: selectors.toList).toArray
+    val sel = if (isPriorityOrdered) {
+      // If channels are ordered by priority, retain the original order
+      (selector :: selectors.toList).toArray
+    } else {
+      // If several channels have pending messages, select randomly the channel to return
+      scala.util.Random.shuffle(selector :: selectors.toList).toArray
+    }
 
     // Add waiters
     var i = 0

src/test/scala/com/github/yruslan/channel/GuaranteesSuite.scala

@@ -17,9 +17,8 @@ package com.github.yruslan.channel
 
 import java.time.Instant
 import java.util.concurrent.{Executors, TimeUnit}
-
 import org.scalatest.wordspec.AnyWordSpec
-import com.github.yruslan.channel.Channel.select
+import com.github.yruslan.channel.Channel.{select, trySelect}
 
 import scala.collection.mutable.{ArrayBuffer, ListBuffer}
 import scala.concurrent._
@@ -118,11 +117,55 @@ class GuaranteesSuite extends AnyWordSpec {
     }
   }
 
+  "Priority is honored for priority selects" when {
+    "several sync input sync output channels are active, a channel is selected according to priority" in {
+      val in1 = Channel.make[Int]
+      val in2 = Channel.make[Int]
+
+      val out1 = Channel.make[Int]
+      val out2 = Channel.make[Int]
+
+      testFairness(in1, in2, out1, out2, isPriority = true)
+    }
+
+    "several sync input async output channels are active, a channel is selected according to priority" in {
+      val in1 = Channel.make[Int]
+      val in2 = Channel.make[Int]
+
+      val out1 = Channel.make[Int](1)
+      val out2 = Channel.make[Int](1)
+
+      testFairness(in1, in2, out1, out2, isPriority = true)
+    }
+
+    "several async input sync output channels are active, a channel is selected according to priority" in {
+      val in1 = Channel.make[Int](1)
+      val in2 = Channel.make[Int](1)
+
+      val out1 = Channel.make[Int]
+      val out2 = Channel.make[Int]
+
+      testFairness(in1, in2, out1, out2, isPriority = true)
+    }
+
+    "several async input async output channels are active, a channel is selected according to priority" in {
+      val in1 = Channel.make[Int](1)
+      val in2 = Channel.make[Int](1)
+
+      val out1 = Channel.make[Int](1)
+      val out2 = Channel.make[Int](1)
+
+      testFairness(in1, in2, out1, out2, isPriority = true)
+    }
+  }
+
+
   /* Full qualified name 'com.github.yruslan.channel.Channel' is used here to make IntelliJ IDEA happy. */
   private def testFairness(in1: com.github.yruslan.channel.Channel[Int],
                            in2: com.github.yruslan.channel.Channel[Int],
                            out1: com.github.yruslan.channel.Channel[Int],
-                           out2: com.github.yruslan.channel.Channel[Int]): Unit = {
+                           out2: com.github.yruslan.channel.Channel[Int],
+                           isPriority: Boolean = false): Unit = {
     val results = new ListBuffer[(Int, Int)]
 
     def balancer(input1: ReadChannel[Int],
@@ -134,14 +177,18 @@ class GuaranteesSuite extends AnyWordSpec {
       var exit = false
 
       while (!exit) {
-        select(
+        trySelect(
+          Duration.Inf,
+          isPriority,
           input1.recver(x => v = x),
           input2.recver(x => v = x),
           finishChannel.recver(_ => exit = true)
         )
 
         if (!exit) {
-          select(
+          trySelect(
+            Duration.Inf,
+            isPriority,
             output1.sender(v) {},
             output2.sender(v) {}
           )
@@ -196,10 +243,17 @@ class GuaranteesSuite extends AnyWordSpec {
     assert(results.size == 100)
     assert(results.map(_._2).sum == 10100) // sum(1..100)*2 = 101*50*2 = 5050*2 = 10100
 
-    // Fairness
     val processedBy = Range(0, 4).map(w => results.count(_._1 == w))
-    assert(processedBy.min > 15)
-    assert(processedBy.max < 35)
+
+    if (isPriority) {
+      // Priority
+      assert(processedBy.min < 5)
+      assert(processedBy.max > 45)
+    } else {
+      // Fairness
+      assert(processedBy.min > 15)
+      assert(processedBy.max < 35)
+    }
   }
 
 }