l3nz
Stream: return the element I'm halting on
I have a (possibly infinite) stream of events coming asynchronously, so that they do have a delay. I want to stop the stream immediately when l I find a value that’s divisible by seven, and return it.
I tried with Stream.transform/5, because I need to return the very value I’m halting the stream on. So I would expect to :halt on the correct value (in this case 42) and that the “last” function would be called with value 42, so I could emit it. But I see the “after” function being called - at which time it’s too late to emit a value - but not the “last” function.
Stream.cycle([1, 2, 5, 6, 23, 42, 19])
|> Stream.transform(
fn -> [] end,
fn v, a ->
if rem(v, 7) == 0 do
{:halt, v}
else
{[v], a}
end
end,
fn a -> IO.puts("last #{a}") end,
fn a -> IO.puts("after #{a}") end
)
|> Enum.take(10)
Anybody has any suggestions? How would you do that?
Marked As Solved
sabiwara
This seems to be describing Enum.find/2, would the following work for your use case?
Stream.cycle([1, 2, 5, 6, 23, 42, 19])
|> Enum.find(& rem(&1, 7) == 0)
Also Liked
sabiwara
It only reads the elements it needs:
findis lazy and will halt as soon as it finds something- it doesn’t need to build a new list
You can confirm it as follows (plus the fact it doesn’t consume infinite time and memory
):
Stream.cycle([1, 2, 5, 6, 23, 42, 19])
|> Stream.map(&IO.inspect/1)
|> Enum.find(& rem(&1, 7) == 0)
The only issue is a potential infinite loop, if the condition is never met.
Enum and Stream both work with any enumerable inputs including streams, but you can think of most functions as roughly part of one of its categories:
- “producer”
Enumfunctions likemap/2/filter/2that are returning a new list => theStreamequivalent would avoid building the list and return a new stream instead - “consumer”
Enumfunctions likereduce/3,sum/1,max/1, that are returning an accumulator and not a list => can be used to consume non-infinite streams, they have no equivalent in theStreammodule. - “lazy consumer”
Enumfunctions likereduce_while/2,find/2,any?/2, that halt early to return a value => can be used to consume potential infinite streams, they have no equivalent in theStreammodule.
LostKobrakai
This?
[1, 2, 3, 4, 5]
|> Stream.transform(false, fn
x, false -> {[x], x == 3}
_, true -> {:halt, true}
end)
|> Enum.into([])
# [1, 2, 3]
No, it won’t. Both Enum and Stream iterate their inputs reduction by reduction/enumeration by enumeration. The difference is their output. Stream apis return lazy enumerables. Enum api eagerly calculate their results.







