[core] Add Channel.takeWith for inline take+map (#1475)

Author: fwbrasil

kyo-core/shared/src/main/scala/kyo/Channel.scala

@@ -162,16 +162,24 @@ object Channel:
           *   The taken element
           */
         def take(using Frame): A < (Abort[Closed] & Async) =
+            takeWith(identity)
+
+        /** Takes an element from the channel and applies an inline function, avoiding a `.map` closure allocation.
+          *
+          * @return
+          *   The result of applying the function to the taken element
+          */
+        inline def takeWith[B, S](inline f: A => B < S)(using Frame): B < (S & Abort[Closed] & Async) =
             Sync.Unsafe.defer {
                 self.poll().foldError(
                     {
-                        case Present(value) => value
-                        case Absent         => self.takeFiber().safe.get
+                        case Present(value) => f(value)
+                        case Absent         => self.takeFiber().safe.use(f)
                     },
                     Abort.error
                 )
             }
-        end take
+        end takeWith
 
         /** Takes [[n]] elements from the channel, semantically blocking until enough elements are present. Note that if enough elements are
           * not added to the channel it can block indefinitely.

kyo-core/shared/src/test/scala/kyo/ChannelTest.scala

@@ -122,6 +122,33 @@ class ChannelTest extends Test:
             v  <- f.get
         yield assert(!d1 && v == 1)
     }
+    "takeWith" - {
+        "applies function to taken value" in run {
+            for
+                c <- Channel.init[Int](2)
+                _ <- c.put(1)
+                v <- c.takeWith(_ * 10)
+            yield assert(v == 10)
+        }
+        "blocks when empty then applies function" in run {
+            for
+                c  <- Channel.init[Int](2)
+                f  <- Fiber.initUnscoped(c.takeWith(_ + 5))
+                _  <- Async.sleep(10.millis)
+                d1 <- f.done
+                _  <- c.put(3)
+                _  <- untilTrue(f.done)
+                v  <- f.get
+            yield assert(!d1 && v == 8)
+        }
+        "fails on closed channel" in run {
+            for
+                c <- Channel.init[Int](2)
+                _ <- c.close
+                r <- Abort.run[Closed](c.takeWith(_ * 2))
+            yield assert(r.isFailure)
+        }
+    }
     "putBatch" - {
         "non-nested" - {
             "should put a batch" in run {