Generic HTTP/2 workers

lib/pigeon.ex

@@ -9,11 +9,12 @@ defmodule Pigeon do
 
   def start(_type, _args) do
     opts = [strategy: :one_for_one, name: :pigeon]
+    Application.put_env(:chatterbox, :ssl_options, [])
     Supervisor.start_link(workers(), opts)
   end
 
   defp workers do
-    adm_worker() ++ apns_workers()
+    adm_worker() ++ apns_workers() ++ gcm_workers()
   end
 
   def adm_worker do
@@ -38,4 +39,13 @@ defmodule Pigeon do
       true -> []
     end
   end
+
+  defp gcm_workers do
+    name = Pigeon.GCMWorker.default_name()
+    cond do
+      config = Application.get_env(:pigeon, :gcm) ->
+        [worker(Pigeon.GCMWorker, [name, config], id: name)]
+      true -> []
+    end
+  end
 end

lib/pigeon/apns.ex

@@ -12,11 +12,12 @@ defmodule Pigeon.APNS do
 
   def push(notification, opts \\ [])
   def push(notification, opts) when is_list(notification) do
+    timeout = opts[:timeout] || @default_timeout
     case opts[:on_response] do
       nil ->
         tasks = for n <- notification, do: Task.async(fn -> do_sync_push(n, opts) end)
         tasks
-        |> Task.yield_many(@default_timeout + 500)
+        |> Task.yield_many(timeout + 500)
         |> Enum.map(fn {task, response} -> response || Task.shutdown(task, :brutal_kill) end)
         |> group_responses
       on_response -> push(notification, on_response, opts)
@@ -31,15 +32,17 @@ defmodule Pigeon.APNS do
 
   defp do_sync_push(notification, opts) do
     pid = self()
-    on_response = fn(x) -> send pid, {:ok, x} end
+    ref = :erlang.make_ref
+    on_response = fn(x) -> send pid, {ref, x} end
+    timeout = opts[:timeout] || @default_timeout
 
     worker_name = opts[:name] || Config.default_name
     GenServer.cast(worker_name, {:push, :apns, notification, on_response})
 
     receive do
-      {:ok, x} -> x
+      {^ref, x} -> x
     after
-      @default_timeout -> {:error, :timeout, notification}
+      timeout -> {:error, :timeout, notification}
     end
   end
 

lib/pigeon/apns_config.ex

