intersperse.scala

//I'm trying to implement `intersperse` from : https://fs2.io/guide.html#exercises-stream-transforming

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

object Intersperse {

  //Stream("Alice","Bob","Carol").intersperse("|").toList
  //res37: List[String] = List("Alice", "|", "Bob", "|", "Carol")
  def intersperse[F[_], O](sep: O): Pipe[F, O, O] = {

    def go(s: Stream[F, O]): Pull[F, O, Unit] = {
      s.pull.uncons.flatMap {
        case Some((ch, nextStream)) => 
          ch.toList match {
            case Nil => go(nextStream)
            case h :: Nil => Pull.output(Chunk(h)) >> go(nextStream) //this could emit an unbalanced sep
            case xs => Pull.output(Chunk.seq[O](xs.init.flatMap(x => List(x, sep)) :+ xs.last)) >> go(nextStream)
          }
        case None => Pull.done
      }
    }

    in => go(in).stream
  }

usage:

import Intersperse.intersperse

val s1 = Stream("Alice","Bob","Carol")
val x1 = s1.through(intersperse("#"))
x1.toList

res25: List[String] = List(Alice, #, Bob, #, Carol)

But if we do something even:

x1.take(2).toList
> res26: List[String] = List(Alice, #)

//we get a trailing '#'.

Questions

• Is this the expected implementation for the exercise? How can it be improved? It seems a little clunky
• Is there any way to get rid of the trailing '#' when creating the stream?
• Should I have implemented an implicit wrapper class to add the intersperse method to Stream?

Another possible implementation for intersperse:

  def intersperse2[F[_], O](s1: Stream[F, O], sep: O): Stream[F, O] = 
    s1.chunks.flatMap(ch => Stream.emits(ch.toVector.flatMap(x => Vector(x, sep))))