unsubscribe test works

lib/behavior/gen_observable.ex

@@ -186,6 +186,20 @@ defmodule Observables.GenObservable do
     {:noreply, state}
   end
 
+  def handle_cast(:unsubscribe, state) do
+    # First notify all our dependees that we no longer want their data.
+    # This has as a side effect qwe no longer listen to them as well.
+    state.listeningto
+    |> Enum.map(fn producer ->
+      stop_send_to(producer, self())
+    end)
+
+    # Now it should be safe to stop.
+    GenObservable.stop(self())
+
+    {:noreply, state}
+  end
+
   def handle_info({:event, value}, state) do
     cast(self(), {:event, value})
     {:noreply, state}
@@ -235,7 +249,15 @@ defmodule Observables.GenObservable do
   Makes an observable stop and gracefully shut down.
   """
   def stop(producer, reason \\ :normal) do
-    cast(producer, {:stop, reason})
+    #cast(producer, {:stop, reason})
+    cast(producer, :stop)
+  end
+
+  @doc """
+  Unsubscribes ourself from all our dependencies.
+  """
+  def unsubscribe(producer) do
+    cast(producer, :unsubscribe)
   end
 
   ###########

lib/subject.1.ex

@@ -0,0 +1,11 @@
+defmodule Observables.Observable do 
+    alias Observables.GenObservable 
+
+    @moduledoc """
+    WRITE ME PLEASE KIND SIR
+    """
+
+    def unsubscribe() do
+        GenObservable.unsubscribe(self())
+    end
+end

test/unsubscribe_test.exs

@@ -0,0 +1,28 @@
+defmodule UnsubscribeTest do
+  use ExUnit.Case
+  alias Observables.{Obs}
+  require Logger
+
+  @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 
+        send(testproc, v)
+      end
+    end)
+
+    Test.Util.sleep(2000)
+
+    1..5
+    |> Enum.map(fn v ->
+      assert_receive(^v, 5000, "did not get this message #{v}")
+    end)
+  end
+end