scan.scala

import fs2.Stream
import fs2.Pipe
import fs2.Chunk
import fs2.Pull
import cats.effect.IO
import scala.language.higherKinds

//implementing scan from https://fs2.io/guide.html#exercises-stream-transforming
//Stream.range(1,10).scan(0)(_ + _).toList // running sum
// res38: List[Int] = List(0, 1, 3, 6, 10, 15, 21, 28, 36, 45)
object Scan {

  def scan[F[_], O, A](s: Stream[F, O])(init: A, f: (A, O) => A): Stream[F, A] = {
    def go(s1: Stream[F, O], acc: A): Pull[F, A, Unit] = {
      s1.pull.uncons.flatMap {
        case Some((chunk, next)) =>
            if (chunk.isEmpty) go(next, acc)
            else {
              val first = chunk.apply(0) //safe because chunk is not empty
              val remaining = Stream.chunk(chunk.drop(1)) ++ next
              val result = f(acc, first)
              Pull.output(Chunk(acc)) >> go(remaining, result)
            }
        case None => Pull.output(Chunk(acc)) >> Pull.done
      }
    }

    go(s, init).stream
  }
}

usage:

import fs2.Stream
import fs2.Pipe
import fs2.Chunk
import fs2.Pull
import cats.effect.IO
import scala.language.higherKinds
import net.ssanj.fs2exercises.Scan._

val s1 = Stream.range(1,10)
scan[fs2.Pure, Int, Int](s1)(0, _ + _).toList

How can we make this implementation better?

using uncons1 as recommended by @ajfitzpatrick:

  def scan2[F[_], O, A](s: Stream[F, O])(init: A, f: (A, O) => A): Stream[F, A] = {
    def go(s1: Stream[F, O], acc: A): Pull[F, A, Unit] = {
      s1.pull.uncons1.flatMap {
        case Some((element, next)) =>
          val result = f(acc, element)
          Pull.output(Chunk(acc)) >> go(next, result)
        case None => Pull.output(Chunk(acc)) >> Pull.done
      }
    }

    go(s, init).stream
  }