Skip to content
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

Jump to bottom

Stream.zip for more than two streams #3411

Open
nounder opened this issue Aug 4, 2024 · 2 comments
Open

Stream.zip for more than two streams #3411

nounder opened this issue Aug 4, 2024 · 2 comments
Labels
enhancement New feature or request

Comments

@nounder
Copy link

nounder commented Aug 4, 2024 •
edited
Loading

What version of Effect is running?

3.6.0

What steps can reproduce the bug?

import { Effect, Stream } from "effect"

const stream = Stream.make(1, 2, 3).pipe(
	Stream.broadcast(3, 10),
	Stream.map(([s1, s2, s3]) => ({
		n1: s1,
		n2: s2.pipe(Stream.map((v) => v * 2)),
		n3: s2.pipe(Stream.map((v) => v * 3)),
	})),
	Stream.flatMap(({ n1, n2, n3 }) =>
		Stream.zip(n1, n2, n3), // TYPESCRIPT REPORTS: Expected 1-2 arguments, but got 3. ts (2554) [11, 22]
	),
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)

What is the expected behavior?

Passing more than 2 streams to Stream.zip works per documentation :

Zipping is a process of combining two or more streams to create a new stream by pairing elements from the input streams.

What do you see instead?

Expected 1-2 arguments, but got 3. ts (2554) [11, 22]

when passing more than 2 streams to Stream.zip

Additional information

No response
@nounder nounder added the bug Something isn't working label Aug 4, 2024
@nounder
Copy link
Author

nounder commented Aug 5, 2024

Here"s workaround using Effect.all:

import { Effect, Stream } from "effect"

const stream = Stream.make(1, 2, 3).pipe(
	Stream.mapEffect((v) =>
		Effect.all(
			{
				n1: Effect.succeed(v),
				n2: Effect.succeed(v * 2),
				n3: Effect.succeed(v * 3),
			},
			{ concurrency: "unbounded" },
		),
	),
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)

Unfortunately, this makes consuming stream slower because it has to wait for all effects to finalize for each element.

@mikearnaldi mikearnaldi added enhancement New feature or request and removed bug Something isn't working labels Aug 14, 2024
@mikearnaldi
Copy link
Member

Seems like we are missing a Stream.all

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Assignees
No one assigned
Labels
enhancement New feature or request
Projects
None yet
Milestone
No milestone
Development

No branches or pull requests
2 participants
@nounder @mikearnaldi