-module(freeswitch_monitor). -author('fig@marketelsystems.com'). -behaviour(gen_server). -define(PID_ANNOUNCE, "run_freeswitch.pid="). -define(PREFIX, "freeswitch< "). -define(PROG, "./run_freeswitch"). -define(TIMEOUT, 5000). -record(state, {port, prefix=?PREFIX, pid, fspid, sync=[], trigger, node}). -export([start_link/0, terminate/2, code_change/3, init/1, handle_call/3, handle_cast/2, handle_info/2]). -export([send/1]). %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% gen_server methods start_link() -> process_flag(trap_exit, true), SName = lists:takewhile(fun ($.) -> false; (_C) -> true end, net_adm:localhost()), FsNode = list_to_atom("freeswitch@" ++ SName), syslog:debug("FsNode = ~w calling freeswitch:start_link()~n", [FsNode]), gen_server:start_link({global, freeswitch}, ?MODULE, FsNode, []). init(FsNode) -> %% Have our terminate/2 function run when our supervisor shuts us down. syslog:debug("init(~w) called~n",[FsNode]), State = #state{port=starconf:start_ext(?MODULE, ?PROG), node=FsNode}, syslog:debug("State = ~w~n", [State]), {ok, try_getpid(1000, State)}. terminate(Reason, State) -> case Reason of {process_exited, _} -> ok; _ -> Port = State#state.port, catch port_command(Port, "shutdown\n"), receive {Port, {exit_status, _}} -> ok after 5000 -> starconf:stop_ext(State#state.port, State#state.pid) end end, ok. code_change(OldVsn, State, _Extra) -> starconf:stop_ext(State#state.port, State#state.pid), {ok, #state{port=starconf:start_ext(?MODULE, OldVsn, ?PROG)}}. handle_call(sync, From, #state{fspid=undefined}=State) -> syslog:debug("freeswitch_monitor reached to handle sync call 1.~n"), {foo, State#state.node} ! getpid, {noreply, State#state{sync=[From | State#state.sync]}}; handle_call(sync, _From, State) -> %% We are synced, so return our Pid to link to. syslog:debug("freeswitch_monitor reached to handle sync call 2 and is successful.~n"), {reply, {ok, self()}, State}; handle_call(_, _From, #state{fspid=undefined}=State) -> syslog:error("Freeswitch_monitor is not ready and throw error msg.~n"), {reply, {error, not_ready}, State}; handle_call({pbx_api, Pid, Tag, Cmd, Args}, _From, State) -> %% We background the call so we can continue, but reply to %% the original caller from our child process. CPid = spawn(fun () -> link(State#state.fspid), {api, State#state.node} ! {api, Cmd, Args}, receive {error, Reason} -> Pid ! {Tag, {error, Reason}}; {ok, Reply} -> Pid ! {Tag, {ok, Reply}} end end), {reply, {ok, CPid}, State}; handle_call({pbx_bgapi, Pid, Tag, Cmd, Args}, _From, State) -> %% We background the call so we can continue, but reply to %% the original caller from our child process. CPid = spawn(fun () -> link(State#state.fspid), {bgapi, State#state.node} ! {bgapi, Cmd, Args}, receive {error, Reason} -> Pid ! {Tag, {error, Reason}}; {ok, Reply} -> Pid ! {Tag, {ok, Reply}} end end), {reply, {ok, CPid}, State}; handle_call({pbx_sendmsg, Pid, Tag, UUID, Headers}, _From, State) -> %% We background the call so we can continue, but reply to %% the original caller from our child process. CPid = spawn(fun () -> link(State#state.fspid), syslog:debug( "freeswitch_monitor:handle_call() pbx_sendmsg invoking sendmsg, UUID=~p, Headers=~p~n", [ UUID, Headers ] ), {sendmsg, State#state.node} ! {sendmsg, UUID, Headers}, receive ok -> Pid ! { Tag, { ok } }; {error, Reason} -> syslog:debug( "freeswitch_monitor:handle_call() pbx_sendmsg received {error, ~p}~n", [Reason] ), Pid ! { Tag, {error, Reason} } after ?TIMEOUT -> timeout end end), {reply, {ok, CPid}, State}; handle_call({pbx_bind, Section, Pid}, _From, State) -> Status = freeswitch_bind:start(State#state.node, Section, Pid), {reply, Status, State}; handle_call(Request, _From, State) -> error_logger:error_msg("~p:~p received unrecognized request ~p~n", [?MODULE, self(), Request]), {reply, {error, unrecognized_request}, State}. handle_cast({pbx_api, _Pid, _Tag, _Cmd, _Args}=Request, State) -> handle_call(Request, undefined, State), {noreply, State}; handle_cast({send, Cmd}, State) -> catch port_command(State#state.port, [Cmd, "\n"]), {noreply, State}; handle_cast(Message, State) -> error_logger:error_msg("~p:~p received unrecognized cast ~p~n", [?MODULE, self(), Message]), {noreply, State}. handle_info({Port, {exit_status, Status}}, #state{port=Port}=State) -> case Status of 0 -> {stop, normal, State}; _N -> {stop, {process_exited, Status}, State} end; handle_info({try_getpid, Timeout}, State) -> {noreply, try_getpid(Timeout, State)}; handle_info({ok, Pid}, #state{sync=Sync}=State) when is_pid(Pid) -> syslog:info("~p:~p got ~p pid ~p~n", [?MODULE, self(), State#state.node, Pid]), case State#state.trigger of undefined -> ok; Trigger -> timer:cancel(Trigger), ok end, lists:foreach(fun (From) -> gen_server:reply(From, {ok, self()}) end, Sync), {noreply, State#state{trigger=undefined, fspid=Pid, sync=[]}}; handle_info({Port, {data, {noeol, Output}}}, #state{port=Port}=State) -> syslog:info("~s~s", [State#state.prefix, Output]), {noreply, State#state{prefix=""}}; handle_info({Port, {data, {eol, Output}}}, #state{port=Port}=State) -> syslog:info("~s~s~n", [State#state.prefix, Output]), Len = string:len(?PID_ANNOUNCE), Pid = case string:substr(Output, 1, Len) of ?PID_ANNOUNCE -> string:substr(Output, Len + 1); _ -> State#state.pid end, {noreply, State#state{prefix=?PREFIX, pid=Pid}}; handle_info(Message, State) -> syslog:info("~p:handle_info() Got info ~p~n", [?MODULE, Message]), {noreply, State}. try_getpid(Timeout, #state{fspid=undefined}=State) -> syslog:debug("~p:try_getpid() trying to get ~p pid~n", [?MODULE, State#state.node]), {foo, State#state.node} ! getpid, {ok, Trigger} = timer:send_after(Timeout, {try_getpid, Timeout}), State#state{trigger=Trigger}; try_getpid(_Timeout, State) -> State. %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% API send(Cmd) -> gen_server:cast({global, freeswitch}, {send, Cmd}). -module(pbx). -behaviour(gen_server). %% External API -export([start_link/0, start_link/1, status/0, api/1, api/2, bgapi/1, bgapi/2, sendmsg/2, sync/1, bind/1]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -export([asteriskDial/2, asteriskHangup/1, callCustomer/3]). -export([fsnodes/0, fsvars/0, fsfuns/0, pbxSubscribe/2, pbxUnsubscribe/2]). -record(st, {pid}). -include_lib("figure/include/gut.hrl"). start_link() -> start_link({global, freeswitch}). start_link(FsPid) -> gen_server:start_link({local, ?MODULE}, ?MODULE, [FsPid], []). callCustomer( OperatorId, PhoneNumber, CustomerName ) -> syslog:debug( "pbx:callCustomer() OperatorId=~p, PhoneNumber=\"~s\", CustomerName=\"~s\"~n", [ OperatorId, PhoneNumber, CustomerName ] ), { ok, DialerPrefix } = mnesia_sofia:getSofiaConfigValue( dialerPrefix ), { ok, DialerSuffix } = mnesia_sofia:getSofiaConfigValue( dialerSuffix ), { ok, DialerTimeout } = mnesia_sofia:getSofiaConfigValue( dialerTimeout ), { ok, DialerErlangNode } = mnesia_sofia:getSofiaConfigValue( dialerErlangNode ), OriginateCmd = "{call_timeout=" ++ DialerTimeout ++ ",originate_timeout=" ++ DialerTimeout ++ "}" ++ DialerPrefix ++ PhoneNumber ++ DialerSuffix ++ " '&erlang( pbx_called_cust pursuit@" ++ DialerErlangNode ++ " )'", syslog:debug( "pbx:callCustomer() OriginateCmd is \"~s\"~n", [OriginateCmd] ), case api( originate, OriginateCmd ) of { ok, ResultStrA } -> LenA = string:len( ResultStrA ), UuidA = string:substr( ResultStrA, 5, LenA - 5 ), syslog:debug( "pbx:callCustomer() UUID A is \"~s\"~n", [ UuidA ] ), { ok, PbxCustomerId } = pbx_mnesia:queueCalledCustomer( UuidA, PhoneNumber, CustomerName, OperatorId ), { ok, PbxCustomerId }; { error, "-ERR NO_ANSWER\n" } -> syslog:warn( "pbx:callCustomer() No answer at ~s~n", [PhoneNumber] ), { error, no_answer }; { error, "-ERR NORMAL_TEMPORARY_FAILURE\n" } -> syslog:error( "pbx:callCustomer() Error dialing ~s -- Normal temporary failure~n", [PhoneNumber] ), { error, normal_temporary_failure }; ErrorA -> UuidA = "UUIDA", syslog:error( "pbx:callCustomer() Error dialing ~s -- ~p~n", [PhoneNumber, ErrorA] ), { error, untrapped_error } end. %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% Server callbacks. init([FsPid]) -> {ok, Pid} = sync(FsPid), link(Pid), {ok, #st{pid=FsPid}}. handle_call({api, Pid, Tag, Cmd, Args}, _From, State) -> Ret = gen_server:call(State#st.pid, {pbx_api, Pid, Tag, Cmd, Args}), {reply, Ret, State}; handle_call({bgapi, Pid, Tag, Cmd, Args}, _From, State) -> Ret = gen_server:call(State#st.pid, {pbx_bgapi, Pid, Tag, Cmd, Args}), {reply, Ret, State}; handle_call({sendmsg, Pid, Tag, UUID, Headers}, _From, State) -> Ret = gen_server:call(State#st.pid, {pbx_sendmsg, Pid, Tag, UUID, Headers}), {reply, Ret, State}; handle_call({phone, Pid, {PbxSessionID, PhoneNumber, Name}}, _From, State) -> syslog:debug("handle_call phone reached.~n"), {reply, {error, not_implemented_yet}, State}; handle_call({barge, Pid, {PbxSessionID, TargetPbxSessionID}}, _From, State) -> syslog:debug("handle_call barge reached.~n"), {reply, {error, not_implemented_yet}, State}; handle_call({whisper, Pid, {PbxSessionID, TargetPbxSessionID}}, _From, State) -> syslog:debug("handle_call whisper reached.~n"), {reply, {error, not_implemented_yet}, State}; handle_call({disconnect, Pid, {PbxSessionID}}, _From, State) -> syslog:debug("handle_call disconnect reached.~n"), {reply, {error, not_implemented_yet}, State}; handle_call({hang_up, Pid, {PbxSessionId}}, _From, State) -> syslog:debug("handle_call hang_up reached.~n"), {reply, {error, not_implemented_yet}, State}; handle_call({record, Pid, {PbxSessionId}}, _From, State) -> syslog:debug("handle_call record reached.~n"), {reply, {error, not_implemented_yet}, State}; handle_call({stop_recording, Pid, {PbxSessionId}}, _From, State) -> syslog:debug("handle_call stop_recording reached.~n"), {reply, {error, not_implemented_yet}, State}; handle_call({bind, Pid, Section}, _From, State) -> syslog:debug("handle_call bind reached.~n"), Status = gen_server:call(State#st.pid, {pbx_bind, Section, Pid}), {reply, Status, State}; handle_call(Request, _From, State) -> syslog:debug("handle_call unrecognized reached.~n"), error_logger:error_msg("~p:~p received unrecognized request ~p~n", [?MODULE, self(), Request]), {reply, {error, unrecognized_request}, State}. handle_cast(Msg, State) -> syslog:debug("handle_cast reached.~n"), error_logger:error_msg("~p:~p received unrecognized cast ~p~n", [?MODULE, self(), Msg]), {noreply, State}. handle_info(Info, State) -> syslog:debug("handle_info reached.~n"), error_logger:error_msg("~p:~p received unrecognized info ~p~n", [?MODULE, self(), Info]), {noreply, State}. terminate(_Reason, _State) -> syslog:debug("terminate reached.~n"), ok. code_change(_OldVsn, State, _Extra) -> syslog:debug("code_change reached.~n"), {ok, State}. %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% Public stateful API. status() -> api(status). api(Str) when is_list(Str) -> case lists:splitwith(fun ($ ) -> false; (_) -> true end, Str) of {Cmd, [$ | Args]} -> api(list_to_atom(Cmd), Args); {Cmd, []} -> api(list_to_atom(Cmd)) end; api(Cmd) when is_atom(Cmd) -> api(Cmd, []). api(Cmd, Args) when is_list(Cmd) -> api(list_to_atom(Cmd), Args); api(Cmd, Args) when is_atom(Cmd) -> Ref = make_ref(), case gen_server:call(?MODULE, {api, self(), Ref, Cmd, Args}) of {ok, Pid} -> syslog:debug( "pbx:api() Sent to Pid ~p~n", [Pid] ), MonitorRef = erlang:monitor(process, Pid), Ret = receive {Ref, Result} -> Result; {'DOWN', MonitorRef, _, _, _} -> {error, retry} end, erlang:demonitor(MonitorRef), receive {'DOWN', MonitorRef, _, _, _} -> ok after 0 -> ok end, Ret; Other -> Other end. bgapi(Str) when is_list(Str) -> case lists:splitwith(fun ($ ) -> false; (_) -> true end, Str) of {Cmd, [$ | Args]} -> bgapi(list_to_atom(Cmd), Args); {Cmd, []} -> bgapi(list_to_atom(Cmd)) end; bgapi(Cmd) when is_atom(Cmd) -> bgapi(Cmd, []). bgapi(Cmd, Args) when is_list(Cmd) -> bgapi(list_to_atom(Cmd), Args); bgapi(Cmd, Args) when is_atom(Cmd) -> Ref = make_ref(), case gen_server:call(?MODULE, {bgapi, self(), Ref, Cmd, Args}) of {ok, Pid} -> syslog:debug( "pbx:bgapi() Sent to Pid ~p~n", [Pid] ), MonitorRef = erlang:monitor(process, Pid), Ret = receive {Ref, Result} -> Result; {'DOWN', MonitorRef, _, _, _} -> {error, retry} end, erlang:demonitor(MonitorRef), receive {'DOWN', MonitorRef, _, _, _} -> ok after 0 -> ok end, Ret; Other -> Other end. sendmsg(UUID, Headers) when is_list(UUID), is_list( Headers ) -> Ref = make_ref(), case gen_server:call(?MODULE, {sendmsg, self(), Ref, UUID, Headers}) of {ok, Pid} -> syslog:debug( "pbx:sendmsg() Sent to Pid ~p~n", [Pid] ), MonitorRef = erlang:monitor(process, Pid), Ret = receive {Ref, Result} -> Result; {'DOWN', MonitorRef, _, _, _} -> {error, retry} end, erlang:demonitor(MonitorRef), receive {'DOWN', MonitorRef, _, _, _} -> ok after 0 -> ok end, Ret; Other -> syslog:debug( "pbx:sendmsg() Returning ~p~n", [Other] ), Other end. %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% Other functions sync(FsPid) -> catch(net_adm:world()), try gen_server:call(FsPid, sync, infinity) catch exit:{noproc,_} -> receive after 2000 -> ok end, syslog:debug("~p:sync() trying to sync with ~p~n", [?MODULE, FsPid]), sync(FsPid), erlang:get_stacktrace() end. bind(Section) -> gen_server:call(?MODULE, {bind, self(), Section}). fsnodes() -> nd(str("MarkeTel's PBX interface"), [{program, nd(str("PBX program"), [{function, nd(str("PBX functions"), fsfuns())}])} ]). fsfuns() -> [func(pbxSubscribe), func(pbxUnsubscribe)]. fsvars() -> []. %% Initialize the pbx_dict. %% FIXME: Maybe change to accept a Fun, which gets the dict and %% returns the new dict so that these are always paired. pbxStart() -> case get(pbx_dict) of undefined -> Dict = dict:new(), listen(), put(pbx_dict, Dict), Dict; Dict -> Dict end. pbxStop(Dict) -> put(pbx_dict, Dict). %% Subscribe to inbound calls on a given extension. pbxSubscribe(_D, [#str{s=Ext}]) -> Dict = pbxStart(), Dict2 = dict:update_counter(handle, 1, Dict), Handle = dict:fetch(handle, Dict2), Dict3 = dict:append({ext, Ext}, Handle, Dict2), Dict4 = dict:store(Handle, Ext, Dict3), pbxStop(Dict4), Handle. asteriskDial(Packet, Out) -> {value, {_Name0, ExtDashUnique}} = lists:keysearch(destination, 1, Packet), Ext = case string:rchr(ExtDashUnique, $-) of 0 -> ExtDashUnique; Pos -> string:substr(ExtDashUnique, 1, Pos - 1) end, %% Get the callerid for the inbound call. {value, {_, PhoneNumber}} = lists:keysearch(callerid, 1, Packet), Dict = pbxStart(), case dict:find({ext, Ext}, Dict) of {ok, Handles} -> {value, {_, Uid2}} = lists:keysearch(destuniqueid, 1, Packet), case dict:find({already, Uid2}, Dict) of {ok, _} -> %% Already done, so skip. ok; error -> lists:foreach(fun (H) -> Out(['pbx.Connect', H, str(Ext), [callerId, str(PhoneNumber)]]) end, Handles), pbxStop(dict:store({already, Uid2}, true, Dict)) end; error -> ok end, ok. %% Clean up the already list when channel hangups occur. asteriskHangup(Packet) -> Dict = pbxStart(), {value, {_Name, Channel}} = lists:keysearch(uniqueid, 1, Packet), pbxStop(dict:erase({already, Channel}, Dict)). %% Stop listening for a given extension. pbxUnsubscribe(_D, [Handle]) -> Dict = pbxStart(), Ext = dict:fetch(Handle, Dict), Dict2 = case dict:fetch({ext, Ext}, Dict) of Handles when is_list(Handles) -> Handles2 = lists:filter(fun (H) -> H /= Handle end, Handles), case Handles2 of [] -> dict:erase({ext, Ext}, Dict); _Other -> dict:store({ext, Ext}, Handles2, Dict) end end, Dict3 = dict:erase(Handle, Dict2), pbxStop(Dict3), ok. listen() -> %%If we need a remote Pursuit, here it is. case file:consult("/var/lib/pursuit.db/opts") of {ok, [Terms]} -> case lists:keysearch([asterisk_manager, node], 1, Terms) of {value, {_, Node}} -> gen_server:call({asterisk_manager, Node}, {listen, self()}); false -> asterisk_manager:listen() end; {error, enoent} -> asterisk_manager:listen() end.