Stages refactoring

lib/weather_station_manager.ex

@@ -10,9 +10,9 @@ defmodule WeatherStationManager do
     children = [
       supervisor(WeatherStationManager.Repo, []),
       supervisor(WeatherStationManager.Endpoint, []),
-      worker(Tizio.Source, []),
-      worker(Tizio.DBManager, []),
-      worker(Tizio.TwitterManager, [])
+      worker(UpdatesManager, []),
+      worker(UpdatesDBWorker, []),
+      worker(UpdatesTwitterWorker, [])
     ]
 
     # See http://elixir-lang.org/docs/stable/elixir/Supervisor.html

test/controllers/updates_controller_test.exs

@@ -13,11 +13,10 @@ defmodule WeatherStationManager.UpdatesControllerTest do
   end
 
   test "the update is broadcasted to the workers", %{conn: conn} do
-    with_mock Tizio.Source, [broadcast_update: fn(_spam) -> :eggs end] do
+    with_mock UpdatesManager, [broadcast_update: fn(_spam) -> :eggs end] do
       post conn, "/api/updates", @update
 
-      assert called Tizio.Source.broadcast_update(@update)
+      assert called UpdatesManager.broadcast_update(@update)
     end
   end
 end
-

web/controllers/updates_controller.ex

@@ -2,11 +2,10 @@ defmodule WeatherStationManager.UpdatesController do
   use WeatherStationManager.Web, :controller
 
   def create(conn, params) do
-    Tizio.Source.broadcast_update(params)
+    UpdatesManager.broadcast_update(params)
 
     conn
     |> put_status(201)
     |> json(%{ status: "OK" })
   end
 end
-

web/stages/updates_db_worker.ex

@@ -0,0 +1,21 @@
+defmodule UpdatesDBWorker do
+  use GenStage
+
+  require Logger
+
+  def start_link() do
+    GenStage.start_link(__MODULE__, :ok)
+  end
+
+  def init(:ok) do
+    { :consumer, :ok, subscribe_to: [ UpdatesManager ] }
+  end
+
+  def handle_events(events, _from, state) do
+    Logger.debug "[STAGE] - Write to DB: START"
+    Enum.each events, fn(_spam) -> Logger.debug "processing SQL..." end
+    Logger.debug "[STAGE] - Write to DB: DONE"
+
+    { :noreply, [], state }
+  end
+end

web/stages/updates_manager.ex

@@ -0,0 +1,25 @@
+defmodule UpdatesManager do
+  use GenStage
+
+  require Logger
+
+  def start_link() do
+    GenStage.start_link(__MODULE__, :ok, name: __MODULE__)
+  end
+
+  def init(:ok) do
+    { :producer, :ok, dispatcher: GenStage.BroadcastDispatcher }
+  end
+
+  def handle_demand(_demand, state) do
+    { :noreply, [], state }
+  end
+
+  def broadcast_update(update) do
+    GenStage.call(__MODULE__, { :broadcast, update })
+  end
+
+  def handle_call({ :broadcast, update }, _from, state) do
+    { :reply, %{ :ok => "Weather update received." }, [ update ], state }
+  end
+end

web/stages/updates_twitter_worker.ex

@@ -0,0 +1,21 @@
+defmodule UpdatesTwitterWorker do
+  use GenStage
+
+  require Logger
+
+  def start_link() do
+    GenStage.start_link(__MODULE__, :ok)
+  end
+
+  def init(:ok) do
+    { :consumer, :ok, subscribe_to: [ UpdatesManager ] }
+  end
+
+  def handle_events(events, _from, state) do
+    Logger.debug "[STAGE] - Tweet about it: START"
+    Enum.each events, fn(_spam) -> Logger.debug "processing tweet..." end
+    Logger.debug "[STAGE] - Tweet about it: DONE"
+
+    { :noreply, [], state }
+  end
+end