@@ -3,11 +3,18 @@ defmodule Pigeon.APNS.Config do
     Validates configuration settings that initialize APNSWorkers.
   """
 
-  def default_name, do: :default
+  def default_name, do: :apns_default
 
   def config(name) do
     config = Application.get_env(:pigeon, :apns)[name]
+    config(name, config)
+  end
+
+  def config(name, config) do
     %{
+      production_endpoint: config[:production_endpoint] ||  "api.push.apple.com",
+      development_endpoint: config[:development_endpoint] ||  "api.development.push.apple.com",
+      port: config[:port] ||  443,
       name: name,
       mode: config[:mode],
       cert: cert(config[:cert]),

lib/pigeon/apns_worker.ex

@@ -2,95 +2,42 @@ defmodule Pigeon.APNSWorker do
   @moduledoc """
     Handles all APNS request and response parsing over an HTTP2 connection.
   """
-  use GenServer
+  use Pigeon.GenericH2Worker, ping_interval: 600_000
   require Logger
 
-  @ping_period 600_000 # 10 minutes
-
-  defp apns_production_api_uri, do: "api.push.apple.com"
-  defp apns_development_api_uri, do: "api.development.push.apple.com"
-
-  def push_uri(mode) do
-    case mode do
-      :dev -> apns_development_api_uri()
-      :prod -> apns_production_api_uri()
+  def host(config) do
+    case config[:mode] do
+      :dev -> config[:development_endpoint]
+      :prod -> config[:production_endpoint]
     end
   end
 
-  def start_link(config) do
-    GenServer.start_link(__MODULE__, {:ok, config}, name: config[:name])
-  end
-
-  def stop, do: :gen_server.cast(self(), :stop)
-
-  def init({:ok, config}), do: initialize_worker(config)
-
-  def initialize_worker(config) do
-    mode = config[:mode]
-    case connect_socket(config, 0) do
-      {:ok, socket} ->
-        Process.send_after(self(), :ping, @ping_period)
-        {:ok, %{
-          apns_socket: socket,
-          mode: mode,
-          config: config,
-          stream_id: 1,
-          queue: %{}
-        }}
-      {:closed, _socket} ->
-        Logger.error """
-          Socket closed unexpectedly.
-          """
-        {:stop, {:error, :bad_connection}}
-      {:error, :timeout} ->
-        Logger.error """
-          Failed to establish SSL connection. Is the certificate signed for :#{mode} mode?
-          """
-        {:stop, {:error, :timeout}}
-      {:error, :invalid_config} ->
-        Logger.error """
-          Invalid configuration.
-          """
-        {:stop, {:error, :invalid_config}}
+  def port(config) do
+    case config[:use_2197] do
+      true -> 2197
+      _ -> config[:port]
     end
   end
 
-  def connect_socket(_config, 3), do: {:error, :timeout}
-  def connect_socket(config, tries) do
-    uri = config[:mode] |> push_uri |> to_char_list
-    case connect_socket_options(config) do
-      {:ok, options} -> do_connect_socket(config, uri, options, tries)
-      error -> error
-    end
+  def encode_notification(notification) do
+    Pigeon.Notification.json_payload(notification.payload)
   end
 
-  def connect_socket_options(config) do
+  def socket_options(config) do
     cert = get_opt(config, :cert, :certfile)
     key = get_opt(config, :key, :keyfile)
     cond do
       cert && key ->
         options =
           [cert,
           key,
-          {:password, ''},
-          {:packet, 0},
-          {:reuseaddr, true},
-          {:active, true},
-          :binary]
-          |> optional_add_2197(config)
+          {:password, ''}]
         {:ok, options}
       true ->
         {:error, :invalid_config}
     end
   end
 
-  defp optional_add_2197(options, config) do
-    case config[:use_2197] do
-      true -> options ++ [{:port, 2197}]
-      _ -> options
-    end
-  end
-
   defp get_opt(config, key_1, key_2) do
     cond do
       config[key_1] -> {key_1, config[key_1]}
@@ -99,44 +46,11 @@ defmodule Pigeon.APNSWorker do
     end
   end
 
-  defp do_connect_socket(config, uri, options, tries) do
-    case Kadabra.open(uri, :https, options) do
-      {:ok, socket} -> {:ok, socket}
-      {:error, reason} ->
-        Logger.error(inspect(reason))
-        connect_socket(config, tries + 1)
-    end
-  end
-
-  def handle_cast(:stop, state), do: { :noreply, state }
-
-  def handle_cast({:push, :apns, notification}, state) do
-    send_push(state, notification, nil)
-  end
-
-  def handle_cast({:push, :apns, notification, on_response}, state) do
-    send_push(state, notification, on_response)
-  end
-
-  def handle_cast(msg, state) do
-    Logger.debug "Recv: #{inspect(msg)}"
-    {:noreply, state}
-  end
-
-  def send_push(state, notification, on_response) do
-    %{apns_socket: socket, stream_id: stream_id, queue: queue} = state
-    json = Pigeon.Notification.json_payload(notification.payload)
-    req_headers = [
-      {":method", "POST"},
-      {":path", "/3/device/#{notification.device_token}"},
-      {"content-length", "#{byte_size(json)}"}]
-      |> put_apns_id(notification)
-      |> put_apns_topic(notification)
-
-    Kadabra.request(socket, req_headers, json)
-    new_q = Map.put(queue, "#{stream_id}", {notification, on_response})
-    new_stream_id = stream_id + 2
-    { :noreply, %{state | stream_id: new_stream_id, queue: new_q } }
+  def req_headers(_config, notification) do
+    [{"content-type", "application/json"},
+     {"accept", "application/json"}]
+    |> put_apns_id(notification)
+    |> put_apns_topic(notification)
   end
 
   defp put_apns_id(headers, notification) do
@@ -153,16 +67,24 @@ defmodule Pigeon.APNSWorker do
     end
   end
 
-  defp parse_error(data) do
-    {:ok, response} = Poison.decode(data)
+  def req_path(notification) do
+    "/3/device/#{notification.device_token}"
+  end
+
+  defp parse_error(_notification, _headers, body) do
+    {:ok, response} = Poison.decode(body)
     response["reason"] |> Macro.underscore |> String.to_existing_atom
   end
 
-  defp log_error(reason, notification) do
-    Logger.error("#{reason}: #{error_msg(reason)}\n#{inspect(notification)}")
+  defp parse_response(notification, headers, _body) do
+    case List.keyfind(headers, "apns-id", 1) do
+      nil -> notification
+      apns_id ->
+        %{notification | id: apns_id}
+    end
   end
 
-  def error_msg(error) do
+  def error_msg(_code, error) do
     case error do
       # 400
       :bad_collapse_id ->
@@ -216,7 +138,7 @@ defmodule Pigeon.APNSWorker do
       :missing_provider_token ->
         """
         No provider certificate was used to connect to APNs and Authorization header was missing
-        or no provider token was specified." 
+        or no provider token was specified."
         """
 
       # 404
@@ -257,51 +179,4 @@ defmodule Pigeon.APNSWorker do
         ""
     end
   end
-
-  def handle_info(:ping, state) do
-    Kadabra.ping(state.apns_socket)
-    Process.send_after(self(), :ping, @ping_period)
-
-    { :noreply, state }
-  end
-
-  def handle_info({:end_stream, %Kadabra.Stream{id: stream_id, headers: headers, body: body}},
-                                                %{apns_socket: _socket, queue: queue} = state) do
-
-    {notification, on_response} = queue["#{stream_id}"]
-
-    case get_status(headers) do
-      "200" ->
-        notification = %{notification | id: get_apns_id(headers)}
-        unless on_response == nil do on_response.({:ok, notification}) end
-        new_queue = Map.delete(queue, "#{stream_id}")
-        {:noreply, %{state | queue: new_queue}}
-      nil ->
-        {:noreply, state}
-      _error ->
-        reason = parse_error(body)
-        log_error(reason, notification)
-        unless on_response == nil do on_response.({:error, reason, notification}) end
-        new_queue = Map.delete(queue, "#{stream_id}")
-        {:noreply, %{state | queue: new_queue}}
-    end
-  end
-
-  def handle_info({:ping, _from}, state), do: {:noreply, state}
-
-  def handle_info({:ok, _from}, state), do: {:noreply, state}
-
-  defp get_status(headers) do
-    case Enum.find(headers, fn({key, _val}) -> key == ":status" end) do
-      {":status", status} -> status
-      nil -> nil
-    end
-  end
-
-  defp get_apns_id(headers) do
-    case Enum.find(headers, fn({key, _val}) -> key == "apns-id" end) do
-      {"apns-id", id} -> id
-      nil -> nil
-    end
-  end
 end