Stream.interval/1 and Stream.timer/1

[Peter Hamilton]

From another thread, I mentioned that timer based streams were useful in my Streamz codebase (tldr, Streamz is exploring more reactive programming in Elixir). The two function I have built and would like to add to Elixir's Stream module are:

Stream.interval(t) - produces a stream that emits value `:ok` every `t` milliseconds. Generally a building block for other functionality. Example:

min_max_memory_usage_per_minute = Stream.interval(1000)
  |> Stream.map fn(_) -> get_memory_usage end
  |> Stream.chunk(60) # group every 60 seconds together
  |> Stream.map &( {Enum.min(&1), Enum.max(&1)} )
  |> Enum.each fn ({min, max) ->
    send_to_graphite("min_memory_usage", min)
    send_to_graphite("max_memory_usage", max)
  end

Stream.timer(t) - Produces a stream that emits one value `:ok` every `t` milliseconds. Shorthand for `Stream.interval(t) |> Stream.take(1)`. Very useful with Streamz.take_until/2, but I can't think of a whole lot of other uses off the top of my head. It is the compliment to interval and since it is trivial I think it should be included.

So a few things to mention:

1. Consuming these streams blocks execution. It's sort of the point. Other streams do as well, such as GenEvent.stream.
   
2. There are numerous tweaks we can make. Should the functions emit something user specified rather than :ok? A value? A function? Mapping or scanning the stream is simple enough and so I'm inclined to leave it at :ok.
3. Compare these to :timer.apply_interval/4 and :timer.apply_after/4. The biggest difference is that it works with the Enum/Stream modules and you can return the result (if you use Enum.take, otherwise it's an infinite stream).
4. Maybe Stream.timer should be called Stream.after?

Thoughts on this? This functionality is a standard part of other reactive libraries in other languages and I've found at least a few uses for them. I'd like to see them added to the standard library.

[José Valim]

+1 for adding those two functions.

Stream.interval(t) - produces a stream that emits value `:ok` every `t` milliseconds. Generally a building block for other functionality.

I would follow the Rx API and make Stream.interval/0 be an increasing counter. It is at least more useful than :ok.

 

Stream.timer(t) - Produces a stream that emits one value `:ok` every `t` milliseconds. Shorthand for `Stream.interval(t) |> Stream.take(1)`. Very useful with Streamz.take_until/2, but I can't think of a whole lot of other uses off the top of my head. It is the compliment to interval and since it is trivial I think it should be included.

Rx returns 0 for Stream.timer/1 but it could be :ok too. I don't have a strong feeling on this one. Also, we should definitely call it Stream.timer/1 since after is a keyword (so Stream.after/1 is kind of a no-go).

I am really glad that we are start to merge some of the Streamz stuff back into Elixir.

[Paul Schoenfelder]

I'm a fan. I've used Rx.NET's Observables pretty extensively, and this is one of the first things I used in that API. So +1 from me.

I do think the value for interval should be able to be provided by the user. I like the idea of using a function to generate the value for each interval. :ok seems like a good default though.

Paul

--
You received this message because you are subscribed to the Google Groups "elixir-lang-core" group.
To unsubscribe from this group and stop receiving emails from it, send an email to elixir-lang-co...@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Peter Hamilton]

In master already. :)

`Stream.interval(1000) |> Stream.map some_fun`  works well without much overhead. The version merged returns a counter.

[Paul Schoenfelder]

That's excellent, I already have an idea for something to build with this :).

Paul