Setup HTTPoison option recv_timeout based on consul wait option

lib/consul.ex

@@ -26,4 +26,3 @@ defmodule Consul do
     Supervisor.start_link([], opts)
   end
 end
-

lib/consul/agent.ex

@@ -18,37 +18,31 @@ defmodule Consul.Agent do
 
   @spec checks(Keyword.t) :: Endpoint.response
   def checks(opts \\ []) do
-    build_url([@agent, @checks], opts)
-      |> req_get()
+    req_get([@agent, @checks], opts)
   end
 
   @spec join(binary, Keyword.t) :: Endpoint.response
   def join(address, opts \\ []) do
-    build_url([@agent, @join, address], opts)
-      |> req_get()
+    req_get([@agent, @join, address], opts)
   end
 
   @spec force_leave(binary, Keyword.t) :: Endpoint.response
   def force_leave(node, opts \\ []) do
-    build_url([@agent, @force_leave, node], opts)
-      |> req_get()
+    req_get([@agent, @force_leave, node], opts)
   end
 
   @spec members(Keyword.t) :: Endpoint.response
   def members(opts \\ []) do
-    build_url([@agent, @members], opts)
-      |> req_get()
+    req_get([@agent, @members], opts)
   end
 
   @spec self(Keyword.t) :: Endpoint.response
   def self(opts \\ []) do
-    build_url([@agent, @self], opts)
-      |> req_get()
+    req_get([@agent, @self], opts)
   end
 
   @spec services(Keyword.t) :: Endpoint.response
   def services(opts \\ []) do
-    build_url([@agent, @services], opts)
-      |> req_get()
+    req_get([@agent, @services], opts)
   end
 end

lib/consul/agent/service.ex

@@ -15,13 +15,11 @@ defmodule Consul.Agent.Service do
 
   @spec register(map, Keyword.t) :: Endpoint.response
   def register(%{"Name" => _} = body, opts \\ []) do
-    build_url([@agent, @service, @register], opts)
-      |> req_put(Poison.encode!(body))
+    req_put([@agent, @service, @register], opts, Poison.encode!(body))
   end
 
   @spec deregister(binary, Keyword.t) :: Endpoint.response
   def deregister(id, opts \\ []) do
-    build_url([@agent, @service, @deregister, id], opts)
-      |> req_delete()
+    req_delete([@agent, @service, @deregister, id], opts)
   end
 end

lib/consul/catalog.ex

@@ -18,37 +18,31 @@ defmodule Consul.Catalog do
 
   @spec datacenters(Keyword.t) :: Endpoint.response
   def datacenters(opts \\ []) do
-    build_url([@catalog, @datacenters], opts)
-      |> req_get()
+    req_get([@catalog, @datacenters], opts)
   end
 
   @spec deregister(map, Keyword.t) :: Endpoint.response
   def deregister(%{"Datacenter" => _, "Node" => _} = body, opts \\ []) do
-    build_url([@catalog, @deregister], opts)
-      |> req_put(Poison.encode!(body))
+    req_put([@catalog, @deregister], Poison.encode!(body), opts)
   end
 
   @spec nodes(Keyword.t) :: Endpoint.response
   def nodes(opts \\ []) do
-    build_url([@catalog, @nodes], opts)
-      |> req_get()
+    req_get([@catalog, @nodes], opts)
   end
 
   @spec node(binary, Keyword.t) :: Endpoint.response
   def node(id, opts \\ []) do
-    build_url([@catalog, @node, id], opts)
-      |> req_get()
+    req_get([@catalog, @node, id], opts)
   end
 
   @spec services(Keyword.t) :: Endpoint.response
   def services(opts \\ []) do
-    build_url([@catalog, @services], opts)
-      |> req_get()
+    req_get([@catalog, @services], opts)
   end
 
   @spec service(binary, Keyword.t) :: Endpoint.response
   def service(name, opts \\ []) do
-    build_url([@catalog, @service, name], opts)
-      |> req_get()
+    req_get([@catalog, @service, name], opts)
   end
 end

lib/consul/endpoint.ex

@@ -6,6 +6,7 @@
 
 defmodule Consul.Endpoint do
   @path_separator "/"
