#7 Add filter() support for channels

Author: yruslan

README.md

@@ -348,7 +348,7 @@ Since Scala is a functional language, this implementation of channels supports f
 val chInt = Channel.make[Int](2)
 
 // The channel of strings is the result of mapping the channel of integer 
-val chString = ch1.map(v => v.toString)
+val chString = chInt.map(v => v.toString)
 
 // Send some integers
 chInt.send(1)
@@ -360,6 +360,26 @@ val s1: String = chString.recv()
 val s2: String = chString.recv()
 ```
 
+#### filter()
+
+```scala
+// Creating a channel of integers
+val ch1 = Channel.make[Int](3)
+
+// Filter the original channel 
+val chFiltered = ch1.filter(v => v != 2)
+
+// Send some integers
+ch1.send(1)
+ch1.send(2)
+ch1.send(3)
+ch1.close()
+
+// Receive filtered values
+val v1 = chFiltered.recv() // 1
+val v2 = chFiltered.recv() // 3
+```
+
 ## Reference
 
 *ToDo*

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

@@ -102,10 +102,6 @@ abstract class Channel[T] extends ReadChannel[T] with WriteChannel[T] {
     }
   }
 
-  final override def map[U](f: T => U): ReadChannel[U] = {
-    new ChannelDecoratorMap[T, U](this, f)
-  }
-
   protected def fetchValueOpt(): Option[T]
 
   final override def sender(value: T) (action: => Unit = {}): Selector = {

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

@@ -0,0 +1,87 @@
+/*
+ * The MIT License (MIT)
+ *
+ * Copyright (c) 2020 Ruslan Yushchenko
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ *
+ * For more information, please refer to <http://opensource.org/licenses/MIT>
+ */
+
+package com.github.yruslan.channel
+
+import com.github.yruslan.channel.impl.Selector
+
+import java.time.Instant
+import java.time.Instant.now
+import scala.concurrent.duration.{Duration, MILLISECONDS}
+
+class ChannelDecoratorFilter[T](inputChannel: ReadChannel[T], pred: T => Boolean) extends ChannelDecorator[T](inputChannel) with ReadChannel[T] {
+  override def recv(): T = {
+    var v = inputChannel.recv()
+    var found = pred(v)
+
+    while (!found) {
+      v = inputChannel.recv()
+      found = pred(v)
+    }
+    v
+  }
+
+  override def tryRecv(): Option[T] = {
+    var valueOpt = inputChannel.tryRecv()
+    var found = valueOpt.isEmpty || valueOpt.forall(v => pred(v))
+
+    while (!found) {
+      valueOpt = inputChannel.tryRecv()
+      found = valueOpt.isEmpty || valueOpt.forall(v => pred(v))
+    }
+    valueOpt
+  }
+
+  override def tryRecv(timeout: Duration): Option[T] = {
+    if (timeout == Duration.Zero) {
+      return tryRecv()
+    }
+
+    val timeoutMilli = if (timeout.isFinite) timeout.toMillis else 0L
+    val startInstant = Instant.now()
+    var valueOpt = inputChannel.tryRecv(timeout)
+    var found = valueOpt.isEmpty || valueOpt.forall(v => pred(v))
+    var elapsedTime = java.time.Duration.between(startInstant, now).toMillis
+
+    if (found || elapsedTime >= timeoutMilli) {
+      valueOpt
+    } else {
+      while (!found && elapsedTime < timeoutMilli) {
+        val newTimeout = Duration(timeoutMilli - elapsedTime, MILLISECONDS)
+        valueOpt = inputChannel.tryRecv(newTimeout)
+        found = valueOpt.isEmpty || valueOpt.forall(v => pred(v))
+        elapsedTime = java.time.Duration.between(startInstant, now).toMillis
+      }
+      valueOpt
+    }
+  }
+
+  override def recver(action: T => Unit): Selector = inputChannel.recver(t => if (pred(t)) action(t))
+
+  override def fornew(action: T => Unit): Unit = inputChannel.fornew(t => if (pred(t)) action(t))
+
+  override def foreach(action: T => Unit): Unit = inputChannel.foreach(t => if (pred(t)) action(t))
+}

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

@@ -41,6 +41,4 @@ class ChannelDecoratorMap[T, U](inputChannel: ReadChannel[T], f: T => U) extends
   override def fornew(action: U => Unit): Unit = inputChannel.fornew(t => action(f(t)))
 
   override def foreach(action: U => Unit): Unit = inputChannel.foreach(t => action(f(t)))
-
-  override def map[K](mapFunction: U => K): ReadChannel[K] = new ChannelDecoratorMap[U, K](this, mapFunction)
 }

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

@@ -40,5 +40,6 @@ trait ReadChannel[T] extends ChannelLike {
   def fornew(f: T => Unit): Unit
   def foreach(f: T => Unit): Unit
 
-  def map[U](f: T => U): ReadChannel[U]
+  def map[U](f: T => U): ReadChannel[U] = new ChannelDecoratorMap[T, U](this, f)
+  def filter(f: T => Boolean): ReadChannel[T] = new ChannelDecoratorFilter[T](this, f)
 }