buffer

lib/behavior/gen_observable.ex

@@ -121,6 +121,12 @@ defmodule Observables.GenObservable do
         cast(self(), :stop)
         {:noreply, state}
 
+      {:ok, :done, {:value, value}} ->
+        Logger.debug("#{inspect(self())} stopping.")
+        cast(self(), {:notify_all, value})
+        cast(self(), :stop)
+        {:noreply, state}
+
       {:ok, :continue} ->
         Logger.debug("#{inspect(self())} going on.")
         # Remove observee.

lib/obs.ex

@@ -10,7 +10,8 @@ defmodule Observables.Obs do
     Distinct,
     Each,
     Filter,
-    StartsWith
+    StartsWith,
+    Buffer
   }
 
   alias Enum
@@ -20,7 +21,7 @@ defmodule Observables.Obs do
   # GENERATORS ###################################################################
 
   @doc """
-  from_pid/1 can be considered to be a subject. Any process that implements 
+  from_pid/1 can be considered to be a subject. Any process that implements
   """
   def from_pid(producer) do
     {fn consumer ->
@@ -184,6 +185,23 @@ defmodule Observables.Obs do
      end, pid}
   end
 
+  @doc """
+  Periodically gather items emitted by an Observable into bundles of size `size` and emit
+  these bundles rather than emitting the items one at a time.
+
+  Source: http://reactivex.io/documentation/operators/buffer.html
+  """
+  def buffer({observable_fn, _parent_pid}, size) do
+    {:ok, pid} = GenObservable.start_link(Buffer, [size])
+
+    observable_fn.(pid)
+
+    # Create the continuation.
+    {fn observer ->
+      GenObservable.send_to(pid, observer)
+    end, pid}
+  end
+
   # TERMINATORS ##################################################################
 
   def print({observable_fn, parent_pid}) do

lib/observables/buffer.ex

@@ -0,0 +1,31 @@
+defmodule Observables.Buffer do
+  @moduledoc false
+  use Observables.GenObservable
+
+  def init([size]) do
+    Logger.debug("Buffer: #{inspect(self())}")
+    # We don't keep state in merge.
+    {:ok, %{:size => size, :buffer => []}}
+  end
+
+  def handle_event(v, %{:size => size, :buffer => bs}) do
+    count = Enum.count(bs)
+
+    case {bs, size, count} do
+      {b, s, c} when s <= c ->
+        {:value, b, %{:size => s, :buffer => [v]}}
+
+      {b, s, c} when s > c ->
+        {:novalue, %{:size => s, :buffer => b ++ [v]}}
+        # {[], max_size, c} ->
+        #   {:novalue, %{:max_size => max_size, :buffer => [v]}}
+    end
+  end
+
+  def handle_done(pid, %{:size => size, :buffer => bs}) do
+
+    # We are buffering, and if our dependency stops we need to flush our buffer or we are losing values.
+    Logger.debug("#{inspect(self())}: dependency stopping: #{inspect(pid)}")
+    {:ok, :done, {:value, bs}}
+  end
+end