+  @timeout_delay  1000
 
   @type response :: {:ok, Consul.Response.t} | {:error, Consul.Response.t}
 
@@ -21,36 +22,66 @@ defmodule Consul.Endpoint do
     path <> "?" <> URI.encode_query(opts)
   end
 
+  @spec build_options(Keyword.t, Keyword.t) :: Keyword.t
+  def build_options(params, options) do
+    case Keyword.get(params, :wait) do
+      nil -> options
+      dur -> Keyword.merge([recv_timeout: parse_duration(dur) * 1000 + @timeout_delay], options)
+    end
+  end
+
+
+  defp parse_duration(s) when is_binary(s), do: parse_duration(s, {[], 0})
+
+  defp parse_duration(<<>>, {_, sec}), do: sec
+  defp parse_duration(<< x, rest :: binary >>, {buf, sec}) when x >= ?0 and x <= ?9,
+    do: parse_duration(rest, {[x|buf], sec})
+  defp parse_duration(<< ?s, rest :: binary >>, {buf, sec}),
+    do: parse_duration(rest, {[], sec + List.to_integer(Enum.reverse(buf))})
+  defp parse_duration(<< ?m, rest :: binary >>, {buf, sec}),
+    do: parse_duration(rest, {[], sec + List.to_integer(Enum.reverse(buf)) * 60})
+  defp parse_duration(<< ?h, rest :: binary >>, {buf, sec}),
+    do: parse_duration(rest, {[], sec + List.to_integer(Enum.reverse(buf)) * 3600})
+  defp parse_duration(_, _), do: raise(ArgumentError, "Invalid consul duration")
+
   defmacro __using__(handler: handler) do
     quote do
       alias Consul.Request, as: Request
-      import unquote(__MODULE__), only: [build_url: 1, build_url: 2]
+      import unquote(__MODULE__), only: [build_url: 1, build_url: 2, build_options: 2]
 
-      defp req_get(url, headers \\ [], options \\ []) do
+      defp req_get(path, params \\ [], headers \\ [], options \\ []) do
+        url     = build_url(path, params)
+        options = build_options(params, options)
         Request.get(url, headers, options) |> handle_response
       end
 
-      defp req_put(url, body, headers \\ [], options \\ []) do
+      defp req_put(path, body, params \\ [], headers \\ [], options \\ []) do
+        url = build_url(path, params)
         Request.put(url, body, headers, options) |> handle_response
       end
 
-      defp req_head(url, headers \\ [], options \\ []) do
+      defp req_head(path, params \\ [], headers \\ [], options \\ []) do
+        url = build_url(path, params)
         Request.head(url, headers, options) |> handle_response
       end
 
-      defp req_post(url, body, headers \\ [], options \\ []) do
+      defp req_post(path, body, params \\ [], headers \\ [], options \\ []) do
+        url = build_url(path, params)
         Request.post(url, body, headers, options) |> handle_response
       end
 
-      defp req_patch(url, body, headers \\ [], options \\ []) do
+      defp req_patch(path, body, params \\ [], headers \\ [], options \\ []) do
+        url = build_url(path, params)
         Request.patch(url, body, headers, options) |> handle_response
       end
 
-      defp req_delete(url, headers \\ [], options \\ []) do
+      defp req_delete(path, params \\ [], headers \\ [], options \\ []) do
+        url = build_url(path, params)
         Request.delete(url, headers, options) |> handle_response
       end
 
-      defp req_options(url, headers \\ [], options \\ []) do
+      defp req_options(path, params \\ [], headers \\ [], options \\ []) do
+        url = build_url(path, params)
         Request.options(url, headers, options) |> handle_response
       end
 

lib/consul/event.ex

@@ -33,8 +33,7 @@ defmodule Consul.Event do
 
   @spec fire(binary, binary, Keyword.t) :: Endpoint.response
   def fire(name, payload \\ "", opts \\ []) when is_binary(payload) do
-    build_url([@event, @fire, name], opts)
-      |> req_put(payload)
+    req_put([@event, @fire, name], payload, opts)
   end
 
   @doc """
@@ -57,8 +56,7 @@ defmodule Consul.Event do
 
   @spec list(Keyword.t) :: Endpoint.response
   def list(opts \\ []) do
-    build_url([@event, @list], opts)
-      |> req_get()
+    req_get([@event, @list], opts)
   end
 
   @doc """

