anyrtpproxy can handled most commands

TODO: start threads for relaying

4 files changed, 66 insertions, 13 deletions

diff --git a/anyrtpproxy/anyrtpproxy.cpp b/anyrtpproxy/anyrtpproxy.cpp
index df79563..280f99b 100644
--- a/anyrtpproxy/anyrtpproxy.cpp
+++ b/anyrtpproxy/anyrtpproxy.cpp
@@ -248,10 +248,6 @@ int main(int argc, char* argv[])
 
   SyncQueue queue;
 	ConnectToList connect_to = gOpt.getConnectTo();
-// Example
-//  gRtpSessionTable.addSession(std::string("callid"),RtpSession());
-//  SyncCommand sc (std::string("callid"));
-//  queue.push(sc);
 	ThreadParam p( queue,*(new OptionConnectTo()));
   if ( gOpt.getLocalSyncPort())
     pthread_create(&syncListenerThread, NULL, syncListener, &p);
@@ -264,7 +260,7 @@ int main(int argc, char* argv[])
     pthread_create(& connectThreads.back(),  NULL, syncConnector, point);
   }
 
-  CommandHandler cmd(gOpt.getControlInterface().addr_, gOpt.getControlInterface().port_);
+  CommandHandler cmd(queue, gOpt.getControlInterface().addr_, gOpt.getControlInterface().port_);
   
   int ret = sig.run();
   return ret;
diff --git a/anyrtpproxy/commandHandler.cpp b/anyrtpproxy/commandHandler.cpp
index cac5c4b..efb3c82 100644
--- a/anyrtpproxy/commandHandler.cpp
+++ b/anyrtpproxy/commandHandler.cpp
@@ -37,15 +37,20 @@
 #include "commandHandler.h"
 #include "../buffer.h"
 #include "../log.h"
+#include "../syncQueue.h"
+#include "../syncCommand.h"
+#include "../rtpSessionTable.h"
 
 #define MAX_COMMAND_LENGTH 1000
 
-CommandHandler::CommandHandler(u_int16_t lp) : running_(true), control_sock_(lp), local_address_("0.0.0.0"), local_port_(lp)
+CommandHandler::CommandHandler(SyncQueue& q, u_int16_t lp) : queue_(q), running_(true), control_sock_(lp), 
+                                                             local_address_("0.0.0.0"), local_port_(lp)
 {
   pthread_create(&thread_, NULL, run, this);
 }
 
-CommandHandler::CommandHandler(string la, u_int16_t lp) : running_(true), control_sock_(la, lp), local_address_(la), local_port_(lp)
+CommandHandler::CommandHandler(SyncQueue& q, string la, u_int16_t lp) : queue_(q), running_(true), control_sock_(la, lp), 
+                                                                        local_address_(la), local_port_(lp)
 {
   pthread_create(&thread_, NULL, run, this);
 }
@@ -159,7 +164,26 @@ string CommandHandler::handleRequest(string modifiers, string call_id, string ad
   std::cout << "received request[" << modifiers << "] command ('" << call_id << "','" << addr  << "','" << port 
             << "','" << from_tag << "','" << to_tag << "')" << std::endl;
 
-  return RET_OK;
+  try 
+  {
+    gRtpSessionTable.addSession(call_id, new RtpSession());
+    u_int16_t port = 35000; // TODO: get next available port
+    RtpSession& session = gRtpSessionTable.getSession(call_id);
+    session.setLocalPort(port);   
+    session.setLocalAddr("0.0.0.0"); // TODO: read this from config;
+    session.setRemotePort1(port);
+    session.setRemoteAddr1(addr);
+    SyncCommand sc(call_id);
+    queue_.push(sc);
+
+    ostringstream oss;
+    oss << port;
+    return oss.str();
+  }
+  catch(std::exception& e)
+  {
+    return RET_ERR_UNKNOWN; // TODO: change to corret error value
+  }
 }
 
 string CommandHandler::handleResponse(string modifiers, string call_id, string addr, string port, string from_tag, string to_tag)
@@ -167,14 +191,42 @@ string CommandHandler::handleResponse(string modifiers, string call_id, string a
   std::cout << "received response[" << modifiers << "] command ('" << call_id << "','" << addr  << "','" << port 
             << "','" << from_tag << "','" << to_tag << "')" << std::endl;
 
-  return RET_OK;
+  try
+  {
+    RtpSession& session = gRtpSessionTable.getSession(call_id);
+    u_int16_t port = session.getLocalPort();
+    session.setRemotePort2(port);
+    session.setRemoteAddr2(addr);
+    SyncCommand sc(call_id);
+    queue_.push(sc);
+
+    ostringstream oss;
+    oss << port;
+    return oss.str();
+  }
+  catch(std::exception& e)
+  {
+    return RET_ERR_UNKNOWN; // TODO: change to corret error value
+  }
 }
 
 string CommandHandler::handleDelete(string call_id, string from_tag, string to_tag)
 {
   std::cout << "received delete command ('" << call_id << "','" << from_tag << "','" << to_tag << "')" << std::endl;
 
-  return RET_OK;
+  try
+  {
+    RtpSession& session = gRtpSessionTable.getSession(call_id);
+    session.isDead(true);
+    SyncCommand sc(call_id);
+    queue_.push(sc);
+
+    return RET_OK;
+  }
+  catch(std::exception& e)
+  {
+    return RET_ERR_UNKNOWN; // TODO: change to corret error value
+  }
 }
 
 string CommandHandler::handleVersion()
@@ -194,7 +246,7 @@ string CommandHandler::handleVersionF(string date_code)
 
 string CommandHandler::handleInfo()
 {
-  std::cout << "received info command" << std::endl;  
+  std::cout << "received info command, ignoring" << std::endl;  
   return RET_OK;
 }
 
diff --git a/anyrtpproxy/commandHandler.h b/anyrtpproxy/commandHandler.h
index 41bff2d..6e75bc4 100644
--- a/anyrtpproxy/commandHandler.h
+++ b/anyrtpproxy/commandHandler.h
@@ -34,14 +34,15 @@
 #include <string>
 #include "../datatypes.h"
 #include "../PracticalSocket.h"
+#include "../syncQueue.h"
 
 using std::string;
 
 class CommandHandler
 {
 public:
-  CommandHandler(u_int16_t lp);
-  CommandHandler(string la, u_int16_t lp);
+  CommandHandler(SyncQueue& q, u_int16_t lp);
+  CommandHandler(SyncQueue& q, string la, u_int16_t lp);
   ~CommandHandler();
   
   bool isRunning();
@@ -54,6 +55,7 @@ public:
 
   #define RET_OK "0"
   #define RET_ERR_SYNTAX "E1"
+  #define RET_ERR_UNKNOWN "E2"
 
   #define BASE_VERSION "20040107"
   #define SUP_VERSION "20050322"
@@ -73,6 +75,8 @@ private:
   string handleInfo();
 
   pthread_t thread_;
+  SyncQueue& queue_;
+
   bool running_;
   UDPSocket control_sock_;
   string local_address_;
diff --git a/anyrtpproxy/options.cpp b/anyrtpproxy/options.cpp
index 0c55c90..b81f348 100644
--- a/anyrtpproxy/options.cpp
+++ b/anyrtpproxy/options.cpp
@@ -57,6 +57,7 @@ Options::Options() : control_interface_("0.0.0.0", 22222)
   username_ = "nobody";
   chroot_dir_ = "/var/run";
   daemonize_ = true;
+	local_sync_port_ = 2023;
 }
 
 Options::~Options()