Skip to content

Add evalFlatten methods for stream of effects #2851

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Add evalFlatten methods for stream of effects #2851

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a03fc81
Add evalFlatten methods
bplommer Mar 17, 2022
0c898d1
Add type annotation, remove parEvalFlattenUnbounded
bplommer Mar 27, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 16 additions & 4 deletions core/shared/src/main/scala/fs2/Stream.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4073,10 +4073,7 @@ object Stream extends StreamLowPriority {
}

def outcomeJoiner: F[Unit] =
outcomes.stream
.evalMap(identity)
.compile
.drain
outcomes.stream.evalFlatten.compile.drain
.guaranteeCase {
case Outcome.Succeeded(_) =>
stop(None) >> output.close.void
Expand Down Expand Up @@ -4117,6 +4114,21 @@ object Stream extends StreamLowPriority {
parJoin(Int.MaxValue)
}

/** Provides syntax for a stream of `F` effects. */
implicit class StreamFOps[F[_], O](private val self: Stream[F, F[O]]) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For what it's worth, I've regretted using this shape of stream every time. You can run yourself out of memory really quickly if those inner F[A] are nontrivial and you have a lot of them.

I'm uncomfortable with encoding this in the library in a way that makes it easier to use, because I personally think this shape should be discouraged

Especially when the implementation here is pretty trivial.

And as a library user, I think it's a lot more clear to the code reader to see an inline evalMap(identity) rather than yet another method they need to learn as part of the api

So I'm a polite 👎 on the PR for those reasons

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

good to know @Daenyth, I'd never think it could blow the memory. Would a suspend help resolve that? (at least to reduce the overhead of the non-trivial ones)

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Probably not because the issue is the memory used by having large Chunk[IO[A]]

It's also possibly my codebase just was doing something really silly and that's why it cost so much memory for that structure


/** Sequences the inner effects into the stream. */
def evalFlatten: Stream[F, O] =
self.evalMap(identity)

/** Evaluates up to `maxConcurrent` inner effects concurrently, emitting
* the results in order.
*/
def parEvalFlatten(
maxConcurrent: Int
)(implicit F: Concurrent[F]): Stream[F, O] = self.parEvalMap(maxConcurrent)(identity)
}

/** Provides syntax for pure streams. */
implicit final class PureOps[O](private val self: Stream[Pure, O]) extends AnyVal {

Expand Down
2 changes: 1 addition & 1 deletion io/shared/src/test/scala/fs2/io/net/unixsocket/UnixSocketsSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ class UnixSocketsSuite extends Fs2Suite with UnixSocketsSuitePlatform {

val clients = (0 until 100).map(b => client(Chunk.singleton(b.toByte)))

(Stream.sleep_[IO](1.second) ++ Stream.emits(clients).evalMap(identity))
(Stream.sleep_[IO](1.second) ++ Stream.emits(clients).evalFlatten)
.concurrently(server)
.compile
.drain
Expand Down