accumulator.sc

//> using dep com.softwaremill.ox::core:0.0.7

import ox.*
import ox.channels.*
import scala.annotation.tailrec

scoped {
  // fast producer
  val c = Channel[Int]()
  fork {
    forever {
      c.send(1)
      Thread.sleep(100)
    }
  }

  // slow consumer
  val d = Channel[Int]()
  val consumer = fork {
    forever {
      println("Received: " + d.receive().orThrow)
      Thread.sleep(1000)
    }
  }

  // accumulator process
  fork {
    @tailrec def loop(acc: Int): Unit =
      select(d.sendClause(acc), c.receiveClause).orThrow match
        case d.Sent()      => loop(0)
        case c.Received(n) => loop(acc + n)

    loop(0)
  }

  // unless there's an exception, this will never complete
  consumer.join()
}