Modified the orkaudio API, adding the record and pause HTTP commands. The record command commences or un-pauses recording while the pause command pauses recording - discarding RTP packets from when the pause command is issued. Both commands require the orkuid and party to be specified as HTTP parameters. Note that this represents a change in the arguments required for the StartCapture function in DLLs. Also added an event streaming feature which streams out all tape messages as they are reported in real-time to a client connected. Clients should connect via HTTP, on port 59150. The port is configurable by setting the parameter EventStreamingServerPort in config.xml

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

5 files changed, 145 insertions, 10 deletions

diff --git a/orkaudio/OrkAudio.cpp b/orkaudio/OrkAudio.cpp
index 0083f8b..edda45a 100644
--- a/orkaudio/OrkAudio.cpp
+++ b/orkaudio/OrkAudio.cpp
@@ -44,7 +44,7 @@
 #include "filters/audiogain/AudioGain.h"
 #include "TapeProcessor.h"
 #include <list>
-
+#include "EventStreaming.h"
 
 static volatile bool serviceStop = false;
 
@@ -213,6 +213,8 @@ void MainThread()
 	ObjectFactory::GetSingleton()->RegisterObject(objRef);
 	objRef.reset(new RecordMsg);
 	ObjectFactory::GetSingleton()->RegisterObject(objRef);
+	objRef.reset(new PauseMsg);
+	ObjectFactory::GetSingleton()->RegisterObject(objRef);
 	//objRef.reset(new TestMsg);
 	//ObjectFactory::GetSingleton()->RegisterObject(objRef);
 
@@ -276,6 +278,12 @@ void MainThread()
 		LOG4CXX_INFO(LOG.rootLog, CStdString("Failed to create Http server"));
 	}
 
+	// Create streaming server on port 59150 (default)
+	if(!ACE_Thread_Manager::instance()->spawn(ACE_THR_FUNC(EventStreamingServer::run), (void *)CONFIG.m_eventStreamingServerPort))
+	{
+		LOG4CXX_INFO(LOG.rootLog, CStdString("Failed to create event streaming server"));
+	}
+
 	if(capturePluginOk)
 	{
 		CapturePluginProxy::Singleton()->Run();
diff --git a/orkaudio/audiocaptureplugins/common/AudioCapturePluginCommon.h b/orkaudio/audiocaptureplugins/common/AudioCapturePluginCommon.h
index 4697317..a6defae 100644
--- a/orkaudio/audiocaptureplugins/common/AudioCapturePluginCommon.h
+++ b/orkaudio/audiocaptureplugins/common/AudioCapturePluginCommon.h
@@ -30,8 +30,9 @@ DLL_EXPORT void __CDECL__  Run();
 DLL_EXPORT void __CDECL__  Initialize();
 DLL_EXPORT void __CDECL__  Shutdown();
 DLL_EXPORT void __CDECL__  Configure(DOMNode*);
-DLL_EXPORT void __CDECL__  StartCapture(CStdString& party);
+DLL_EXPORT void __CDECL__  StartCapture(CStdString& party, CStdString& orkuid);
 DLL_EXPORT void __CDECL__  StopCapture(CStdString& party);
+DLL_EXPORT void __CDECL__  PauseCapture(CStdString& party, CStdString& orkuid);
 }
 
 #endif
diff --git a/orkaudio/audiocaptureplugins/voip/RtpSession.cpp b/orkaudio/audiocaptureplugins/voip/RtpSession.cpp
index cf1ba08..a8b2d40 100644
--- a/orkaudio/audiocaptureplugins/voip/RtpSession.cpp
+++ b/orkaudio/audiocaptureplugins/voip/RtpSession.cpp
@@ -2383,6 +2383,36 @@ void RtpSessions::Hoover(time_t now)
 	}
 }
 
+void RtpSessions::StartCaptureOrkuid(CStdString& orkuid)
+{
+	std::map<CStdString, RtpSessionRef>::iterator pair;
+	bool found = false;
+	CStdString logMsg;
+	RtpSessionRef session;
+
+	for(pair = m_byIpAndPort.begin(); pair != m_byIpAndPort.end() && found == false; pair++)
+	{
+		session = pair->second;
+
+		if(session->OrkUidMatches(orkuid))
+		{
+			session->m_keep = true;
+			found = true;
+		}
+	}
+
+	if(found)
+	{
+		logMsg.Format("[%s] StartCaptureOrkuid: Started capture, orkuid:%s", session->m_trackingId, orkuid);
+	}
+	else
+	{
+		logMsg.Format("StartCaptureOrkuid: No session has orkuid:%s", orkuid);
+	}
+
+	LOG4CXX_INFO(m_log, logMsg);
+}
+
 void RtpSessions::StartCapture(CStdString& party)
 {
 	std::map<CStdString, RtpSessionRef>::iterator pair;
@@ -2403,16 +2433,76 @@ void RtpSessions::StartCapture(CStdString& party)
 
 	if(found)
 	{
-		logMsg.Format("[%s] Started capture, party:%s", session->m_trackingId, party);
+		logMsg.Format("[%s] StartCapture: Started capture, party:%s", session->m_trackingId, party);
 	}	
 	else
 	{
-		logMsg.Format("No session has party %s", party);
+		logMsg.Format("StartCapture: No session has party %s", party);
 	}
 	
 	LOG4CXX_INFO(m_log, logMsg);
 }
 
