Peeking at the first element or chunk

[TomasMikula]

Hi folks,

given a Stream[F, A], I need to test whether it successfully produces the first element (or chunk) and then behave differently in case of success versus failure. (In case of success, I want to return HTTP status 200 and use the stream as response body; in case of failure, I want to return HTTP status 500.)

For that, I imagine an operation like

def prefetchChunk[F[_], A](s: Stream[F, A]): Resource[F, Stream[F, A]] =
  ???

where

• the Resource gives a scope to any resources acquired while fetching the first chunk;
• the resulting Stream contains all elements of the original stream, with the first chunk (if any) already evaluated.

Is this possible?
Thanks!

[armanbilge]

Yes, I recently implemented something very close to this.
https://github.com/armanbilge/calico/blob/0baef0034b982091a6b92735843ba432606104c3/calico/src/main/scala/calico/syntax.scala#L67-L76

[TomasMikula]

Thanks, that was fast!

This is what I ended up with:

  implicit class StreamOps[A](stream: Stream[IO, A]) {
    def prefetchFirstElem: Resource[IO, Stream[IO, A]] =
      stream
        .pull
        .uncons1
        .flatMap {
          case None => Pull.output1(Stream.empty[IO])
          case Some((a, s)) => Pull.output1(Stream(a) ++ s)
        }
        .streamNoScope
        .compile
        .resource
        .lastOrError
  }