Initial impl of parking lot app using fs2

build.sbt

@@ -54,6 +54,17 @@ lazy val scala2 = project
     )
   )
 
+lazy val fs2_lot = project
+  .in(file("fs2-lot"))
+  .settings(
+    name := "fs2-lot",
+    scalaVersion := scala3Version,
+    libraryDependencies ++= Seq(
+      "co.fs2" %% "fs2-core" % "3.12.0",
+      "co.fs2" %% "fs2-io" % "3.12.0"
+    )
+  )
+
 lazy val ManualTestConfig = config("manual") extend (Test)
 lazy val manualTestSettings = inConfig(ManualTestConfig)(Defaults.testSettings)
 

fs2-lot/src/main/scala/com/yadavan88/parking/FS2Lot.scala

@@ -0,0 +1,243 @@
+package com.yadavan88.parking
+
+import cats.effect.IO
+import fs2.Stream
+import scala.concurrent.duration._
+import cats.effect.std.Console
+import cats.effect.kernel.Ref
+import fs2.concurrent.SignallingRef
+import cats.syntax.traverse._
+import cats.syntax.all._
+import scala.util.Random
+import fs2.Pipe
+import cats.effect.IOApp
+
+object FS2Lot {
+
+  val incomingCarSimulationRate = 10.seconds
+  val outgoingCarSimulationRate = 6.seconds
+  val outgoingCarStartDelay = 40.second
+  val statusStreamRate = 3.seconds
+
+  def init(totalFloors: Int, spotsPerFloor: Int): ParkingLot = {
+    val spots = (0 until totalFloors).flatMap { floor =>
+      (0 until spotsPerFloor).map(id => ParkingSpot(id, floor, None))
+    }
+    ParkingLot(spots.toList, totalFloors, spotsPerFloor)
+  }
+
+  def incomingCarSimulation(
+      pausedRef: SignallingRef[IO, Boolean]
+  ): Stream[IO, Car] = {
+    Stream
+      .awakeEvery[IO](incomingCarSimulationRate)
+      .interruptWhen(pausedRef)
+      .scan(0) { (counter, _) => counter + 1 }
+      .map(counter => Car(counter))
+  }
+
+  def outgoingCarSimulation(
+      lotRef: Ref[IO, ParkingLot],
+      pausedRef: SignallingRef[IO, Boolean]
+  ): Stream[IO, Car] = {
+    Stream
+      .awakeEvery[IO](outgoingCarSimulationRate)
+      .interruptWhen(pausedRef)
+      .evalMap { _ =>
+        for {
+          lot <- lotRef.get
+          occupiedSpots = lot.spots.filter(_.isOccupied)
+          spot <- IO(occupiedSpots(Random.nextInt(occupiedSpots.size)))
+        } yield spot.carId.map(id => Car(id, Some(spot.id), Some(spot.floor)))
+      }
+      .unNone
+  }
+
+  def parkCar(lotRef: Ref[IO, ParkingLot]): Pipe[IO, Car, Car] = { cars =>
+    cars.evalMap { car =>
+      for {
+        lot <- lotRef.get
+        spot <- IO(lot.findAvailableSpot)
+        updatedCar <- spot match {
+          case Some(s) =>
+            lotRef.update(_.occupySpot(s.id, s.floor, car.id).get) *>
+              Console[IO].println(
+                s"Car ${car.displayId} parked at ${s.name}"
+              ) *>
+              IO(car.withSpot(s.id, s.floor))
+          case None =>
+            Console[IO].println(s"Car ${car.displayId} couldn't find a spot") *>
+              IO(car)
+        }
+      } yield updatedCar
+    }
+  }
+
+  def removeCar(lotRef: Ref[IO, ParkingLot]): Pipe[IO, Car, Car] = { cars =>
+    cars.evalMap { car =>
+      for {
+        lot <- lotRef.get
+        _ <- (car.spotId, car.floor) match {
+          case (Some(id), Some(f)) =>
+            lotRef.update(_.releaseSpot(id, f).get) *>
+              Console[IO].println(
+                s"Car ${car.displayId} left from L${f}-${id}"
+              )
+          case _ =>
+            Console[IO].println(
+              s"Car ${car.displayId} not found in parking lot"
+            )
+        }
+      } yield car
+    }
+  }
+
+  def statusStream(
+      lotRef: Ref[IO, ParkingLot],
+      pausedRef: SignallingRef[IO, Boolean]
+  ): Stream[IO, Unit] = {
+    Stream
+      .awakeEvery[IO](statusStreamRate)
+      .interruptWhen(pausedRef)
+      .evalMap { _ =>
+        for {
+          lot <- lotRef.get
+          _ <- Console[IO].println("\n=== Parking Lot Status ===")
+          _ <- (0 until lot.totalFloors).toList.traverse { floor =>
+            val (occupied, total) = lot.getOccupancyByFloor(floor)
+            val status =
+              if (lot.blockedFloors.contains(floor)) "[CLOSED]" else ""
+            val occupancy = s"$occupied/$total"
+            val padding = " " * (3 - occupancy.length)
+            Console[IO].println(
+              s"Floor $floor - $padding$occupancy occupied $status"
+            )
+          }
+          _ <- Console[IO].println("=======================\n")
+        } yield ()
+      }
+  }
+
+  def parseCommand(input: String): Option[Command] = {
+    input.trim.toLowerCase match {
+      case "pause"           => Some(Command.Pause)
+      case "resume"          => Some(Command.Resume)
+      case "status"          => Some(Command.Status)
+      case "quit"            => Some(Command.Quit)
+      case s"block $floor"   => Some(Command.Block(floor.toInt))
+      case s"unblock $floor" => Some(Command.Unblock(floor.toInt))
+      case _                 => None
+    }
+  }
+
+  def handlePause(pausedRef: Ref[IO, Boolean]): IO[Unit] = {
+    for {
+      isPaused <- pausedRef.get
+      _ <-
+        if (!isPaused) {
+          pausedRef.set(true) *>
+            Console[IO].println(
+              "System paused. Use 'resume' command to continue."
+            )
+        } else IO.unit
+    } yield ()
+  }
+
+  def handleResume(pausedRef: Ref[IO, Boolean]): IO[Unit] = {
+    for {
+      isPaused <- pausedRef.get
+      _ <-
+        if (isPaused) {
+          pausedRef.set(false) *>
+            Console[IO].println("System resumed")
+        } else IO.unit
+    } yield ()
+  }
+
+  def handleStatus(lotRef: Ref[IO, ParkingLot]): IO[Unit] = {
+    for {
+      lot <- lotRef.get
+      _ <- (0 until lot.totalFloors).toList.traverse { floor =>
+        val (occupied, total) = lot.getOccupancyByFloor(floor)
+        val status =
+          if (lot.blockedFloors.contains(floor)) "MAINTENANCE"
+          else "OPERATIONAL"
+        Console[IO].println(
+          s"Floor $floor ($status): $occupied/$total spots occupied"
+        )
+      }
+    } yield ()
+  }
+
+  def handleBlock(lotRef: Ref[IO, ParkingLot], floor: Int): IO[Unit] = {
+    lotRef.update(_.blockFloor(floor)) *>
+      Console[IO].println(s"Floor $floor blocked for maintenance")
+  }
+
+  def handleUnblock(lotRef: Ref[IO, ParkingLot], floor: Int): IO[Unit] = {
+    lotRef.update(_.unblockFloor(floor)) *>
+      Console[IO].println(s"Floor $floor unblocked")
+  }
+
+  def handleQuit: IO[Unit] = {
+    IO.raiseError(new Exception("Quitting program"))
+  }
+
+  def handleInvalidCommand: IO[Unit] = {
+    Console[IO].println("Invalid command")
+  }
+
+  def userInputStream(
+      lotRef: Ref[IO, ParkingLot],
+      pausedRef: Ref[IO, Boolean]
+  ): Stream[IO, Unit] = {
+    Stream
+      .repeatEval {
+        for {
+          _ <- Console[IO].println(
+            """
+              |Available Commands:
+              |  pause          - Pause car simulation
+              |  resume         - Resume car simulation
+              |  status         - Show current parking lot status
+              |  block <floor>  - Block a floor for maintenance (e.g., block 2)
+              |  unblock <floor>- Unblock a floor (e.g., unblock 2)
+              |  quit           - Exit the application
+              |Enter command:""".stripMargin
+          )
+          input <- Console[IO].readLine
+          command = parseCommand(input)
+          _ <- command match {
+            case Some(Command.Pause)          => handlePause(pausedRef)
+            case Some(Command.Resume)         => handleResume(pausedRef)
+            case Some(Command.Status)         => handleStatus(lotRef)
+            case Some(Command.Block(floor))   => handleBlock(lotRef, floor)
+            case Some(Command.Unblock(floor)) => handleUnblock(lotRef, floor)
+            case Some(Command.Quit)           => handleQuit
+            case None                         => handleInvalidCommand
+          }
+        } yield ()
+      }
+  }
+
+  def run(totalFloors: Int, spotsPerFloor: Int): IO[Unit] = {
+    for {
+      lotRef <- Ref[IO].of(init(totalFloors, spotsPerFloor))
+      pausedRef <- SignallingRef[IO].of(false)
+      _ <- (
+        incomingCarSimulation(pausedRef).through(parkCar(lotRef)).void merge
+          outgoingCarSimulation(lotRef, pausedRef)
+            .through(removeCar(lotRef))
+            .void merge
+          statusStream(lotRef, pausedRef) merge
+          userInputStream(lotRef, pausedRef)
+      ).compile.drain
+    } yield ()
+  }
+}
+
+object Main extends IOApp.Simple {
+  def run: IO[Unit] = {
+    FS2Lot.run(5, 5)
+  }
+}

