fixes RTP_TIMEOUT bug in signaling only b2b situations

- on RTP timeout, AmSession::onRtpTimeout is called
 - session can be removed from media processor (detached) 
   by AmMediaProcessor::removeSession
 - removing session from MediaProcessor and clearing audio 
   is AmMediaProcessor::clearSession (default action on 
   RTP error)
 - added detach from media processor to conf_auth and ann_b2b



git-svn-id: http://svn.berlios.de/svnroot/repos/sems/trunk@209 8eb893ce-cfd4-0310-b710-fb5ebe64c474

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;
     }

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)

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");

apps/ivr/Ivr.h

@@ -144,6 +144,7 @@ public:
     void onSipReply(const AmSipReply& r);
     void onSipRequest(const AmSipRequest& r);
 
+    void onRtpTimeout();
 };
 
 #endif

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,"

core/AmMediaProcessor.cpp

@@ -26,6 +26,7 @@
  */
 
 #include "AmMediaProcessor.h"
+#include "AmRtpStream.h"
 
 #include <assert.h>
 #include <sys/time.h>
@@ -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<AmSession*>::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<AmSession*>::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;

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);
+
 };
 
 

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;
     }

core/AmRtpStream.h

@@ -33,6 +33,7 @@
 #include "AmThread.h"
 #include "SampleArray.h"
 #include "AmRtpPacket.h"
+#include "AmEvent.h"
 
 #include <netinet/in.h>
 
@@ -253,6 +254,16 @@ struct AmRtpStreamInfo
 		    AmAudio* audio_rec = NULL);
 };
 
+class AmRtpTimeoutEvent
+	: public AmEvent
+{
+	
+public:
+	AmRtpTimeoutEvent() 
+		: AmEvent(0) { }
+	~AmRtpTimeoutEvent() { }
+};
+
 #endif
 
 // Local Variables:

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<AmRtpTimeoutEvent*>(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")

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:

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());

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