diff --git a/apps/ann_b2b/AnnounceB2B.cpp b/apps/ann_b2b/AnnounceB2B.cpp index 3904387d..d9a820a9 100644 --- a/apps/ann_b2b/AnnounceB2B.cpp +++ b/apps/ann_b2b/AnnounceB2B.cpp @@ -31,6 +31,8 @@ #include "AmUtils.h" #include "AmApi.h" +#include "AmMediaProcessor.h" + #include "sems.h" #include "log.h" @@ -125,6 +127,10 @@ void AnnounceCallerDialog::process(AmEvent* event) if (getCalleeStatus() != None) return; + // detach this session from the media + // because we will stay in signaling only + AmMediaProcessor::instance()->removeSession(this); + connectCallee(callee_addr, callee_uri); return; } diff --git a/apps/conf_auth/conf_auth.py b/apps/conf_auth/conf_auth.py index f8229926..2b6b2dd0 100644 --- a/apps/conf_auth/conf_auth.py +++ b/apps/conf_auth/conf_auth.py @@ -107,6 +107,7 @@ class IvrDialog(IvrDialogBase): def onEmptyQueue(self): if self.state == connect: debug("connecting to " + self.conf_to + "uri: " + self.conf_uri) + self.disconnectMedia() self.mute() self.setRelayonly() self.connectCallee(self.conf_to, self.conf_uri) diff --git a/apps/ivr/Ivr.cpp b/apps/ivr/Ivr.cpp index 15598671..af1ccb89 100644 --- a/apps/ivr/Ivr.cpp +++ b/apps/ivr/Ivr.cpp @@ -661,6 +661,10 @@ void IvrDialog::onSipRequest(const AmSipRequest& r){ AmB2BSession::onSipRequest(r); } +void IvrDialog::onRtpTimeout() { + callPyEventHandler("onRtpTimeout",NULL); +} + void IvrDialog::process(AmEvent* event) { DBG("IvrDialog::process\n"); diff --git a/apps/ivr/Ivr.h b/apps/ivr/Ivr.h index 40ef1ea3..68a4534a 100644 --- a/apps/ivr/Ivr.h +++ b/apps/ivr/Ivr.h @@ -144,6 +144,7 @@ public: void onSipReply(const AmSipReply& r); void onSipRequest(const AmSipRequest& r); + void onRtpTimeout(); }; #endif diff --git a/apps/ivr/IvrDialogBase.cpp b/apps/ivr/IvrDialogBase.cpp index 1a4b02cc..37e33254 100644 --- a/apps/ivr/IvrDialogBase.cpp +++ b/apps/ivr/IvrDialogBase.cpp @@ -3,6 +3,7 @@ #include "Ivr.h" #include "IvrSipDialog.h" +#include "AmMediaProcessor.h" //#include "AmSessionTimer.h" @@ -66,6 +67,18 @@ IvrDialogBase_dealloc(IvrDialogBase* self) // // Event handlers // +static PyObject* IvrDialogBase_onRtpTimeout(IvrDialogBase* self, PyObject*) +{ + DBG("no script implementation for onRtpTimeout(). Stopping session. \n"); + + assert(self->p_dlg); + self->p_dlg->setStopped(); + self->p_dlg->postEvent(0); + + Py_INCREF(Py_None); + return Py_None; +} + // static PyObject* IvrDialogBase_onSessionStart(IvrDialogBase* self, PyObject*) // { // DBG("no script implementation for onSessionStart(self,hdrs) !!!\n"); @@ -234,6 +247,29 @@ static PyObject* IvrDialogBase_unmute(IvrDialogBase* self, PyObject* args) return Py_None; } +static PyObject* IvrDialogBase_remove_mediaprocessor(IvrDialogBase* self, + PyObject* args) +{ + assert(self->p_dlg); + + AmMediaProcessor::instance()->removeSession(self->p_dlg); + + Py_INCREF(Py_None); + return Py_None; +} + +static PyObject* IvrDialogBase_add_mediaprocessor(IvrDialogBase* self, + PyObject* args) +{ + assert(self->p_dlg); + + AmMediaProcessor::instance()->addSession(self->p_dlg, + self->p_dlg->getCallgroup()); + + Py_INCREF(Py_None); + return Py_None; +} + // DTMF static PyObject* IvrDialogBase_enableDTMFDetection(IvrDialogBase* self, PyObject* args) @@ -416,7 +452,13 @@ IvrDialogBase_redirect(IvrDialogBase *self, PyObject* args) static PyMethodDef IvrDialogBase_methods[] = { + // Event handlers + + {"onRtpTimeout", (PyCFunction)IvrDialogBase_onRtpTimeout, METH_NOARGS, + "Gets called on RTP timeout" + }, + // {"onSessionStart", (PyCFunction)IvrDialogBase_onSessionStart, METH_VARARGS, // "Gets called on session start" // }, @@ -457,9 +499,14 @@ static PyMethodDef IvrDialogBase_methods[] = { "mute the RTP stream (don't send packets)" }, {"unmute", (PyCFunction)IvrDialogBase_unmute, METH_NOARGS, - "unmute the RTP stream (do send packets)" + "unmute the RTP stream (send packets)" + }, + {"connectMedia", (PyCFunction)IvrDialogBase_add_mediaprocessor, METH_NOARGS, + "enable the processing of audio and RTP" + }, + {"disconnectMedia", (PyCFunction)IvrDialogBase_remove_mediaprocessor, METH_NOARGS, + "disable the processing of audio and RTP" }, - // DTMF {"enableDTMFDetection", (PyCFunction)IvrDialogBase_enableDTMFDetection, METH_NOARGS, "enable the dtmf detection" @@ -467,7 +514,6 @@ static PyMethodDef IvrDialogBase_methods[] = { {"disableDTMFDetection", (PyCFunction)IvrDialogBase_disableDTMFDetection, METH_NOARGS, "disable the dtmf detection" }, - // B2B {"connectCallee", (PyCFunction)IvrDialogBase_b2b_connectCallee, METH_VARARGS, "call given party as (new) callee," diff --git a/core/AmMediaProcessor.cpp b/core/AmMediaProcessor.cpp index b833bcdf..9a16247d 100644 --- a/core/AmMediaProcessor.cpp +++ b/core/AmMediaProcessor.cpp @@ -26,6 +26,7 @@ */ #include "AmMediaProcessor.h" +#include "AmRtpStream.h" #include #include @@ -112,8 +113,16 @@ void AmMediaProcessor::addSession(AmSession* s, postRequest(new SchedRequest(InsertSession,s)); } -void AmMediaProcessor::removeSession(AmSession* s) -{ +void AmMediaProcessor::clearSession(AmSession* s) { + removeFromProcessor(s, ClearSession); +} + +void AmMediaProcessor::removeSession(AmSession* s) { + removeFromProcessor(s, RemoveSession); +} + +void AmMediaProcessor::removeFromProcessor(AmSession* s, + unsigned int r_type) { DBG("AmMediaProcessor::removeSession\n"); group_mut.lock(); // get scheduler @@ -140,7 +149,7 @@ void AmMediaProcessor::removeSession(AmSession* s) session2callgroup.erase(s); group_mut.unlock(); - threads[sched_thread]->postRequest(new SchedRequest(RemoveSession,s)); + threads[sched_thread]->postRequest(new SchedRequest(r_type,s)); } /* the actual session scheduler thread */ @@ -228,10 +237,14 @@ void AmMediaProcessorThread::processAudio(unsigned int ts) break; case RTP_TIMEOUT: + postRequest(new SchedRequest(AmMediaProcessor::RemoveSession,s)); + s->postEvent(new AmRtpTimeoutEvent()); + break; + case RTP_BUFFER_SIZE: default: ERROR("AmRtpAudio::receive() returned %i\n",ret); - postRequest(new SchedRequest(AmMediaProcessor::RemoveSession,s)); + postRequest(new SchedRequest(AmMediaProcessor::ClearSession,s)); break; } } @@ -294,6 +307,17 @@ void AmMediaProcessorThread::process(AmEvent* e) break; case AmMediaProcessor::RemoveSession:{ + AmSession* s = sr->s; + set::iterator s_it = sessions.find(s); + if(s_it != sessions.end()){ + sessions.erase(s_it); + s->detached.set(true); + DBG("Session removed from the scheduler\n"); + } + } + break; + + case AmMediaProcessor::ClearSession:{ AmSession* s = sr->s; set::iterator s_it = sessions.find(s); if(s_it != sessions.end()){ @@ -305,6 +329,7 @@ void AmMediaProcessorThread::process(AmEvent* e) } break; + default: ERROR("AmMediaProcessorThread::process: unknown event id."); break; diff --git a/core/AmMediaProcessor.h b/core/AmMediaProcessor.h index 6cde1aa8..dcc7c907 100644 --- a/core/AmMediaProcessor.h +++ b/core/AmMediaProcessor.h @@ -97,15 +97,26 @@ class AmMediaProcessor AmMediaProcessor(); ~AmMediaProcessor(); - + + void removeFromProcessor(AmSession* s, unsigned int r_type); public: - enum { InsertSession, RemoveSession }; + /** + * InsertSession : inserts the session to the processor + * RemoveSession : remove the session from the processor + * ClearSession : remove the session from processor and clear audio + */ + enum { InsertSession, RemoveSession, ClearSession }; static AmMediaProcessor* instance(); void init(); + /** Add session s to processor */ void addSession(AmSession* s, const string& callgroup); + /** Remove session s from processor */ void removeSession(AmSession* s); + /** Remove session s from processor and clear its audio */ + void clearSession(AmSession* s); + }; diff --git a/core/AmRtpStream.cpp b/core/AmRtpStream.cpp index 88d7884e..ee39fa5f 100644 --- a/core/AmRtpStream.cpp +++ b/core/AmRtpStream.cpp @@ -402,7 +402,7 @@ int AmRtpStream::nextAudioPacket(AmRtpPacket& p, unsigned int ts, unsigned int m timersub(&now,&last_recv_time,&diff); if(diff.tv_sec > DEAD_RTP_TIME){ - WARN("Last received packet is too old.\n"); + WARN("RTP Timeout detected. Last received packet is too old.\n"); DBG("diff.tv_sec = %i\n",(unsigned int)diff.tv_sec); return RTP_TIMEOUT; } diff --git a/core/AmRtpStream.h b/core/AmRtpStream.h index ae26d6ea..1b1a8543 100644 --- a/core/AmRtpStream.h +++ b/core/AmRtpStream.h @@ -33,6 +33,7 @@ #include "AmThread.h" #include "SampleArray.h" #include "AmRtpPacket.h" +#include "AmEvent.h" #include @@ -253,6 +254,16 @@ struct AmRtpStreamInfo AmAudio* audio_rec = NULL); }; +class AmRtpTimeoutEvent + : public AmEvent +{ + +public: + AmRtpTimeoutEvent() + : AmEvent(0) { } + ~AmRtpTimeoutEvent() { } +}; + #endif // Local Variables: diff --git a/core/AmSession.cpp b/core/AmSession.cpp index 78a3d1d1..ea5c4b6c 100644 --- a/core/AmSession.cpp +++ b/core/AmSession.cpp @@ -334,7 +334,11 @@ void AmSession::on_stop() { //sess_stopped.set(true); DBG("AmSession::on_stop()\n"); - AmMediaProcessor::instance()->removeSession(this); + + if (!getDetached()) + AmMediaProcessor::instance()->clearSession(this); + else + clearAudio(); } void AmSession::destroy() @@ -442,6 +446,12 @@ void AmSession::process(AmEvent* ev) onDtmf(dtmf_ev->event(), dtmf_ev->duration()); return; } + + AmRtpTimeoutEvent* timeout_ev = dynamic_cast(ev); + if(timeout_ev){ + onRtpTimeout(); + return; + } } @@ -509,13 +519,24 @@ void AmSession::onSipRequest(const AmSipRequest& req) } } + void AmSession::onSipReply(const AmSipReply& reply) { CALL_EVENT_H(onSipReply,reply); int status = dlg.getStatus(); dlg.updateStatus(reply); - + + if (status != dlg.getStatus()) + DBG("Dialog status changed %s -> %s (stopped=%s) \n", + AmSipDialog::status2str[status], + AmSipDialog::status2str[dlg.getStatus()], + sess_stopped.get() ? "true" : "false"); + else + DBG("Dialog status stays %s (stopped=%s)\n", AmSipDialog::status2str[status], + sess_stopped.get() ? "true" : "false"); + + if (negotiate_onreply) { if(status < AmSipDialog::Connected){ @@ -650,6 +671,12 @@ void AmSession::onSendReply(const AmSipRequest& req, unsigned int code, CALL_EVENT_H(onSendReply,req,code,reason,content_type,body,hdrs); } +void AmSession::onRtpTimeout() +{ + DBG("stopping Session.\n"); + setStopped(); +} + void AmSession::sendUpdate() { dlg.update(""// getRequestHeaders("UPDATE") diff --git a/core/AmSession.h b/core/AmSession.h index a3236197..c17bb43c 100644 --- a/core/AmSession.h +++ b/core/AmSession.h @@ -343,6 +343,10 @@ public: virtual void onSipRequest(const AmSipRequest& req); virtual void onSipReply(const AmSipReply& reply); + + /** This callback is called if RTP timeout encountered */ + virtual void onRtpTimeout(); + /* only called by AmSipDialog */ virtual void onSendRequest(const string& method, const string& content_type, @@ -358,7 +362,6 @@ public: string& hdrs); }; - #endif // Local Variables: diff --git a/core/AmSipDialog.cpp b/core/AmSipDialog.cpp index a5e16802..2114576c 100644 --- a/core/AmSipDialog.cpp +++ b/core/AmSipDialog.cpp @@ -5,6 +5,13 @@ #include "AmServer.h" #include "sems.h" +char* AmSipDialog::status2str[4] = { + "Disconnected", + "Pending", + "Connected", + "Disconnecting" }; + + AmSipDialog::~AmSipDialog() { DBG("callid = %s\n",callid.c_str()); diff --git a/core/AmSipDialog.h b/core/AmSipDialog.h index b53fcd6e..751ba86f 100644 --- a/core/AmSipDialog.h +++ b/core/AmSipDialog.h @@ -81,6 +81,8 @@ public: Disconnecting }; + static char* status2str[4]; + string user; // local user string domain; // local domain string sip_ip; // destination IP of first received message