+void RtpSessions::PauseCapture(CStdString& party)
+{
+	std::map<CStdString, RtpSessionRef>::iterator pair;
+	bool found = false;
+	CStdString logMsg;
+	RtpSessionRef session;
+
+	for(pair = m_byIpAndPort.begin(); pair != m_byIpAndPort.end() && found == false; pair++)
+	{
+		session = pair->second;
+
+		if (session->PartyMatches(party))
+		{
+			session->m_keep = false;
+			found = true;
+		}
+	}
+
+	if(found)
+	{
+		logMsg.Format("[%s] PauseCapture: Paused capture, party:%s", session->m_trackingId, party);
+	}	
+	else
+	{
+		logMsg.Format("PauseCapture: No session has party %s", party);
+	}
+	
+	LOG4CXX_INFO(m_log, logMsg);
+}
+
+void RtpSessions::PauseCaptureOrkuid(CStdString& orkuid)
+{
+	std::map<CStdString, RtpSessionRef>::iterator pair;
+	bool found = false;
+	CStdString logMsg;
+	RtpSessionRef session;
+
+	for(pair = m_byIpAndPort.begin(); pair != m_byIpAndPort.end() && found == false; pair++)
+	{
+		session = pair->second;
+
+		if(session->OrkUidMatches(orkuid))
+		{
+			session->m_keep = false;
+			found = true;
+		}
+	}
+
+	if(found)
+	{
+		logMsg.Format("[%s] PauseCaptureOrkuid: Paused capture, orkuid:%s", session->m_trackingId, orkuid);
+	}
+	else
+	{
+		logMsg.Format("PauseCaptureOrkuid: No session has orkuid:%s", orkuid);
+	}
+
+	LOG4CXX_INFO(m_log, logMsg);
+}
+
 //==========================================================
 SipInviteInfo::SipInviteInfo()
 {
diff --git a/orkaudio/audiocaptureplugins/voip/RtpSession.h b/orkaudio/audiocaptureplugins/voip/RtpSession.h
index a24c882..6695985 100644
--- a/orkaudio/audiocaptureplugins/voip/RtpSession.h
+++ b/orkaudio/audiocaptureplugins/voip/RtpSession.h
@@ -257,6 +257,9 @@ public:
 	void Hoover(time_t now);
 	EndpointInfoRef GetEndpointInfo(struct in_addr endpointIp);
 	void StartCapture(CStdString& party);
+	void StartCaptureOrkuid(CStdString& orkuid);
+	void PauseCapture(CStdString& party);
+	void PauseCaptureOrkuid(CStdString& orkuid);
 
 private:
 	RtpSessionRef findByEndpointIp(struct in_addr endpointIpAddr, int passThruPartyId = 0);
diff --git a/orkaudio/audiocaptureplugins/voip/VoIp.cpp b/orkaudio/audiocaptureplugins/voip/VoIp.cpp
index 25cda38..63fa355 100644
--- a/orkaudio/audiocaptureplugins/voip/VoIp.cpp
+++ b/orkaudio/audiocaptureplugins/voip/VoIp.cpp
@@ -83,8 +83,9 @@ public:
 	void Initialize();
 	void Run();
 	void Shutdown();
-	void StartCapture(CStdString& port);
+	void StartCapture(CStdString& port, CStdString& orkuid);
 	void StopCapture(CStdString& port);
+	void PauseCapture(CStdString& port, CStdString& orkuid);
 	void ReportPcapStats();
 	pcap_t* OpenDevice(CStdString& name);
 	void AddPcapDeviceToMap(CStdString& deviceName, pcap_t* pcapHandle);
@@ -3199,7 +3200,12 @@ void VoIp::Shutdown()
 #endif
 }
 
-void VoIp::StartCapture(CStdString& port)
+void VoIp::StartCapture(CStdString& port, CStdString& orkuid)
+{
+	;
+}
+
+void VoIp::PauseCapture(CStdString& port, CStdString& orkuid)
 {
 	;
 }
@@ -3225,15 +3231,42 @@ void __CDECL__ Shutdown()
 	VoIpSingleton::instance()->Shutdown();
 }
 
-void __CDECL__ StartCapture(CStdString& party)
+void __CDECL__ StartCapture(CStdString& party, CStdString& orkuid)
 {
 	CStdString logMsg;
 
-	//logMsg.Format("StartCapture:%s", party);
-	//LOG4CXX_INFO(s_voipPluginLog, logMsg);
+	logMsg.Format("StartCapture: party:%s orkuid:%s", party, orkuid);
+	LOG4CXX_INFO(s_voipPluginLog, logMsg);
 
 	MutexSentinel mutexSentinel(s_mutex);
-	RtpSessionsSingleton::instance()->StartCapture(party);
+
+	if(orkuid.size())
+	{
+		RtpSessionsSingleton::instance()->StartCaptureOrkuid(orkuid);
+	}
+	else
+	{
+		RtpSessionsSingleton::instance()->StartCapture(party);
+	}
+}
+
+void __CDECL__ PauseCapture(CStdString& party, CStdString& orkuid)
+{
+	CStdString logMsg;
+
+	logMsg.Format("PauseCapture: party:%s orkuid:%s", party, orkuid);
+	LOG4CXX_INFO(s_voipPluginLog, logMsg);
+
+	MutexSentinel mutexSentinel(s_mutex);
+
+	if(orkuid.size())
+	{
+		RtpSessionsSingleton::instance()->PauseCaptureOrkuid(orkuid);
+	}
+	else
+	{
+		RtpSessionsSingleton::instance()->PauseCapture(party);
+	}
 }
 
 void __CDECL__ StopCapture(CStdString& party)