fs2-lot/src/main/scala/com/yadavan88/parking/Models.scala

@@ -0,0 +1,77 @@
+package com.yadavan88.parking
+
+import scala.util.Random
+
+sealed trait Command
+object Command {
+  case object Pause extends Command
+  case object Resume extends Command
+  case object Status extends Command
+  case object Quit extends Command
+  case class Block(floor: Int) extends Command
+  case class Unblock(floor: Int) extends Command
+}
+case class Car(id: Int, spotId: Option[Int] = None, floor: Option[Int] = None) {
+  def withSpot(spotId: Int, floor: Int): Car = copy(spotId = Some(spotId), floor = Some(floor))
+  def displayId: String = s"CAR-${String.format("%03d", id)}"
+}
+case class ParkingSpot(id: Int, floor: Int, carId: Option[Int] = None) {
+  def name = s"L${floor}-${id}"
+  def isOccupied: Boolean = carId.isDefined
+}
+case class ParkingLot(
+    spots: List[ParkingSpot],
+    totalFloors: Int,
+    spotsPerFloor: Int,
+    blockedFloors: Set[Int] = Set.empty
+) {
+  def findAvailableSpot: Option[ParkingSpot] = {
+    (0 until totalFloors)
+      .filterNot(blockedFloors.contains) 
+      .flatMap(findAvailableSpotOnFloor)
+      .headOption 
+  }
+
+  def findAvailableSpotOnFloor(floor: Int): Option[ParkingSpot] =
+    spots.find(spot =>
+      spot.floor == floor && !spot.isOccupied && !blockedFloors.contains(floor)
+    )
+
+  def occupySpot(spotId: Int, floorId: Int, carId: Int): Option[ParkingLot] = {
+    val spot = spots.find(s => s.id == spotId && s.floor == floorId)
+    spot match {
+      case Some(s) =>
+        val updatedSpots = spots.map { spot =>
+          if (spot.id == spotId && spot.floor == floorId) spot.copy(carId = Some(carId))
+          else spot
+        }
+        Some(copy(spots = updatedSpots))
+      case None => None
+    }
+  }
+
+  def releaseSpot(spotId: Int, floorId: Int): Option[ParkingLot] = {
+    val spot = spots.find(s => s.id == spotId && s.floor == floorId)
+    spot match {
+      case Some(s) =>
+        val updatedSpots = spots.map { spot =>
+          if (spot.id == spotId && spot.floor == floorId) spot.copy(carId = None)
+          else spot
+        }
+        Some(copy(spots = updatedSpots))
+      case None => None
+    }
+  }
+
+  def getOccupancyByFloor(floor: Int): (Int, Int) = {
+    val floorSpots = spots.filter(_.floor == floor)
+    val occupied = floorSpots.count(_.carId.isDefined)
+    (occupied, floorSpots.size)
+  }
+
+  def blockFloor(floor: Int): ParkingLot =
+    copy(blockedFloors = blockedFloors + floor)
+
+  def unblockFloor(floor: Int): ParkingLot =
+    copy(blockedFloors = blockedFloors - floor)
+}