functional-programming-snake/scala/solutions/Stream.scala

package snake

/**
 * Something that happens to the program: time passing (ms since start), a key going down or up, or (from Module 13)
 * a mouse click at canvas pixel (x, y). Every stream ignores the kinds of event it does not use.
 */
enum Event:
  case Time(t: Long)
  case KeyDown(key: String)
  case KeyUp(key: String)
  case Click(x: Int, y: Int)

/**
 * m08 stream: an event stream is a pure automaton (a Mealy machine).
 * There is no hidden state: whatever a stream "remembers" lives in the next stream it returns.
 */
final case class Stream[A](step: Event => (List[A], Stream[A])):

  // Each combinator steps this stream and wraps the *next* stream in the same combinator.
  def map[B](f: A => B): Stream[B] =
    Stream { e =>
      val (out, next) = step(e)
      (out.map(f), next.map(f))
    }

  def filter(p: A => Boolean): Stream[A] =
    Stream { e =>
      val (out, next) = step(e)
      (out.filter(p), next.filter(p))
    }

  def filterMap[B](f: A => Option[B]): Stream[B] =
    Stream { e =>
      val (out, next) = step(e)
      (out.flatMap(f), next.filterMap(f))
    }

  def merge(other: Stream[A]): Stream[A] =
    Stream { e =>
      val (outA, nextA) = step(e)
      val (outB, nextB) = other.step(e)
      (outA ++ outB, nextA.merge(nextB))
    }

  // The accumulator travels in the argument of the next scan, not in a variable.
  // scanLeft keeps z as its first element; drop it, we only emit new accumulators.
  def scan[B](z: B)(f: (B, A) => B): Stream[B] =
    Stream { e =>
      val (out, next) = step(e)
      val accs = out.scanLeft(z)(f).tail
      (accs, next.scan(accs.lastOption.getOrElse(z))(f))
    }

  // Only the first step is special; afterwards we are just `next`.
  def startWith(v: A): Stream[A] =
    Stream { e =>
      val (out, next) = step(e)
      (v :: out, next)
    }

  // (m10) `last` is the previous real time, `clock` the time this stream has been told. A long silence
  // (a background tab) moves `clock` by at most `maxGap`, so this stream never has to catch up on minutes of beats.
  private def clampFrom(maxGap: Long, last: Long, clock: Long): Stream[A] =
    Stream {
      case Event.Time(t) =>
        val now = clock + math.min(math.max(t - last, 0L), maxGap)
        val (out, next) = step(Event.Time(now))
        (out, next.clampFrom(maxGap, t, now))
      case e =>
        val (out, next) = step(e)
        (out, next.clampFrom(maxGap, last, clock))
    }

  def clampTime(maxGap: Long): Stream[A] = clampFrom(maxGap, 0, 0)

  def run(events: List[Event]): List[A] =
    events
      .foldLeft((List.empty[A], this)) { case ((outs, stream), e) =>
        val (out, next) = stream.step(e)
        (outs ++ out, next)
      }
      ._1

object Stream:

  // Sources may return themselves as the next stream, since they remember nothing.
  lazy val keyDowns: Stream[String] = Stream {
    case Event.KeyDown(key) => (List(key), keyDowns)
    case _                  => (Nil, keyDowns)
  }

  lazy val keyUps: Stream[String] = Stream {
    case Event.KeyUp(key) => (List(key), keyUps)
    case _                => (Nil, keyUps)
  }

  // (m13, given in src/) The position of every Click event, in canvas pixels. Like the key sources it remembers nothing.
  lazy val clicks: Stream[(Int, Int)] = Stream {
    case Event.Click(x, y) => (List((x, y)), clicks)
    case _                 => (Nil, clicks)
  }

  // `last` is the time of the previous Time event. Multiples of ms in (last, t] fire, ascending.
  private def intervalFrom(ms: Long, last: Long): Stream[Long] = Stream {
    case Event.Time(t) => ((last / ms + 1 to t / ms).map(_ * ms).toList, intervalFrom(ms, t))
    case _             => (Nil, intervalFrom(ms, last))
  }

  def interval(ms: Long): Stream[Long] = intervalFrom(ms, 0)