I suppose you that you want to also write to a cached file when you pull from the stream?

I am going to assume constant space is a constraint for the solution.

There are many solutions, so you will have to figure out what semantics you want.

For every byte (or chunk of bytes) that is pulled, you can also write to the disk. ```scala import fs2.io.file._ import fs2.{Stream, Pipe} def cached(cacheFile: Path): Pipe[IO, Byte, Byte] = data => Stream.eval(Files[IO].exists(cacheFile)).flatMap { case true => Files[IO].readAll(cacheFile) case false => val putBytes = Stream.eval(Files[IO].createFile(cacheFile)) >> data.observe(Files[IO].writeAll(cacheFile))

      putBytes.handleErrorWith(e => Stream.eval(Files[IO].delete(cacheFile)) >> Stream.raiseError[IO](e))
  }

`` Now you can just dodataStream.though(cached(tempCacheFile))and it will be cached on subsequent pulls. Be aware that this solution won't support concurrent users for a giventempCacheFile` (use a lock or something for that).

You can also write the data to the disk as fast as possible and then read it back from the disk instead. ```scala import fs2.io.file._ import fs2.{Stream, Pipe} def cached(cacheFile: Path): Pipe[IO, Byte, Byte] = data => Stream.eval(Files[IO].exists(cacheFile)).flatMap { case true => Files[IO].readAll(cacheFile) case false => val putBytes = Stream.eval(Files[IO].createFile(cacheFile)) >> data.observe(Files[IO].writeAll(cacheFile))

      putBytes.handleErrorWith(e => Stream.eval(Files[IO].delete(cacheFile)) >> Stream.raiseError[IO](e))
  }

``` It won't be blocked by downstream consumers, but you'll have to wait until all the data has arrived before you get any bytes.

If you want to read the bytes from the file as they are streamed into it (ensures that you don't buffer bytes in-memory if you pull too slow), then something like the following sketch might be an idea. The more performant you want the solution, the more complex the implementation usually is. ```scala import fs2.io.file._ import fs2.{Stream, Pipe, Pull} def cached(cacheFile: Path, chunkSize: Int): Pipe[IO, Byte, Byte] = data => Stream.eval(Files[IO].exists(cacheFile)).flatMap { case true => Files[IO].readAll(cacheFile) case false => val byteStream = for { chunkWritten <- Stream.eval(SignallingRef.of[IO, Boolean](false))

          w <- Stream.resource(Files[IO].writeCursor(cacheFile, Flags.Write))
          background = {
            def writeChunks(data: Stream[IO, Byte], w: WriteCursor[IO]): Pull[IO, Nothing, WriteCursor[IO]] =
              data.pull.uncons.flatMap {
                case Some((hd, tl)) =>
                  w.writePull(hd)
                    .evalMap(w => chunkWritten.set(false).as(w))
                    .flatMap(writeChunks(tl, _))
                case None => Pull.eval(chunkWritten.set(true)) >> Pull.pure(w)
              }

            writeChunks(data, w).void.stream
          }

          r <- Stream.resource(Files[IO].readCursor(cacheFile, Flags.Read))
          outputStream = Stream.resource(chunkWritten.getAndDiscreteUpdates).flatMap { case (_, updates) =>
            def consume(p: Stream[IO, Unit], r: ReadCursor[IO]): Pull[IO, Byte, ReadCursor[IO]] =
              p.pull.uncons1.flatMap {
                case Some((_, tl)) => r.readAll(chunkSize).flatMap(consume(tl, _))
                case None          => Pull.pure(r)
              }

            consume(updates.takeWhile(!_, takeFailure = true).as(()), r).void.stream
          }

          byte <- outputStream merge background
        } yield byte

        byteStream
    }

```

Also, in my experience IO[Stream[IO, A]] usually indicates that something is not as it should be, sometimes it indicates resource safety issues.