lib/consul/health.ex

@@ -16,25 +16,22 @@ defmodule Consul.Health do
 
   @spec checks(binary, Keyword.t) :: Endpoint.response
   def checks(id, opts \\ []) do
-    build_url([@health, @checks, id], opts)
-      |> req_get()
+    req_get([@health, @checks, id], opts)
   end
 
   @spec node(binary, Keyword.t) :: Endpoint.response
   def node(id, opts \\ []) do
-    build_url([@health, @node, id], opts)
-      |> req_get()
+    req_get([@health, @node, id], opts)
   end
 
   @spec service(binary, Keyword.t) :: Endpoint.response
   def service(id, opts \\ []) do
-    build_url([@health, @service, id], opts)
+    req_get([@health, @service, id], opts)
       |> req_get()
   end
 
   @spec state(binary, Keyword.t) :: Endpoint.response
   def state(id, opts \\ []) do
-    build_url([@health, @state, id], opts)
-      |> req_get()
+    req_get([@health, @state, id], opts)
   end
 end

lib/consul/kv.ex

@@ -13,9 +13,7 @@ defmodule Consul.Kv do
 
   @spec fetch(binary | [binary], Keyword.t) :: Endpoint.response
   def fetch(key, opts \\ []) do
-    List.flatten([@kv, key])
-      |> build_url(opts)
-      |> req_get()
+    List.flatten([@kv, key]) |> req_get(opts)
   end
 
   @spec fetch!(binary | [binary], Keyword.t) :: Response.t | no_return
@@ -30,9 +28,7 @@ defmodule Consul.Kv do
 
   @spec keys(binary | [binary]) :: Endpoint.response
   def keys(prefix) do
-    List.flatten([@kv, prefix])
-      |> build_url(keys: true)
-      |> req_get()
+    List.flatten([@kv, prefix]) |> req_get(keys: true)
   end
 
   @spec keys!(binary | [binary]) :: Response.t | no_return
@@ -47,7 +43,7 @@ defmodule Consul.Kv do
 
   @spec put(binary | [binary], term, Keyword.t) :: boolean
   def put(key, value, opts \\ []) do
-    case List.flatten([@kv, key]) |> build_url(opts) |> req_put(to_string(value)) do
+    case List.flatten([@kv, key]) |> req_put(to_string(value), opts) do
       {:ok, %{body: body}} ->
         body
       error ->

lib/consul/session.ex

@@ -18,8 +18,7 @@ defmodule Consul.Session do
 
   @spec create(map, Keyword.t) :: Endpoint.response
   def create(body, opts \\ []) do
-    build_url([@session, @create], opts)
-      |> req_put(Poison.encode!(body))
+    req_put([@session, @create], Poison.encode!(body), opts)
   end
 
   @spec create!(map, Keyword.t) :: binary | no_return
@@ -34,31 +33,26 @@ defmodule Consul.Session do
 
   @spec destroy(binary, Keyword.t) :: Endpoint.response
   def destroy(session_id, opts \\ []) do
-    build_url([@session, @destroy, session_id], opts)
-      |> req_put("")
+    req_put([@session, @destroy, session_id], "", opts)
   end
 
   @spec info(binary, Keyword.t) :: Endpoint.response
   def info(session_id, opts \\ []) do
-    build_url([@session, @info, session_id], opts)
-      |> req_get()
+    req_get([@session, @info, session_id], opts)
   end
 
   @spec node(binary, Keyword.t) :: Endpoint.response
   def node(node_id, opts \\ []) do
-    build_url([@session, @node, node_id], opts)
-      |> req_get()
+    req_get([@session, @node, node_id], opts)
   end
 
   @spec list(Keyword.t) :: Endpoint.response
   def list(opts \\ []) do
-    build_url([@session, @list], opts)
-      |> req_get()
+    req_get([@session, @list], opts)
   end
 
   @spec renew(binary, Keyword.t) :: Endpoint.response
   def renew(session_id, opts \\ []) do
-    build_url([@session, @renew, session_id], opts)
-      |> req_put("")
+    req_put([@session, @renew, session_id], "", opts)
   end
 end