1. Subsequent SIP INVITES that could be associated to an existing session now disregarded because they could be disrupting valid sessions. We need to store the new INVITES and only use them when they are validated by a matching RTP stream.

2. New RTP streams are now logged within one session by the VoIP plugin.
3. Added a Debug config boolean that has the trackingId alpha counter reset to 0 (AAA) when enabled.

git-svn-id: https://oreka.svn.sourceforge.net/svnroot/oreka/trunk@380 09dcff7a-b715-0410-9601-b79a96267cd0

6 files changed, 124 insertions, 22 deletions

diff --git a/orkaudio/audiocaptureplugins/voip/RtpSession.cpp b/orkaudio/audiocaptureplugins/voip/RtpSession.cpp
index 9bc0e01..a8ca873 100644
--- a/orkaudio/audiocaptureplugins/voip/RtpSession.cpp
+++ b/orkaudio/audiocaptureplugins/voip/RtpSession.cpp
@@ -10,7 +10,7 @@
  * Please refer to http://www.gnu.org/copyleft/gpl.html
  *
  */
-
+#pragma warning( disable: 4786 ) // disables truncated symbols in browse-info warning
 #define _WINSOCKAPI_		// prevents the inclusion of winsock.h
 
 #include "Utils.h"
@@ -19,6 +19,7 @@
 #include "AudioCapturePlugin.h"
 #include "AudioCapturePluginCommon.h"
 #include <list>
+#include "ConfigManager.h"
 #include "VoIpConfig.h"
 #include "ace/OS_NS_arpa_inet.h"
 
@@ -456,6 +457,16 @@ bool RtpSession::AddRtpPacket(RtpPacketInfoRef& rtpPacket)
 
 	m_numRtpPackets++;
 
+	bool hasSourceAddress = m_rtpAddressList.HasAddressOrAdd(rtpPacket->m_sourceIp, rtpPacket->m_sourcePort);
+	bool hasDestAddress = m_rtpAddressList.HasAddressOrAdd(rtpPacket->m_destIp, rtpPacket->m_destPort);
+	if(	hasSourceAddress == false || hasDestAddress == false )
+	{
+		rtpPacket->ToString(logMsg);
+		logMsg.Format("[%s] new RTP stream s%d: %s", 
+							m_trackingId, channel, logMsg);
+		LOG4CXX_INFO(m_log, logMsg);
+	}
+
 	if(m_log->isDebugEnabled())
 	{
 		CStdString debug;
@@ -542,6 +553,10 @@ CStdString RtpSession::ProtocolToString(int protocolEnum)
 RtpSessions::RtpSessions()
 {
 	m_log = Logger::getLogger("rtpsessions");
+	if(CONFIG.m_debug)
+	{
+		m_alphaCounter.Reset();
+	}
 }
 
 
@@ -552,7 +567,7 @@ void RtpSessions::ReportSipInvite(SipInviteInfoRef& invite)
 
 	CStdString ipAndPort = CStdString(szFromRtpIp) + "," + invite->m_fromRtpPort;
 	std::map<CStdString, RtpSessionRef>::iterator pair;
-	
+
 	pair = m_byIpAndPort.find(ipAndPort);
 	if (pair != m_byIpAndPort.end())
 	{
@@ -566,22 +581,25 @@ void RtpSessions::ReportSipInvite(SipInviteInfoRef& invite)
 		RtpSessionRef session = pair->second;
 		if(!session->m_ipAndPort.Equals(ipAndPort))
 		{
-			// The session RTP connection address has changed
-			// Remove session from IP and Port map
-			m_byIpAndPort.erase(session->m_ipAndPort);
-			// ... update
-			session->m_ipAndPort = ipAndPort;
-			session->ReportSipInvite(invite);
-			// ... and reinsert
-			m_byIpAndPort.insert(std::make_pair(session->m_ipAndPort, session));
-
-			LOG4CXX_INFO(m_log, "[" + session->m_trackingId + "] updated with new INVITE data");
+			//===== The following is disabled because it disrupts valid sessions	====
+			//===== We need to make sure that at least one RTP packet has been		====
+			//===== seen that validates any new INVITE associated with the session	====
+			//// The session RTP connection address has changed
+			//// Remove session from IP and Port map
+			//m_byIpAndPort.erase(session->m_ipAndPort);
+			//// ... update
+			//session->m_ipAndPort = ipAndPort;
+			//session->ReportSipInvite(invite);
+			//// ... and reinsert
+			//m_byIpAndPort.insert(std::make_pair(session->m_ipAndPort, session));
+			//
+			//LOG4CXX_INFO(m_log, "[" + session->m_trackingId + "] updated with new INVITE data");
 		}
 		return;
 	}
 
 	// create new session and insert into both maps
-	CStdString trackingId = alphaCounter.GetNext();
+	CStdString trackingId = m_alphaCounter.GetNext();
 	RtpSessionRef session(new RtpSession(trackingId));
 	session->m_ipAndPort = ipAndPort;
 	session->m_callId = invite->m_callId;
@@ -621,7 +639,7 @@ void RtpSessions::ReportSkinnyCallInfo(SkCallInfoStruct* callInfo, IpHeaderStruc
 	}
 
 	// create new session and insert into the callid map
-	CStdString trackingId = alphaCounter.GetNext();
+	CStdString trackingId = m_alphaCounter.GetNext();
 	RtpSessionRef session(new RtpSession(trackingId));
 	session->m_callId = callId;
 	session->m_endPointIp = ipHeader->ip_dest;	// CallInfo message always goes from CM to endpoint 
@@ -1012,7 +1030,7 @@ void RtpSessions::ReportRtpPacket(RtpPacketInfoRef& rtpPacket)
 	if(numSessionsFound == 0)
 	{
 		// create new Raw RTP session and insert into IP+Port map
-		CStdString trackingId = alphaCounter.GetNext();
+		CStdString trackingId = m_alphaCounter.GetNext();
 		RtpSessionRef session(new RtpSession(trackingId));
 		session->m_protocol = RtpSession::ProtRawRtp;
 		session->m_ipAndPort = ipAndPort;	// (1) In the case of a PSTN Gateway automated answer, This is the destination IP+Port of the first packet which is good, because it is usually the IP+Port of the PSTN Gateway.
diff --git a/orkaudio/audiocaptureplugins/voip/RtpSession.h b/orkaudio/audiocaptureplugins/voip/RtpSession.h
index d7f9659..77db42d 100644
--- a/orkaudio/audiocaptureplugins/voip/RtpSession.h
+++ b/orkaudio/audiocaptureplugins/voip/RtpSession.h
@@ -117,6 +117,7 @@ private:
 	unsigned int m_highestRtpSeqNumDelta;
 	double m_minRtpSeqDelta;
 	double m_minRtpTimestampDelta;
+	TcpAddressList m_rtpAddressList;
 };
 typedef boost::shared_ptr<RtpSession> RtpSessionRef;
 
@@ -147,7 +148,7 @@ private:
 	std::map<CStdString, RtpSessionRef> m_byCallId;
 	std::map<unsigned int, EndpointInfoRef> m_endpoints;
 	LoggerPtr m_log;
-	AlphaCounter alphaCounter;
+	AlphaCounter m_alphaCounter;
 };
 typedef ACE_Singleton<RtpSessions, ACE_Thread_Mutex> RtpSessionsSingleton;
 
