Implement call cluster distribution

Author: 7h0ma5

lib/call/application.ex

@@ -17,6 +17,7 @@ defmodule Call.Application do
       ),
       worker(Call.RabbitMQ, [], restart: :permanent),
       worker(Call.Database, [], restart: :permanent),
+      worker(Call.Dispatcher, [], restart: :permanent),
       worker(DapnetService.CouchDB, [], restart: :permanent),
     ]
 

lib/call/dispatch.ex

@@ -1,5 +1,19 @@
-defmodule Call.Dispatch do
+defmodule Call.Dispatcher do
+  use GenServer
+
   def dispatch(call) do
+    GenServer.call(__MODULE__, {:dispatch, call})
+  end
+
+  def start_link do
+    GenServer.start_link(__MODULE__, {}, [name: __MODULE__])
+  end
+
+  def init(_opts) do
+    {:ok, nil}
+  end
+
+  def handle_call({:dispatch, call}, _from, state) do
     recipients = Map.get(call, "recipients", %{})
 
 #    subscribers = Map.get(recipients, "subscribers", [])
@@ -32,6 +46,8 @@ defmodule Call.Dispatch do
         end)
       end)
     end)
+
+    {:reply, :ok, state}
   end
 
   def send_call(call, transmitter, pager) do

lib/call/rabbitmq.ex

@@ -21,7 +21,7 @@ defmodule Call.RabbitMQ do
   end
 
   def init(_opts) do
-    connect
+    connect()
   end
 
   def handle_call({:publish_call, call}, _from, chan) do
@@ -52,16 +52,20 @@ defmodule Call.RabbitMQ do
   def handle_info({:basic_deliver, payload, %{delivery_tag: tag, routing_key: key}}, chan) do
     :ok = Basic.ack chan, tag
 
-    data = Poison.decode!(payload)
-
+    case Poison.decode(payload) do
+      {:ok, call} ->
+        Call.Dispatcher.dispatch(call)
+        Call.Database.store call
+      _ -> Logger.warn("Failed to decode remote call")
+    end
 
     {:noreply, chan}
   end
 
   # Automatic Reconnect
   def handle_info({:DOWN, _, :process, _pid, _reason}, _) do
     Logger.warn("RabbitMQ connection closed.")
-    {:ok, chan} = connect
+    {:ok, chan} = connect()
     {:noreply, chan}
   end
 
@@ -91,7 +95,7 @@ defmodule Call.RabbitMQ do
         Logger.error("RabbitMQ connection failed.")
         # Reconnection loop
         :timer.sleep(10000)
-        connect
+        connect()
     end
   end
 

lib/call/router.ex

@@ -21,7 +21,7 @@ defmodule Call.Router do
         |> Map.put("created_on", Timex.now())
         |> Map.put("created_by", user)
 
-        Call.Dispatch.dispatch call
+        Call.Dispatcher.dispatch call
         Call.Database.store call
 
         json_call = Poison.encode!(call)
@@ -35,8 +35,6 @@ defmodule Call.Router do
   end
 
   get "/calls" do
-    IO.inspect conn
-
     case Call.Database.list() do
       nil ->
         send_resp(conn, 404, '{"error": "Not found"}')