5 replies

Dealing with Infinite Streams

Would love some pointers with the pattern when dealing with infinite streams. Consider the code:

import { Chunk, Effect, Sink, Stream } from "effect";

const program = Effect.gen(function* (_) {
  const infiniteNumberStream = Stream.iterate(1, (n) => n + 1)
  const lines = yield* _(infiniteNumberStream.pipe(Stream.run(Sink.take(6))));
  console.log(lines.pipe(Chunk.toArray));
})

Effect.runPromise(program);

import { Chunk, Effect, Sink, Stream } from "effect";

const program = Effect.gen(function* (_) {
  const infiniteNumberStream = Stream.iterate(1, (n) => n + 1)
  const lines = yield* _(infiniteNumberStream.pipe(Stream.run(Sink.take(6))));
  console.log(lines.pipe(Chunk.toArray));
})

Effect.runPromise(program);

We could also write this as

const lines = yield* _(infiniteNumberStream.pipe(Stream.take(10), Stream.runCollect));

const lines = yield* _(infiniteNumberStream.pipe(Stream.take(10), Stream.runCollect));

You will get a nice

[ 1, 2, 3, 4, 5, 6 ]

[ 1, 2, 3, 4, 5, 6 ]

as you'd expect. You, however, need to know in advance how many you want to take. I feel this is a pretty niche usecase.

Far more often you create the stream at T0, events and life happens whilst the stream accumulates data, then want to do "something" with those elements in that stream at a given moment in time T1. What at some high level patterns around this?

Effect Community•3y ago•

5 replies

timbotronic

Dealing with Infinite Streams

Would love some pointers with the pattern when dealing with infinite streams. Consider the code:

import { Chunk, Effect, Sink, Stream } from "effect";

const program = Effect.gen(function* (_) {
  const infiniteNumberStream = Stream.iterate(1, (n) => n + 1)
  const lines = yield* _(infiniteNumberStream.pipe(Stream.run(Sink.take(6))));
  console.log(lines.pipe(Chunk.toArray));
})

Effect.runPromise(program);

import { Chunk, Effect, Sink, Stream } from "effect";

const program = Effect.gen(function* (_) {
  const infiniteNumberStream = Stream.iterate(1, (n) => n + 1)
  const lines = yield* _(infiniteNumberStream.pipe(Stream.run(Sink.take(6))));
  console.log(lines.pipe(Chunk.toArray));
})

Effect.runPromise(program);

We could also write this as

const lines = yield* _(infiniteNumberStream.pipe(Stream.take(10), Stream.runCollect));

const lines = yield* _(infiniteNumberStream.pipe(Stream.take(10), Stream.runCollect));

You will get a nice

[ 1, 2, 3, 4, 5, 6 ]

[ 1, 2, 3, 4, 5, 6 ]

Dealing with Infinite Streams

Similar Threads

Dealing with Infinite Streams

Similar Threads

Similar Threads

Similar Threads