map/parJoin

[Fristi]

I've got this stream where we unzip files from a incoming Stream:

This works

archive
  //unzip
  .through(Archive.unzip(blocker))
  .parEvalMapUnordered(5) { entry =>
    entry.body.through(store.put(Path(s"import/$id/${entry.name}"))).compile.drain.as(entry.name)
  }
  .compile
  .toList

While this doesn't

archive
  //unzip
  .through(Archive.unzip(blocker))
  .map(entry => entry.body.through(store.put(Path(s"import/$id/${entry.name}"))).as(entry.name))
  .parJoin(5)
  .compile
  .toList

With the latter, it uploads the file to the blob store, but it doesn't yield the entry.name

[mpilquist]

Check the implementation of store.put -- I'm guessing it doesn't emit anything. E.g., def put(...): Stream[F, Unit] = Stream.eval_(takeAction(...)) emits no output.

[Fristi]

Ah that's correct :) My guess that .map(entry => entry.body.through(store.put(Path(s"import/$id/${entry.name}"))) >> Stream.emit(entry.name)) would work, but it yields the same result

[SystemFw]

you need to use ++ , not >>

[Fristi]

That was my guess aswell! Was missing the .drain on the non-emitting stream.

archive
  //unzip
  .through(Archive.unzip(blocker))
  .map(entry => entry.body.through(store.put(Path(s"import/$id/${entry.name}"))).drain ++ Stream.emit(entry.name))
  .parJoin(5)
  .compile
  .toList

That works 🎉