range can be derived from the rest

lib/obs.ex

@@ -12,7 +12,8 @@ defmodule Observables.Obs do
     Distinct,
     Each,
     Filter,
-    StartsWith
+    StartsWith,
+    Repeat
   }
 
   alias Enum
@@ -56,6 +57,20 @@ defmodule Observables.Obs do
      end, pid}
   end
 
+  @doc """
+  repeat takes a function as argument and an optional interval.
+  The function will be repeatedly executed, and the result will be emitted as an event.
+  """
+  def repeat(f, opts \\ []) do
+    interval = Keyword.get(opts, :interval, 1000)
+    times = Keyword.get(opts, :times, :infinity)
+
+    range(1, times, interval)
+    |> map(fn _ ->
+      f.()
+    end)
+  end
+
   # CONSUMER AND PRODUCER ########################################################
 
   def zip(l, r) do

lib/observables/range.ex

@@ -14,6 +14,12 @@ defmodule Observables.Range do
         state = %{:first => first, :last => last, :current => current, :delay => delay}
       ) do
     case {current, last} do
+      {current, :infinity} ->
+        Process.send_after(self(), {:event, :tick}, delay)
+
+        {:value, current,
+         %{:first => first, :last => last, :current => current + 1, :delay => delay}}
+
       {current, last} when current > last ->
         {:done, state}
 

test/observables_test.exs

@@ -304,4 +304,31 @@ defmodule ObservablesTest do
 
     sleep(5000)
   end
+
+  @tag :repeat
+  test "repeat" do
+    testproc = self()
+
+    Obs.repeat(fn ->
+      send(testproc, :hello)
+    end, [interval: 500, times: 5])
+
+    [:hello, :hello, :hello, :hello, :hello]
+    |> Enum.map(fn x ->
+      receive do
+        ^x -> Logger.debug("Got #{x}")
+      end
+    end)
+
+    receive do
+      x ->
+        Logger.error("Received another value, did not want")
+        assert "received another value: #{inspect(x)}" == ""
+    after
+      1000 ->
+        :ok
+    end
+
+    assert 5 == 5
+  end
 end