delay observable - beta

lib/obs.ex

@@ -16,7 +16,8 @@ defmodule Observables.Obs do
     Scan,
     Take,
     CombineLatest,
-    CombineLatestSilent
+    CombineLatestSilent,
+    Delay
   }
 
   alias Enum
@@ -420,6 +421,21 @@ defmodule Observables.Obs do
      end, pid}
   end
 
+  @doc """
+  Delays the given observable for n miliseconds. 
+  
+  More information: http://reactivex.io/documentation/operators/delay.html
+  """
+  def delay({observable_fn, _parent_pid}, n) do
+    {:ok, pid} = GenObservable.start(Delay, [n])
+
+    observable_fn.(pid)
+
+    {fn observer ->
+       GenObservable.send_to(pid, observer)
+     end, pid}
+  end
+
   # TERMINATORS ##################################################################
 
   @doc """

lib/observables/delay.ex

@@ -0,0 +1,25 @@
+defmodule Observables.Operator.Delay do
+  @moduledoc false
+  use Observables.GenObservable
+
+  def init([delay]) do
+    {:ok, %{:delay => delay}}
+  end
+
+  def handle_event(v, state = %{:delay => delay}) do
+    Logger.warn "Got this: #{inspect v}"
+    case v do
+      {:delayed, x} ->
+        {:value, x, state}
+      v ->
+        Process.send_after(self(), {:event, {:delayed, v}}, delay)
+        {:novalue, state}
+    end
+
+  end
+
+  def handle_done(pid, _state) do
+    Logger.debug("#{inspect(self())}: dependency stopping: #{inspect(pid)}")
+    {:ok, :continue}
+  end
+end

test/delay_test.exs

@@ -0,0 +1,36 @@
+defmodule DelayTest do
+  use ExUnit.Case
+  alias Observables.{Obs}
+  require Logger
+
+  @tag :delay
+  test "delay" do
+    Code.load_file("test/util.ex")
+    testproc = self()
+
+    # Create a range.
+    Obs.range(1, 5, 100)
+    |> Obs.delay(1000)
+    |> Obs.each(fn v ->
+      Logger.error "Got #{v}"
+      send(testproc, v)
+    end)
+
+    # Receive no other values.
+    # receive do
+    #   _x ->
+    #     Logger.error("Received another value, did not want")
+    #     assert "received another value:" == ""
+    # after
+    #   5000 ->
+    #     :ok
+    # end
+
+    # 1..5
+    # |> Enum.map(fn v ->
+    #   assert_receive(^v, 5000, "did not get this message #{v}")
+    # end)
+    Test.Util.sleep(100000)
+
+  end
+end

test/unsubscribe_test.exs

@@ -6,14 +6,14 @@ defmodule UnsubscribeTest do
   @tag :unsubscribe
   test "ubsubscribe" do
     Code.load_file("test/util.ex")
-    
+
     testproc = self()
 
     Obs.range(1, :infinity, 500)
     |> Obs.map(fn v ->
       if v > 5 do
         Observables.Observable.unsubscribe()
-      else 
+      else
         send(testproc, v)
       end
     end)