Merge branch 'WHISTLE-788'

ecallmgr/src/ecallmgr_amqp_pool_worker.erl

@@ -164,6 +164,21 @@ handle_info({'DOWN', Ref, process, Pid, _Info}, #state{status=busy, ref=Ref, par
     ecallmgr_amqp_pool:worker_free(Parent, self(), 0),
     {noreply, #state{}};
 
+handle_info({timeout, ReqRef, req_timeout}, #state{status=busy, from=From, parent=Parent, ref=Ref
+                                ,start=Start, req_ref=ReqRef
+                               }) ->
+    ?LOG("request took too long, timing out caller"),
+    Elapsed = timer:now_diff(erlang:now(), Start),
+    ?LOG("received response after ~b ms, returning to pool ~p", [Elapsed div 1000, Parent]),
+
+    erlang:demonitor(Ref, [flush]),
+    erlang:cancel_timer(ReqRef),
+
+    gen_server:reply(From, {error, timeout}),
+
+    ecallmgr_amqp_pool:worker_free(Parent, self(), Elapsed),
+    {noreply, #state{}};
+
 handle_info(req_timeout, #state{status=busy, from=From, parent=Parent, ref=Ref
                                 ,start=Start, req_ref=ReqRef
                                }) ->

lib/whistle-1.0.0/src/gen_listener.erl

@@ -334,7 +334,7 @@ handle_info({amqp_host_down, _H}=Down, #state{bindings=Bindings, params=Params}=
                 {ok, Q} ->
                     Self = self(),
                     _ = erlang:send_after(?TIMEOUT_RETRY_CONN, Self, is_consuming),
-                    spawn(fun() -> [ add_binding(Self, Type, BindProps) || {Type, BindProps} <- Bindings ] end),
+                    proc_lib:spawn(fun() -> [ add_binding(Self, Type, BindProps) || {Type, BindProps} <- Bindings ] end),
                     {noreply, State#state{queue=Q, is_consuming=false}, hibernate};
                 {error, _} ->
                     ?LOG("failed to start amqp, waiting another second"),
@@ -411,17 +411,17 @@ process_req(#state{queue=Queue, responders=Responders, module=Module, module_sta
     case Props1 of
         ignore -> ignore;
         _Else ->
-            spawn_link(fun() -> _ = wh_util:put_callid(JObj), process_req(Props1, Responders, JObj) end)
+            proc_lib:spawn_link(fun() -> _ = wh_util:put_callid(JObj), process_req(Props1, Responders, JObj) end)
     end.
 
 -spec process_req/3 :: (wh_proplist(), responders(), wh_json:json_object()) -> 'ok'.
 process_req(Props, Responders, JObj) ->
     Key = wh_util:get_event_type(JObj),
 
-    Handlers = [spawn_monitor(fun() ->
-                                      _ = wh_util:put_callid(JObj),
-                                      Responder:Fun(JObj, Props)
-                              end)
+    Handlers = [proc_lib:spawn_monitor(fun() ->
+                                               _ = wh_util:put_callid(JObj),
+                                               Responder:Fun(JObj, Props)
+                                       end)
                 || {Evt, {Responder, Fun}} <- Responders,
                    maybe_event_matches_key(Key, Evt)
                ],
@@ -464,7 +464,7 @@ start_amqp(Props) ->
 stop_amqp(<<>>, _) -> ok;
 stop_amqp(Q, Bindings) ->
     Self = self(),
-    spawn(fun() -> [ gen_listener:rm_binding(Self, Type, Prop) || {Type, Prop} <- Bindings] end),
+    proc_lib:spawn(fun() -> [ gen_listener:rm_binding(Self, Type, Prop) || {Type, Prop} <- Bindings] end),
     amqp_util:queue_delete(Q).
 
 -spec set_qos/1 :: ('undefined' | non_neg_integer()) -> 'ok'.

lib/whistle_couch-1.0.0/src/couch_compactor.erl

@@ -25,7 +25,7 @@ start_link() ->
 
 init(Parent) ->
     case {couch_config:fetch(compact_automatically), couch_config:fetch(conflict_strategy)} of
-        {true, undefined} ->
+        {true, null} ->
             ?LOG_SYS("just compacting"),
             proc_lib:init_ack(Parent, {ok, self()}),
             compact_all();
@@ -36,7 +36,7 @@ init(Parent) ->
         {false, _Strategy} ->
             ?LOG_SYS("auto-compaction not enabled"),
             proc_lib:init_ack(Parent, ignore);
-        {undefined, _Strategy} ->
+        {null, _Strategy} ->
             ?LOG_SYS("auto-compaction not enabled"),
             proc_lib:init_ack(Parent, ignore),
             couch_config:store(compact_automatically, false)

lib/whistle_couch-1.0.0/src/couch_config.erl

@@ -14,7 +14,7 @@
 %%%-------------------------------------------------------------------
 -module(couch_config).
 
--export([start_link/0]).
+-export([start_link/0, ready/0]).
 -export([load_config/1, write_config/1]).
 -export([fetch/1, fetch/2]).
 -export([store/2, store/3]).
@@ -33,11 +33,11 @@ start_link() ->
 load_config(Path) ->
     ?LOG("loading ~s", [Path]),
     case file:consult(Path) of
-	{ok, Startup} ->
-	    _ = [cache_from_file(T) || T <- Startup],
-	    ok;
-	{error, enoent}=E ->
-	    E
+        {ok, Startup} ->
+            _ = [cache_from_file(T) || T <- Startup],
+            ok;
+        {error, enoent}=E ->
+            E
     end.
 
 %% convert 3..n-tuples to 2 tuples with the value being (3..n)-1 tuples
@@ -53,9 +53,12 @@ cache_from_file(T) when is_tuple(T) ->
 -spec write_config/1 :: (file:name()) -> 'ok' | {'error', file:posix() | 'badarg' | 'terminated' | 'system_limit'}.
 write_config(Path) ->
     Contents = lists:foldl(fun(I, Acc) -> [io_lib:format("~p.~n", [I]) | Acc] end
-			   , "", whapps_config:get_all_kvs(?CONFIG_CAT)),
+                           , "", whapps_config:get_all_kvs(?CONFIG_CAT)),
     file:write_file(Path, Contents).
 
+ready() ->
+    whapps_config:couch_ready().
+
 fetch(Key) ->
     fetch(Key, undefined).
 
@@ -66,8 +69,8 @@ fetch(Key, Default) ->
 
 fetch(Key, Default, Cache) ->
     case wh_cache:fetch_local(Cache, {?MODULE, Key}) of
-	{error, not_found} -> Default;
-	{ok, V} -> V
+        {error, not_found} -> Default;
+        {ok, V} -> V
     end.
 
 -spec store/2 :: (term(), term()) -> 'ok'.

lib/whistle_couch-1.0.0/src/couch_mgr.erl

@@ -390,8 +390,8 @@ ensure_saved(DbName, Doc, Options) ->
 save_doc(DbName, Doc, Opts) ->
     couch_util:save_doc(get_conn(), DbName, Doc, Opts).
 
--spec save_docs/2 :: (ne_binary(), wh_json:json_objects()) -> {'ok', wh_json:json_objects()} | {'error', atom()}.
--spec save_docs/3 :: (ne_binary(), wh_json:json_objects(), proplist()) -> {'ok', wh_json:json_objects()} | {'error', atom()}.
+-spec save_docs/2 :: (ne_binary(), wh_json:json_objects()) -> {'ok', wh_json:json_objects()}.
+-spec save_docs/3 :: (ne_binary(), wh_json:json_objects(), proplist()) -> {'ok', wh_json:json_objects()}.
 save_docs(DbName, Docs) when is_list(Docs) ->
     save_docs(DbName, Docs, []).
 save_docs(DbName, Docs, Opts) when is_list(Docs) ->
@@ -513,7 +513,14 @@ get_creds() ->
     gen_server:call(?SERVER, get_creds).
 
 get_conn() ->
-    gen_server:call(?SERVER, get_conn).
+    case whereis(?SERVER) of
+        Srv when is_pid(Srv) ->
+            gen_server:call(?SERVER, get_conn);
+        _E ->
+            ?LOG("no server by the name of ~s", [?SERVER]),
+            ST = erlang:get_stacktrace(),
+            ?LOG_STACKTRACE(ST)
+    end.
 
 get_admin_conn() ->
     gen_server:call(?SERVER, get_admin_conn).
@@ -749,6 +756,8 @@ init_state() ->
 
 init_state_from_config(undefined) ->
     init_state_from_config({"localhost", ?DEFAULT_PORT, "", "", ?DEFAULT_ADMIN_PORT});
+init_state_from_config(null) ->
+    init_state_from_config({"localhost", ?DEFAULT_PORT, "", "", ?DEFAULT_ADMIN_PORT});
 init_state_from_config(H) when not is_tuple(H) ->
     init_state_from_config({H, ?DEFAULT_PORT, "", "", ?DEFAULT_ADMIN_PORT});
 init_state_from_config({H, Port}) ->
@@ -762,6 +771,7 @@ init_state_from_config({H, Port, User, Pass, AdminPort}) ->
     AdminConn = couch_util:get_new_connection(H, wh_util:to_integer(AdminPort), User, Pass),
 
     ?LOG_SYS("returning state record"),
+    couch_config:ready(),
     #state{connection=Conn
            ,admin_connection=AdminConn
            ,host={H, wh_util:to_integer(Port), wh_util:to_integer(AdminPort)}