make observables stop if the upstream stops

lib/behavior/gen_observable.ex

@@ -134,10 +134,10 @@ defmodule Observables.GenObservable do
           state.listeningto
           |> Enum.filter(fn sub -> sub != pid end)
 
-        # if count(new_subs) == 0 do
-        #   Logger.warn("#{inspect(self())} all dependencies done, stopping ourselves.")
-        #   cast(self(), :stop)
-        # end
+        if count(new_subs) == 0 do
+          Logger.warn("#{inspect(self())} all dependencies done, stopping ourselves.")
+          cast(self(), :stop)
+        end
 
         {:noreply, %{state | listeningto: new_subs}}
     end
@@ -248,7 +248,7 @@ defmodule Observables.GenObservable do
   @doc """
   Makes an observable stop and gracefully shut down.
   """
-  def stop(producer, reason \\ :normal) do
+  def stop(producer, _reason \\ :normal) do
     # cast(producer, {:stop, reason})
     cast(producer, :stop)
   end

lib/observables/delay.ex

@@ -23,6 +23,6 @@ defmodule Observables.Operator.Delay do
 
   def handle_done(pid, _state) do
     Logger.debug("#{inspect(self())}: dependency stopping: #{inspect(pid)}")
-    {:ok, :continue}
+    {:ok, :done}
   end
 end

lib/observables/distinct.ex

@@ -19,6 +19,6 @@ defmodule Observables.Operator.Distinct do
 
   def handle_done(pid, _state) do
     Logger.debug("#{inspect(self())}: dependency stopping: #{inspect(pid)}")
-    {:ok, :continue}
+    {:ok, :done}
   end
 end

lib/observables/distinct_until_changed.ex

@@ -19,6 +19,6 @@ defmodule Observables.Operator.DistinctUntilChanged do
 
   def handle_done(pid, _state) do
     Logger.debug("#{inspect(self())}: dependency stopping: #{inspect(pid)}")
-    {:ok, :continue}
+    {:ok, :done}
   end
 end

lib/observables/each.ex

@@ -15,6 +15,6 @@ defmodule Observables.Operator.Each do
 
   def handle_done(pid, _state) do
     Logger.debug("#{inspect(self())}: dependency stopping: #{inspect(pid)}")
-    {:ok, :continue}
+    {:ok, :done}
   end
 end

lib/observables/filter.ex

@@ -18,6 +18,6 @@ defmodule Observables.Operator.Filter do
 
   def handle_done(pid, _state) do
     Logger.debug("#{inspect(self())}: dependency stopping: #{inspect(pid)}")
-    {:ok, :continue}
+    {:ok, :done}
   end
 end

lib/observables/from_enum.ex

@@ -19,6 +19,6 @@ defmodule Observables.Operator.FromEnum do
 
   def handle_done(pid, _state) do
     Logger.debug("#{inspect(self())}: dependency stopping: #{inspect(pid)}")
-    {:ok, :continue}
+    {:ok, :done}
   end
 end

lib/observables/map.ex

@@ -14,6 +14,6 @@ defmodule Observables.Operator.Map do
 
   def handle_done(pid, _state) do
     Logger.debug("#{inspect(self())}: dependency stopping: #{inspect(pid)}")
-    {:ok, :continue}
+    {:ok, :done}
   end
 end

lib/observables/merge_continuous.ex

@@ -24,6 +24,7 @@ defmodule Observables.Operator.MergeContinuous do
   end
 
   def handle_done(_pid, state) do
+    # TODO
     {:ok, state}
   end
 end

lib/observables/range.ex

@@ -31,6 +31,6 @@ defmodule Observables.Operator.Range do
 
   def handle_done(pid, _state) do
     Logger.debug("#{inspect(self())}: dependency stopping: #{inspect(pid)}")
-    {:ok, :continue}
+    {:ok, :done}
   end
 end

lib/observables/starts_with.ex

@@ -14,6 +14,6 @@ defmodule Observables.Operator.StartsWith do
 
   def handle_done(pid, _state) do
     Logger.debug("#{inspect(self())}: dependency stopping: #{inspect(pid)}")
-    {:ok, :continue}
+    {:ok, :done}
   end
 end

lib/observables/take.ex

@@ -18,6 +18,6 @@ defmodule Observables.Operator.Take do
 
   def handle_done(pid, _state) do
     Logger.debug("#{inspect(self())}: dependency stopping: #{inspect(pid)}")
-    {:ok, :continue}
+    {:ok, :done}
   end
 end

lib/subject.ex

@@ -18,4 +18,8 @@ defmodule Observables.Subject do
   def next(pid, v) do
     GenObservable.send_event(pid, v)
   end
+
+  def complete(pid) do
+    GenObservable.stop(pid, "complete")
+  end
 end

mix.lock

@@ -1,4 +1,4 @@
 %{
-  "earmark": {:hex, :earmark, "1.2.5", "4d21980d5d2862a2e13ec3c49ad9ad783ffc7ca5769cf6ff891a4553fbaae761", [:mix], [], "hexpm"},
-  "ex_doc": {:hex, :ex_doc, "0.18.3", "f4b0e4a2ec6f333dccf761838a4b253d75e11f714b85ae271c9ae361367897b7", [:mix], [{:earmark, "~> 1.1", [hex: :earmark, repo: "hexpm", optional: false]}], "hexpm"},
+  "earmark": {:hex, :earmark, "1.2.5", "4d21980d5d2862a2e13ec3c49ad9ad783ffc7ca5769cf6ff891a4553fbaae761", [:mix], [], "hexpm", "c57508ddad47dfb8038ca6de1e616e66e9b87313220ac5d9817bc4a4dc2257b9"},
+  "ex_doc": {:hex, :ex_doc, "0.18.3", "f4b0e4a2ec6f333dccf761838a4b253d75e11f714b85ae271c9ae361367897b7", [:mix], [{:earmark, "~> 1.1", [hex: :earmark, repo: "hexpm", optional: false]}], "hexpm", "33d7b70d87d45ed899180fb0413fb77c7c48843188516e15747e00fdecf572b6"},
 }