Skip to content
This repository

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse code

WHISTLE-1441: stop super, listener, and fsm as appropriate

WHISTLE-1441: add the queue id to the current call data returned by the agent FSM

WHISTLE-1441: bind the shared queue queue as non-exclusive

WHISTLE-1441: formatting

WHISTLE-1441: fix startup config params

WHISTLE-1441: generic queue

WHISTLE-1441: check endpoints against registrations

WHISTLE-1441: dialyzer fix

WHISTLE-1441: load endpoints in the FSM

WHISTLE-1441: don't load endpoints in the agent listener; receive them from the agent FSM

WHISTLE-1666: formatting

WHISTLE-1441: stop super, listener, and fsm as appropriate
  • Loading branch information...
commit 1f17cee4e72a78f19d256ac9223552d4b206883b 1 parent 644b6e1
James Aimonetti jamesaimonetti authored
5 lib/whistle_amqp-1.0.0/src/wh_amqp_connection.erl
@@ -295,7 +295,7 @@ init([Broker]) ->
295 295 %% @end
296 296 %%--------------------------------------------------------------------
297 297 handle_call(teardown_channels, _, #state{broker_name=Name}=State) ->
298   - clear_channels(Name),
  298 + _ = clear_channels(Name),
299 299 {reply, ok, State};
300 300
301 301 handle_call(use_federation, _, #state{broker=Broker}=State) ->
@@ -552,7 +552,8 @@ exchange_declare(#'exchange.declare'{type=Type}=ED, true) ->
552 552 ED1.
553 553
554 554 clear_channels(Name) ->
555   - [clear_channel(C) || #wh_amqp_channel{}=C <- ets:tab2list(Name)].
  555 + [clear_channel(C) || #wh_amqp_channel{}=C <- ets:tab2list(Name)],
  556 + ok.
556 557 clear_channel(#wh_amqp_channel{channel=ChPid
557 558 ,channel_ref=ChRef
558 559 ,consumer_ref=ConRef
76 whistle_apps/apps/acdc/src/acdc_agent.erl
@@ -14,7 +14,7 @@
14 14 -export([start_link/2
15 15 ,member_connect_resp/2
16 16 ,member_connect_retry/2
17   - ,bridge_to_member/2
  17 + ,bridge_to_member/3
18 18 ,monitor_call/2
19 19 ,member_connect_accepted/1
20 20 ,channel_hungup/2
@@ -27,6 +27,7 @@
27 27 ,add_acdc_queue/2
28 28 ,rm_acdc_queue/2
29 29 ,get_recording_doc_id/1
  30 + ,stop/1
30 31 ]).
31 32
32 33 %% gen_server callbacks
@@ -50,7 +51,6 @@
50 51 ,acct_id :: ne_binary()
51 52 ,fsm_pid :: pid()
52 53 ,agent_queues :: [ne_binary(),...] | []
53   - ,endpoints :: wh_json:json_objects()
54 54 ,last_connect :: wh_now() % last connection
55 55 ,last_attempt :: wh_now() % last attempt to connect
56 56 ,my_id :: ne_binary()
@@ -141,6 +141,9 @@ start_link(Supervisor, AgentJObj) ->
141 141 )
142 142 end.
143 143
  144 +stop(Srv) ->
  145 + gen_listener:cast(Srv, {stop_agent}).
  146 +
144 147 -spec member_connect_resp/2 :: (pid(), wh_json:json_object()) -> 'ok'.
145 148 member_connect_resp(Srv, ReqJObj) ->
146 149 gen_listener:cast(Srv, {member_connect_resp, ReqJObj}).
@@ -151,8 +154,8 @@ member_connect_retry(Srv, WinJObj) ->
151 154 member_connect_accepted(Srv) ->
152 155 gen_listener:cast(Srv, member_connect_accepted).
153 156
154   -bridge_to_member(Srv, WinJObj) ->
155   - gen_listener:cast(Srv, {bridge_to_member, WinJObj}).
  157 +bridge_to_member(Srv, WinJObj, EPs) ->
  158 + gen_listener:cast(Srv, {bridge_to_member, WinJObj, EPs}).
156 159
157 160 monitor_call(Srv, MonitorJObj) ->
158 161 gen_listener:cast(Srv, {monitor_call, MonitorJObj}).
@@ -207,11 +210,11 @@ init([Supervisor, AgentJObj, Queues]) ->
207 210 AgentId = wh_json:get_value(<<"_id">>, AgentJObj),
208 211 put(callid, AgentId),
209 212
210   - gen_listener:cast(self(), {load_endpoints, Supervisor}),
211   -
212 213 Self = self(),
213 214 AcctId = wh_json:get_value(<<"pvt_account_id">>, AgentJObj),
214 215
  216 + gen_listener:cast(self(), {start_fsm, Supervisor}),
  217 +
215 218 _ = spawn(fun() ->
216 219 put(amqp_publish_as, Self),
217 220
@@ -266,11 +269,14 @@ handle_call(_Request, _From, State) ->
266 269 %% {stop, Reason, State}
267 270 %% @end
268 271 %%--------------------------------------------------------------------
  272 +handle_cast({stop_agent}, #state{supervisor=Supervisor}=State) ->
  273 + acdc_agent_sup:stop(Supervisor),
  274 + {noreply, State};
269 275 handle_cast({start_fsm, Supervisor}, #state{acct_id=AcctId
270 276 ,agent_id=AgentId
271 277 }=State) ->
272 278 {ok, FSMPid} = acdc_agent_sup:start_fsm(Supervisor, AcctId, AgentId),
273   - link(FSMPid),
  279 +
274 280 lager:debug("started FSM at ~p", [FSMPid]),
275 281
276 282 gen_listener:cast(self(), bind_to_member_reqs),
@@ -347,35 +353,6 @@ handle_cast(member_connect_accepted, #state{msg_queue_id=AmqpQueue
347 353 send_member_connect_accepted(AmqpQueue, call_id(Call), AcctId, AgentId, MyId),
348 354 {noreply, State};
349 355
350   -handle_cast({load_endpoints, Supervisor}, #state{acct_db=AcctDb
351   - ,agent_id=AgentId
352   - ,acct_id=AcctId
353   - ,record_calls=ShouldRecord
354   - }=State) ->
355   - lager:debug("loading agent endpoints"),
356   - Call = whapps_call:set_account_id(AcctId
357   - ,whapps_call:set_account_db(AcctDb
358   - ,whapps_call:new()
359   - )
360   - ),
361   - case catch acdc_util:get_endpoints(Call, AgentId) of
362   - [] ->
363   - lager:debug("no endpoints"),
364   - _ = acdc_agent_sup:stop(Supervisor),
365   - {noreply, State};
366   - [_|_]=EPs ->
367   - lager:debug("endpoints: ~p", [EPs]),
368   - gen_listener:cast(self(), {start_fsm, Supervisor}),
369   -
370   - {noreply, State#state{endpoints=EPs
371   - ,record_calls=record_endpoints(EPs, ShouldRecord)
372   - }};
373   - {'EXIT', _E} ->
374   - lager:debug("failed to load endpoints: ~p", [_E]),
375   - _ = acdc_agent_sup:stop(Supervisor),
376   - {noreply, State}
377   - end;
378   -
379 356 handle_cast({member_connect_resp, _}=Msg, #state{my_q = Q}=State) when
380 357 Q =:= undefined orelse Q =:= <<>> ->
381 358 fetch_my_queue(),
@@ -414,20 +391,24 @@ handle_cast({member_connect_retry, WinJObj}, #state{my_id=MyId}=State) ->
414 391 send_member_connect_retry(WinJObj, MyId),
415 392 {noreply, State};
416 393
417   -handle_cast({bridge_to_member, WinJObj}, #state{endpoints=EPs
418   - ,fsm_pid=FSM
419   - }=State) ->
  394 +handle_cast({bridge_to_member, WinJObj, EPs}, #state{fsm_pid=FSM
  395 + ,record_calls=RecordCall
  396 + }=State) ->
420 397 lager:debug("bridging to agent endpoints: ~p", [EPs]),
421 398
422 399 Call = whapps_call:from_json(wh_json:get_value(<<"Call">>, WinJObj)),
423 400 RingTimeout = wh_json:get_value(<<"Ring-Timeout">>, WinJObj),
424 401 lager:debug("ring agent for ~p", [RingTimeout]),
425 402
  403 + ShouldRecord = should_record_endpoints(EPs, RecordCall),
  404 +
426 405 acdc_util:bind_to_call_events(Call),
427 406 _P = spawn(fun() -> maybe_connect_to_agent(FSM, EPs, Call, RingTimeout) end),
428 407
429 408 lager:debug("waiting on successful bridge now: connecting in ~p", [_P]),
430   - {noreply, State#state{call=Call}};
  409 + {noreply, State#state{call=Call
  410 + ,record_calls=ShouldRecord
  411 + }};
431 412
432 413 handle_cast({monitor_call, MonitorJObj}, State) ->
433 414 Call = whapps_call:set_call_id(wh_json:get_value(<<"Call-ID">>, MonitorJObj), whapps_call:new()),
@@ -495,12 +476,6 @@ handle_cast(_Msg, State) ->
495 476 %% {stop, Reason, State}
496 477 %% @end
497 478 %%--------------------------------------------------------------------
498   -handle_info({'EXIT', P, _R}, #state{fsm_pid=P, supervisor=Supervisor}=State) ->
499   - lager:debug("our FSM(~p) died: ~p", [P, _R]),
500   -
501   - gen_listener:cast(self(), {start_fsm, Supervisor}),
502   -
503   - {noreply, State#state{fsm_pid=undefined}};
504 479 handle_info(_Info, State) ->
505 480 lager:debug("unhandled message: ~p", [_Info]),
506 481 {noreply, State}.
@@ -528,7 +503,8 @@ handle_event(_JObj, #state{fsm_pid=FSM}) ->
528 503 %% @spec terminate(Reason, State) -> void()
529 504 %% @end
530 505 %%--------------------------------------------------------------------
531   -terminate(_Reason, _State) ->
  506 +terminate(_Reason, #state{supervisor=Supervisor}) ->
  507 + acdc_agent_sup:stop(Supervisor),
532 508 lager:debug("agent process going down: ~p", [_Reason]).
533 509
534 510 %%--------------------------------------------------------------------
@@ -706,9 +682,9 @@ fetch_my_queue() ->
706 682 _ = spawn(fun() -> gen_listener:cast(Self, {queue_name, gen_listener:queue_name(Self)}) end),
707 683 ok.
708 684
709   --spec record_endpoints/2 :: (wh_json:json_objects(), boolean()) -> boolean().
710   -record_endpoints(_EPs, true) -> true;
711   -record_endpoints(EPs, false) ->
  685 +-spec should_record_endpoints/2 :: (wh_json:json_objects(), boolean()) -> boolean().
  686 +should_record_endpoints(_EPs, true) -> true;
  687 +should_record_endpoints(EPs, false) ->
712 688 lists:any(fun(EP) ->
713 689 wh_json:is_true(<<"record_calls">>, EP, false)
714 690 end, EPs).
39 whistle_apps/apps/acdc/src/acdc_agent_fsm.erl
@@ -79,6 +79,7 @@
79 79 ,caller_exit_key = <<"#">> :: ne_binary()
80 80 ,agent_call_id :: ne_binary()
81 81 ,next_status :: ne_binary()
  82 + ,endpoints = [] :: wh_json:json_objects()
82 83 }).
83 84
84 85 %%%===================================================================
@@ -246,7 +247,9 @@ init([AcctId, AgentId, AgentProc]) ->
246 247 gen_fsm:send_event(self(), send_sync_event),
247 248
248 249 acdc_stats:agent_active(AcctId, AgentId),
249   -
  250 +
  251 + gen_fsm:send_all_state_event(self(), load_endpoints),
  252 +
250 253 {ok, sync, #state{acct_id=AcctId
251 254 ,acct_db=wh_util:format_account_id(AcctId, encoded)
252 255 ,agent_id=AgentId
@@ -344,13 +347,15 @@ ready({sync_req, JObj}, #state{agent_proc=Srv}=State) ->
344 347 acdc_agent:send_sync_resp(Srv, ready, JObj),
345 348 {next_state, ready, State};
346 349
347   -ready({member_connect_win, JObj}, #state{agent_proc=Srv}=State) ->
  350 +ready({member_connect_win, JObj}, #state{agent_proc=Srv
  351 + ,endpoints=[_|_]=EPs
  352 + }=State) ->
348 353 Call = whapps_call:from_json(wh_json:get_value(<<"Call">>, JObj)),
349 354 CallId = whapps_call:call_id(Call),
350 355
351 356 lager:debug("we won us a member: ~s", [CallId]),
352 357
353   - acdc_agent:bridge_to_member(Srv, JObj),
  358 + acdc_agent:bridge_to_member(Srv, JObj, EPs),
354 359
355 360 WrapupTimer = wh_json:get_integer_value(<<"Wrapup-Timeout">>, JObj, 0),
356 361 CallerExitKey = wh_json:get_value(<<"Caller-Exit-Key">>, JObj, <<"#">>),
@@ -734,6 +739,32 @@ paused(current_call, _, State) ->
734 739 handle_event({refresh, AgentJObj}, StateName, State) ->
735 740 lager:debug("refresh agent config: ~p", [AgentJObj]),
736 741 {next_state, StateName, State};
  742 +handle_event(load_endpoints, StateName, #state{acct_db=AcctDb
  743 + ,agent_id=AgentId
  744 + ,acct_id=AcctId
  745 + ,agent_proc=Srv
  746 + }=State) ->
  747 + Setters = [fun(C) -> whapps_call:set_account_id(AcctId, C) end
  748 + ,fun(C) -> whapps_call:set_account_db(AcctDb, C) end
  749 + ],
  750 +
  751 + Call = lists:foldl(fun(F, C) -> F(C) end
  752 + ,whapps_call:new(), Setters
  753 + ),
  754 + case catch acdc_util:get_endpoints(Call, AgentId) of
  755 + [] ->
  756 + lager:debug("no endpoints, going down"),
  757 + acdc_agent:stop(Srv),
  758 + {stop, normal, State};
  759 + [_|_]=EPs ->
  760 + lager:debug("endpoints: ~p", [EPs]),
  761 +
  762 + {next_state, StateName, State#state{endpoints=EPs}};
  763 + {'EXIT', _E} ->
  764 + lager:debug("failed to load endpoints: ~p", [_E]),
  765 + acdc_agent:stop(Srv),
  766 + {stop, normal, State}
  767 + end;
737 768 handle_event(_Event, StateName, State) ->
738 769 lager:debug("unhandled event in state ~s: ~p", [StateName, _Event]),
739 770 {next_state, StateName, State}.
@@ -789,8 +820,10 @@ handle_info(_Info, StateName, State) ->
789 820 %%--------------------------------------------------------------------
790 821 terminate(_Reason, _StateName, #state{acct_id=AcctId
791 822 ,agent_id=AgentId
  823 + ,agent_proc=Srv
792 824 }) ->
793 825 acdc_stats:agent_inactive(AcctId, AgentId),
  826 + acdc_agent:stop(Srv),
794 827 lager:debug("acdc agent fsm terminating while in ~s: ~p", [_StateName, _Reason]).
795 828
796 829 %%--------------------------------------------------------------------
2  whistle_apps/apps/acdc/src/acdc_queue_manager.erl
@@ -77,7 +77,7 @@ handle_member_call(JObj, _Props) ->
77 77 acdc_stats:call_waiting(AcctId, QueueId, whapps_call:call_id(Call)),
78 78
79 79 case acdc_queues_sup:find_queue_supervisor(AcctId, QueueId) of
80   - P when is_pid(P) ->
  80 + P when is_pid(P) ->
81 81 acdc_queue:put_member_on_hold(acdc_queue_sup:queue(P), Call);
82 82 undefined ->
83 83 whapps_call_command:answer(Call),
1  whistle_apps/apps/acdc/src/acdc_queue_shared.erl
@@ -34,6 +34,7 @@
34 34 ,{exclusive, false}
35 35 ]}
36 36 ,{basic_qos, 1}
  37 + ,{queue_options, [{exclusive, false}]}
37 38 ]).
38 39
39 40 -define(SHARED_QUEUE_BINDINGS(AcctId, QueueId), [{self, []}]).
4 whistle_apps/apps/acdc/src/acdc_stats.erl
@@ -280,15 +280,11 @@ agent_ready(AcctId, AgentId) ->
280 280
281 281 -define(BINDINGS, []).
282 282 -define(RESPONDERS, []).
283   --define(QUEUE_NAME, <<"acdc.stats">>).
284   --define(CONSUME_OPTIONS, []).
285 283 start_link() ->
286 284 gen_listener:start_link({local, ?MODULE}
287 285 ,?MODULE
288 286 ,[{bindings, ?BINDINGS}
289 287 ,{responders, ?RESPONDERS}
290   - ,{queue_name, ?QUEUE_NAME}
291   - ,{consume_options, ?CONSUME_OPTIONS}
292 288 ],
293 289 []).
294 290
20 whistle_apps/apps/acdc/src/acdc_util.erl
@@ -40,11 +40,16 @@ get_endpoints(Call, ?NE_BINARY = AgentId) ->
40 40 get_endpoints(_Call, {ok, []}) -> [];
41 41 get_endpoints(_Call, {error, _E}) -> [];
42 42 get_endpoints(Call, {ok, Devices}) ->
  43 + {ok, AcctDoc} = couch_mgr:open_cache_doc(whapps_call:account_db(Call), whapps_call:account_id(Call)),
  44 + AcctRealm = wh_json:get_value(<<"realm">>, AcctDoc),
  45 +
43 46 EPDocs = [EPDoc
44 47 || Device <- Devices,
45 48 (EPDoc = get_endpoint(Call, wh_json:get_value(<<"id">>, Device))) =/= undefined,
46   - wh_json:is_true(<<"enabled">>, EPDoc, false)
  49 + wh_json:is_true(<<"enabled">>, EPDoc, false),
  50 + is_endpoint_registered(EPDoc, AcctRealm)
47 51 ],
  52 +
48 53 lists:foldl(fun(EPDoc, Acc) ->
49 54 case cf_endpoint:build(EPDoc, Call) of
50 55 {ok, EP} -> EP ++ Acc;
@@ -52,6 +57,19 @@ get_endpoints(Call, {ok, Devices}) ->
52 57 end
53 58 end, [], EPDocs).
54 59
  60 +is_endpoint_registered(EPDoc, AcctRealm) ->
  61 + Query = [{<<"Realm">>, AcctRealm}
  62 + ,{<<"Username">>, wh_json:get_value([<<"sip">>, <<"username">>], EPDoc)}
  63 + ,{<<"Fields">>, [<<"Contact">>]}
  64 + ],
  65 + case whapps_util:amqp_pool_request(Query
  66 + ,fun wapi_registration:publish_query_req/1
  67 + ,fun wapi_registration:query_resp_v/1
  68 + ) of
  69 + {ok, _Resp} -> true;
  70 + {error, _E} -> false
  71 + end.
  72 +
55 73 -spec get_endpoint/2 :: (whapps_call:call(), ne_binary()) -> wh_json:json_object() | 'undefined'.
56 74 get_endpoint(Call, ?NE_BINARY = EndpointId) ->
57 75 case couch_mgr:open_doc(whapps_call:account_db(Call), EndpointId) of
8 whistle_apps/apps/crossbar/src/modules/cb_queues.erl
@@ -350,10 +350,10 @@ fetch_all_queue_stats(Context, history) ->
350 350 From = calendar:datetime_to_gregorian_seconds({Today, {0,0,0}}),
351 351
352 352 crossbar_doc:load_view(<<"acdc_stats/stats_per_queue_by_time">>
353   - ,[{startkey, [wh_util:current_tstamp(), <<"\ufff0">>]}
354   - ,{endkey, [From, <<>>]}
355   - ,descending
356   - ]
  353 + ,[{startkey, [wh_util:current_tstamp(), <<"\ufff0">>]}
  354 + ,{endkey, [From, <<>>]}
  355 + ,descending
  356 + ]
357 357 ,Context
358 358 ,fun normalize_queue_results/2
359 359 );

0 comments on commit 1f17cee

Please sign in to comment.
Something went wrong with that request. Please try again.