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

vonH
In asking this question I am more interested about the expressiveness of the language itself and less concerned about the availability of...
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
logicmason
Hi there, I'm working through my first release with elixir/phoenix. I've built a release with distillery and found that it crashes when I...
New
freewebwithme
Using vs code and installed ElixirLS: support and debugger. And I got an error popped up on start up says Failed to run ‘elixir’ comma...
New
myronmarston
The Elixir Typespec docs show the following syntax for keyword lists in typespecs: # ... | [key: type] # keyword lis...
New
makeitrein
Hey all, just started picking up Elixir last week and am writing a scraper as a learning project. Baby step #1 is extracting the number ...
New
alice
Hey, Just curious what are the main benefits of Elixir compared to Clojure? When is Elixir more useful than Clojure and vice versa? Th...
New
beno
I will often find my self writing things similar to: case some_value do nil -> something() "" -> something() _ -> someth...
New
sabri
Can someone explain the settings of pool_size of Ecto in config file? and what is the recommend size? Thanks
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

Other popular topics Top

chrismccord
Phoenix 1.4.0 released Phoenix 1.4 is out! This release ships with exciting new features, most notably with HTTP2 support, improved deve...
688 30048 115
New
yurko
Here are few pieces of (common) Linux knowledge that we use for reasonably small one server apps. We use Ubuntu but this should work for ...
New
albydarned
Hello all! I am typing this post from my new MacBook Pro with the M1 chip. I’m loving it so far, and will probably use it as my daily dr...
New
jerry
Good day to you all. I have been struggling to get a query involving like and ilike to work. Can anyone assist me on this, please? pro...
New
sergio_101
I am VERY much an elixir newbie. I have taken one elixir course and one phoenix course on Udemy. During that course, I saw the instructor...
New
nsuchy
Hi. I’ve noticed that Windows Powershell has it’s own IEX command and you cannot access Elixir’s IEX due to the conflict. This isn’t a cr...
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
chrismccord
As promised, the first release candidate of Phoenix 1.3.0 is out! This release focuses on code generators with improved project structure...
New
mcarvalho
What is the difference between System.get_env and Application.get_env? For example, what are best practices to use one versus another.
New
johnnyicon
Hi all, I've just started learning Elixir and Phoenix Framework, so please pardon my n00bness at this stage. I'm trying to use Postg...
New

We're in Beta

About us Mission Statement