l3nz

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

sabiwara

Elixir Core Team

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

sabiwara

Elixir Core Team

It only reads the elements it needs:

  • find is 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 :wink:):

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” Enum functions like map/2 / filter/2 that are returning a new list => the Stream equivalent would avoid building the list and return a new stream instead
  • “consumer” Enum functions like reduce/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 the Stream module.
  • “lazy consumer” Enum functions like reduce_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 the Stream module.
LostKobrakai

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.

Where Next?

Popular in Questions Top

sergio
In Ruby, I can go: User.find_by(email: "foobar@email.com").update(email: "hello@email.com") How can I do something similar in Elixir? ...
New
dotdotdotPaul
Okay, I'm having a heck of a time trying to figure out how to best handle the validation of belongs_to associations in Ecto. I'm sure I'...
New
pgiesin
This should be a simple problem but I just can’t seem to figure it out. I have a standalone Elixir app that won’t find the database. Dep...
New
jononomo
I am trying to figure out how Mix knows whether the environment is test, dev, or prod -- where is this set? Thanks.
New
script
If I have a string “1000 cfu/ml” . I want to remove the characters and / and space . So the string is like this "1000" What is the ...
New
vac
Hi, I'm quite new in Elixir and I'm trying to format a string to a PEM format. I have the certificate value like MIIDBTCCAe2...... and ...
New
ashish173
I am using Ecto timestamps with postgres, I can see the timestamps() use the :naive_dateime but for my use case I wanted to store the ti...
New
qwerescape
Is there a way to get the call stack or stack trace at any point in the code? Not from exceptions, but an expression that returns how the...
New
Qqwy
Original source of discussion: This topic on the Pragmatic Programmers' Functional Web Development with Elixir, OTP, and Phoenix forum. ...
New
joeerl
Hello again - after a longish gap I’ve decided I really must dig into Elixir and see what’s been happening here - so I have a few questio...
New

Other popular topics Top

srinivasu
How to handle excepions in elixir? Suppose i have A, B, C ,D, E modules. and each module has get() function. A.get() method will call th...
New
dotdotdotPaul
Okay, I'm having a heck of a time trying to figure out how to best handle the validation of belongs_to associations in Ecto. I'm sure I'...
New
lessless
I believe there are people here who are dealing with CSV files import on the daily basis, and since Excel is a really popular tool there ...
New
KronicDeth
Elixir plugin for JetBrain’s IntelliJ Platform (including Rubymine) This is a plugin that adds support for Elixir to JetBrains IntelliJ...
289 35421 110
New
SoCreat
i’m a new one to elixir which editor can i use vs code? or atom? Thanks! :smiley:
New
chrismccord
This release brings a number of exciting features, including integration with the new Phoenix LiveDashboard and Phoenix LiveView. There h...
New
mgjohns61585
Could someone help me? I'm making my first elixir program, number guessing game. I can't figure out how to convert the user's guess from ...
New
Patoshizzle
After calling mix ecto.create I get this error: 17:00:32.162 [error] GenServer #PID<0.412.0> terminating ** (Postgrex.Error) FATAL...
New
romenigld
I am trying to run a deploy with docker and I successfully runned with this command: docker build -t romenigld/blog-prod . but when I t...
New
siddhant3030
Hi, I have to write a raw query for one of my project. But till now I have used ecto queries and don’t have much experience writing raw ...
New

We're in Beta

About us Mission Statement