diff --git a/orkbasecxx/Config.cpp b/orkbasecxx/Config.cpp
index 199b301..2dfa2f3 100644
--- a/orkbasecxx/Config.cpp
+++ b/orkbasecxx/Config.cpp
@@ -53,6 +53,7 @@ Config::Config()
 
 	m_reportingRetryDelay = 5;
 	m_clientTimeout = 5;
+	m_debug = DEBUG_DEFAULT;
 }
 
 void Config::Define(Serializer* s)
@@ -86,6 +87,7 @@ void Config::Define(Serializer* s)
 	s->CsvValue(CAPTURE_PORT_FILTERS_PARAM, m_capturePortFilters);
 	s->CsvValue(TAPE_PROCESSORS_PARAM, m_tapeProcessors);
 	s->IntValue(CAPTURE_FILE_BATCH_SIZE_KBYTE_PARAM, m_captureFileBatchSizeKByte);
+	s->BoolValue(DEBUG_PARAM, m_debug);
 }
 
 void Config::Validate()
diff --git a/orkbasecxx/Config.h b/orkbasecxx/Config.h
index 09fe6a4..30b74c5 100644
--- a/orkbasecxx/Config.h
+++ b/orkbasecxx/Config.h
@@ -73,6 +73,8 @@
 #define TAPE_PROCESSORS_PARAM "TapeProcessors"
 #define CAPTURE_FILE_BATCH_SIZE_KBYTE_PARAM "CaptureFileBatchSizeKByte"
 #define CAPTURE_FILE_BATCH_SIZE_KBYTE_DEFAULT 4
