1 bulb 4 switches

4bulbs.scala

package pl.ayeo

import cats.effect.{ExitCode, IO, IOApp, Ref}
import cats.implicits.*
import fs2.*
import cats.effect.unsafe.implicits.global

import scala.concurrent.duration.*

object Fs2App extends IOApp {

  def producer(message: String, duration: FiniteDuration): IO[String] = IO.sleep(duration) >> IO(message)

  val signal1 =  Stream.repeatEval(producer("A", 2.millis))
  val signal2 =  Stream.repeatEval(producer("B", 3.millis))
  val signal3 =  Stream.repeatEval(producer("C", 4.millis))
  val signal4 =  Stream.repeatEval(producer("D", 5.millis))
  
  val signals: List[Stream[IO, String]] = List(signal1, signal2, signal3, signal4)
  val signal: Stream[IO, String] = signals.fold(Stream.empty)((i, j) => i.merge(j))

  class Bulb(ref: Ref[IO, Int]) {
    def toggle(): IO[Boolean] =  ref.modify(x => (x + 1, x)) >> ref.get.flatMap(IO.println) >> isOn()
    def isOn(): IO[Boolean] = ref.get.map(x => x % 2 == 0)
  }

  def pipes(bulb: Bulb): Pipe[IO, String, Boolean] = in => in.evalMap(signal => bulb.toggle())

  val ioBulb: IO[Bulb] = Ref[IO].of(0).map(r => Bulb(r))

  val finalStream: Stream[IO, Boolean] = Stream.eval(ioBulb).flatMap(bulb => signal.take(1000).through(pipes(bulb)))

  override def run(args: List[String]): IO[ExitCode] = finalStream.compile.drain.as(ExitCode.Success)
}