Support for filter/collect

core/src/main/scala/ox/channels/Cell.scala

@@ -32,14 +32,16 @@ private[ox] class Cell[T] extends CellCompleter[T]:
  def take(): SelectResult[T] | Cell[T] | ChannelState.Closed = cell.take()
  def isAlreadyOwned: Boolean = isOwned.get()
 
-/** Linked cells are created when creating MappedSources. */
-private[ox] class LinkedCell[T, U](linkedTo: CellCompleter[U], f: T => U, createReceived: U => Source[U]#Received)
+/** Linked cells are created when creating CollectSources. */
+private[ox] class LinkedCell[T, U](linkedTo: CellCompleter[U], f: T => Option[U], createReceived: U => Source[U]#Received)
  extends CellCompleter[T] {
  override def complete(t: SelectResult[T]): Unit =
- val u: SelectResult[U] = t match
- case r: Source[T]#Received => createReceived(f(r.value)) // TODO exceptions
- case _ => throw new IllegalStateException() // linked cells can only be created from sources
- linkedTo.complete(u)
+ t match
+ case r: Source[T]#Received =>
+ f(r.value) match // TODO exceptions
+ case Some(u) => linkedTo.complete(createReceived(u))
+ case None => linkedTo.completeWithNewCell() // ignoring the received value
+ case _ => throw new IllegalStateException() // linked cells can only be created from sources
  override def completeWithNewCell(): Unit = linkedTo.completeWithNewCell()
  override def completeWithClosed(s: ChannelState.Closed): Unit = linkedTo.completeWithClosed(s)
  override def tryOwn(): Boolean = linkedTo.tryOwn()

core/src/main/scala/ox/channels/Channel.scala

@@ -287,8 +287,11 @@ class BufferedChannel[T](capacity: Int = 1) extends Channel[T]:
  else c.complete(Received(w)) // sending the element
  drainWaitingReceivesWhenDone()
 
-class MappedSource[T, U](s: Source[T], f: T => U) extends Source[U]:
- override def receive(): U | ChannelClosed = select(List(s.receiveClause)).map(_.value).map(f)
+class CollectSource[T, U](s: Source[T], f: T => Option[U]) extends Source[U]:
+ @tailrec final override def receive(): U | ChannelClosed = select(List(s.receiveClause)).map(_.value).map(f) match
+ case Some(u) => u
+ case None => receive()
+ case c: ChannelClosed => c
  override private[ox] def receiveCellOffer(c: CellCompleter[U]): Unit = s.receiveCellOffer(createLinkedCell(c))
  override private[ox] def receiveCellCleanup(c: CellCompleter[U]): Unit = s.receiveCellCleanup(createLinkedCell(c))
  override private[ox] def trySatisfyWaiting(): Unit | ChannelClosed = s.trySatisfyWaiting()

core/src/main/scala/ox/channels/SourceOps.scala

@@ -5,6 +5,14 @@ import ox.*
 import scala.concurrent.duration.FiniteDuration
 
 trait SourceOps[+T] { this: Source[T] =>
+ // view ops (lazy)
+
+ def mapAsView[U](f: T => U): Source[U] = CollectSource(this, t => Some(f(t)))
+ def filterAsView(f: T => Boolean): Source[T] = CollectSource(this, t => if f(t) then Some(t) else None)
+ def collectAsView[U](f: PartialFunction[T, U]): Source[U] = CollectSource(this, f.lift)
+
+ // run ops (eager)
+
  def map[U](f: T => U)(using Ox, StageCapacity): Source[U] =
  val c2 = Channel[U](summon[StageCapacity].toInt)
  fork {
@@ -24,8 +32,6 @@ trait SourceOps[+T] { this: Source[T] =>
  }
  c2
 
- def mapView[U](f: T => U): Source[U] = MappedSource(this, f)
-
  def take(n: Int)(using Ox, StageCapacity): Source[T] = transform(_.take(n))
 
  def filter(f: T => Boolean)(using Ox, StageCapacity): Source[T] = transform(_.filter(f))