+#define DEBUG_PARAM "Debug"
+#define DEBUG_DEFAULT false
 
 class DLL_IMPORT_EXPORT_ORKBASE Config : public Object
 {
@@ -114,6 +116,7 @@ public:
 	std::list<CStdString> m_capturePortFilters;
 	std::list<CStdString> m_tapeProcessors;
 	int m_captureFileBatchSizeKByte;
+	bool m_debug;
 
 private:
 	log4cxx::LoggerPtr m_log;
diff --git a/orkbasecxx/Utils.cpp b/orkbasecxx/Utils.cpp
index cec44eb..d4e7f8a 100644
--- a/orkbasecxx/Utils.cpp
+++ b/orkbasecxx/Utils.cpp
@@ -1,5 +1,6 @@
 #include "Utils.h"
 #include "ace/OS_NS_stdio.h"
+#include "ace/OS_NS_arpa_inet.h"
 
 //========================================================
 // file related stuff
@@ -63,3 +64,46 @@ bool FileCanOpen(CStdString& path)
 	}
 	return false;
 }
+
+//=====================================================
+// Network related stuff
+
+void TcpAddress::ToString(CStdString& string)
+{
+	char szIp[16];
+	ACE_OS::inet_ntop(AF_INET, (void*)&ip, szIp, sizeof(szIp));
+
+	string.Format("%s,%u", szIp, port);
+}
+
+
+void TcpAddressList::AddAddress(struct in_addr ip, unsigned short port)
+{
+	TcpAddress addr;
+	addr.ip = ip;
+	addr.port = port;
+	m_addresses.push_back(addr);
+}
+
+bool TcpAddressList::HasAddress(struct in_addr ip, unsigned short port)
+{
+	for(std::list<TcpAddress>::iterator it = m_addresses.begin(); it != m_addresses.end(); it++)
+	{
+		if ((unsigned int)((*it).ip.s_addr) == (unsigned int)ip.s_addr  && (*it).port == port)
+		{
+			return true;
+		}
+	}
+	return false;
+}
+
+bool TcpAddressList::HasAddressOrAdd(struct in_addr ip, unsigned short port)
+{
+	if(HasAddress(ip, port) == false)
+	{
+		AddAddress(ip, port);
+		return false;
+	}
+	return true;
+}
+
diff --git a/orkbasecxx/Utils.h b/orkbasecxx/Utils.h
index 31abf5d..67f141e 100644
--- a/orkbasecxx/Utils.h
+++ b/orkbasecxx/Utils.h
@@ -14,6 +14,8 @@
 #ifndef __UTILS_H__
 #define __UTILS_H__
 
+#include <list>
+
 #include "ace/Guard_T.h"	// For some reason, this include must always come before the StdString include
 							// otherwise it gives the following compile error:
 							//	error C2039: 'TryEnterCriticalSection' : is not a member of '`global namespace''
@@ -80,6 +82,26 @@ typedef ACE_Guard<ACE_Thread_Mutex> MutexSentinel;
 
 
 //=====================================================
+// Network related stuff
+typedef struct 
+{
+	void ToString(CStdString& string);
+
+	in_addr ip;
+	unsigned short port;
+} TcpAddress;
+
+class DLL_IMPORT_EXPORT_ORKBASE TcpAddressList
+{
+public:
+	void AddAddress(struct in_addr ip, unsigned short port);
+	bool HasAddress(struct in_addr ip, unsigned short port);
+	bool HasAddressOrAdd(struct in_addr ip, unsigned short port);
+private:
+	std::list<TcpAddress> m_addresses;
+};
+
+//=====================================================
 // Miscellanous stuff
 
 /** A counter that generates a "counting" 3 character strings, i.e. aaa, aab, ..., zzz 
@@ -90,14 +112,21 @@ typedef ACE_Guard<ACE_Thread_Mutex> MutexSentinel;
 class AlphaCounter
 {
 public:
-	inline AlphaCounter::AlphaCounter()
+	inline AlphaCounter::AlphaCounter(int start = 0)
 	{
-		// Generate pseudo-random number from high resolution time least significant two bytes
-		ACE_hrtime_t hrtime = ACE_OS::gethrtime();
-		unsigned short srandom = (short)hrtime;
-		double drandom = (double)srandom/65536.0; 	// 0 <= random < 1 
+		if(start)
+		{
+			m_counter = start;
+		}
+		else
+		{
+			// Generate pseudo-random number from high resolution time least significant two bytes
+			ACE_hrtime_t hrtime = ACE_OS::gethrtime();
+			unsigned short srandom = (short)hrtime;
+			double drandom = (double)srandom/65536.0; 	// 0 <= random < 1 
 
-		m_counter = (unsigned int)(drandom*(26*26*26));
+			m_counter = (unsigned int)(drandom*(26*26*26));
+		}
 	}
 
 	inline CStdString AlphaCounter::GetNext()
@@ -120,6 +149,11 @@ public:
 		string.Format("%c%c%c", char1val, char2val, char3val);
 		return string;
 	}
+
+	inline void AlphaCounter::Reset()
+	{
+		m_counter = 0;
+	}
 private:
 	unsigned int m_counter;
 };