Replace thread worker/sup with poolboy worker pool

Author: craigp

config/dev.exs.sample

@@ -8,3 +8,6 @@ config :gmail, :oauth2, [
  token_type: "Bearer",
  expires_at: 1428085947
 ]
+
+config :gmail, :thread,
+ pool: 100

lib/gmail/base.ex

@@ -54,9 +54,9 @@ defmodule Gmail.Base do
  @spec handle_error(map) :: atom | {atom, String.t} | {atom, map} | map
  def handle_error(response) do
  case response do
- {:ok, %{"error" => %{"code" => 404}} } ->
+ {:ok, %{"error" => %{"code" => 404}}} ->
  {:error, :not_found}
- {:ok, %{"error" => %{"code" => 400, "errors" => errors}} } ->
+ {:ok, %{"error" => %{"code" => 400, "errors" => errors}}} ->
  [%{"message" => error_message}|_rest] = errors
  {:error, error_message}
  {:ok, %{"error" => details}} ->

lib/gmail/history.ex

@@ -24,9 +24,9 @@ defmodule Gmail.History do
  """
  def handle_history_response(response) do
  case response do
- {:ok, %{"error" => %{"code" => 404}} } ->
+ {:ok, %{"error" => %{"code" => 404}}} ->
  :not_found
- {:ok, %{"error" => %{"code" => 400, "errors" => errors}} } ->
+ {:ok, %{"error" => %{"code" => 400, "errors" => errors}}} ->
  {:error, errors}
  {:ok, %{"history" => history}} ->
  {:ok, Helper.atomise_keys(history)}

lib/gmail/http.ex

@@ -17,40 +17,40 @@ defmodule Gmail.HTTP do
 
  def execute({:get, url, path}, %{access_token: access_token}) do
  (url <> path)
- |> HTTPoison.get(get_headers(access_token))
+ |> HTTPoison.get(get_headers(access_token), timeout: :infinity, recv_timeout: :infinity)
  |> do_parse_response
  end
 
  def execute({:post, url, path, data}, %{access_token: access_token}) do
  {:ok, json} = encode(data)
  (url <> path)
- |> HTTPoison.post(json, get_headers(access_token))
+ |> HTTPoison.post(json, get_headers(access_token), timeout: :infinity, recv_timeout: :infinity)
  |> do_parse_response
  end
 
  def execute({:post, url, path}, %{access_token: access_token}) do
  (url <> path)
- |> HTTPoison.post("", get_headers(access_token))
+ |> HTTPoison.post("", get_headers(access_token), timeout: :infinity, recv_timeout: :infinity)
  |> do_parse_response
  end
 
  def execute({:delete, url, path}, %{access_token: access_token}) do
  (url <> path)
- |> HTTPoison.delete(get_headers(access_token))
+ |> HTTPoison.delete(get_headers(access_token), timeout: :infinity, recv_timeout: :infinity)
  |> do_parse_response
  end
 
  def execute({:put, url, path, data}, %{access_token: access_token}) do
  {:ok, json} = encode(data)
  (url <> path)
- |> HTTPoison.put(json, get_headers(access_token))
+ |> HTTPoison.put(json, get_headers(access_token), timeout: :infinity, recv_timeout: :infinity)
  |> do_parse_response
  end
 
  def execute({:patch, url, path, data}, %{access_token: access_token}) do
  {:ok, json} = encode(data)
  (url <> path)
- |> HTTPoison.patch(json, get_headers(access_token))
+ |> HTTPoison.patch(json, get_headers(access_token), timeout: :infinity, recv_timeout: :infinity)
  |> do_parse_response
  end
 
@@ -67,8 +67,8 @@ defmodule Gmail.HTTP do
  :ok
  end
 
- defp do_parse_response({:error, %HTTPoison.Error{id: id, reason: reason}}) do
-{:error, reason}
+ defp do_parse_response({:error, %HTTPoison.Error{reason: reason}}) do
+ {:error, reason}
  end
 
  @spec get_headers(String.t) :: [{String.t, String.t}]

lib/gmail/supervisor.ex

@@ -15,7 +15,7 @@ defmodule Gmail.Supervisor do
  def init(:ok) do
  children = [
  supervisor(Gmail.UserManager, []),
- supervisor(Gmail.Thread.Supervisor, []),
+ supervisor(Gmail.Thread.Pool, []),
  supervisor(Gmail.Message.Supervisor, [])
  ]
  supervise(children, strategy: :one_for_one)

lib/gmail/thread.ex

@@ -85,7 +85,7 @@ defmodule Gmail.Thread do
  @doc """
  Handles a thread resource response from the Gmail API.
  """
-@spec handle_thread_response(atom | {atom, map | String.t}) :: {atom, String.t | map}
+ @spec handle_thread_response(atom | {atom, map | String.t}) :: {atom, String.t | map}
  def handle_thread_response(response) do
  response
  |> handle_error
@@ -104,7 +104,7 @@ defmodule Gmail.Thread do
  @doc """
  Handles a thread list response from the Gmail API.
  """
-@spec handle_thread_list_response(atom | {atom, map | String.t}) :: {atom, String.t | map} | {atom, map, String.t}
+ @spec handle_thread_list_response(atom | {atom, map | String.t}) :: {atom, String.t | map} | {atom, map, String.t}
  def handle_thread_list_response(response) do
  response
  |> handle_error

lib/gmail/thread/pool.ex

@@ -0,0 +1,54 @@
+defmodule Gmail.Thread.Pool do
+
+ @moduledoc """
+ A pool of workers for handling thread operations.
+ """
+
+ alias Gmail.Thread.PoolWorker
+
+ @default_pool_size 20
+
+ @doc false
+ def start_link do
+ poolboy_config = [
+ {:name, {:local, :thread_pool}},
+ {:worker_module, PoolWorker},
+ {:size, pool_size},
+ {:max_overflow, 0}
+ ]
+
+ children = [
+ :poolboy.child_spec(:thread_pool, poolboy_config, [])
+ ]
+
+ options = [
+ strategy: :one_for_one,
+ name: __MODULE__
+ ]
+
+ Supervisor.start_link(children, options)
+ end
+
+ @doc """
+ Gets a thread.
+ """
+ @spec get(String.t, String.t, map, map) :: {atom, map}
+ def get(user_id, thread_id, params, state) do
+ :poolboy.transaction(
+ :thread_pool,
+ fn pid ->
+ PoolWorker.get(pid, user_id, thread_id, params, state)
+ end,
+ :infinity)
+ end
+
+ def pool_size do
+ case Application.get_env(:gmail, :thread) do
+ [pool: size] when is_integer(size) ->
+ size
+ _ ->
+ @default_pool_size
+ end
+ end
+
+end

lib/gmail/thread/pool_worker.ex

@@ -0,0 +1,38 @@
+defmodule Gmail.Thread.PoolWorker do
+
+ @moduledoc """
+ A thread pool worker.
+ """
+
+ use GenServer
+ alias Gmail.{Thread, User}
+
+ @doc false
+ def start_link([]) do
+ GenServer.start_link(__MODULE__, [], [])
+ end
+
+ @doc false
+ def init(state) do
+ {:ok, state}
+ end
+
+ @doc false
+ def handle_call({:get, user_id, thread_id, params, state}, _from, worker_state) do
+ result =
+ user_id
+ |> Thread.get(thread_id, params)
+ |> User.http_execute(state)
+ |> Thread.handle_thread_response
+ {:reply, result, worker_state}
+ end
+
+ @doc """
+ Gets a thread.
+ """
+ @spec get(pid, String.t, String.t, map, map) :: {atom, map}
+ def get(pid, user_id, thread_id, params, state) do
+ GenServer.call(pid, {:get, user_id, thread_id, params, state}, :infinity)
+ end
+
+end