sbc: RTP relay mode

first version, wip.

todo:
 - variable number of streams (now fixed to 3 - consumes many ports)
 - symmetric RTP
 - multi interfaces
 - optimization: parse SDP only once, pass AmSdp object with request event

apps/sbc/SBC.cpp

@@ -583,6 +583,11 @@ void SBCDialog::onInvite(const AmSipRequest& req)
       replaceParameters(call_profile.auth_credentials.pwd, "auth_pwd", REPLACE_VALS);
   }
 
+  if (call_profile.rtprelay_enabled) {
+    DBG("Enabling RTP relay mode\n");
+    rtp_relay_enabled = true;
+  }
+
   // get timer
   if (call_profile.call_timer_enabled || call_profile.prepaid_enabled) {
     if (!timersSupported()) {
@@ -1012,6 +1017,14 @@ void SBCDialog::createCalleeSession()
 		  "to",   callee_dlg.remote_party.c_str(),
 		  "ruri", callee_dlg.remote_uri.c_str());
 
+  try {
+    initializeRTPRelay(callee_session);
+  }
+  catch (...) {
+    delete callee_session;
+    throw;
+  }
+
   callee_session->start();
   
   AmSessionContainer* sess_cont = AmSessionContainer::instance();

apps/sbc/SBCCallProfile.cpp

@@ -173,6 +173,8 @@ bool SBCCallProfile::readFromConfiguration(const string& name,
 
   refuse_with = cfg.getParameter("refuse_with");
 
+  rtprelay_enabled = cfg.getParameter("enable_rtprelay") == "yes";
+
   md5hash = "<unknown>";
   if (!cfg.getMD5(profile_file_name, md5hash)){
     ERROR("calculating MD5 of file %s\n", profile_file_name.c_str());
@@ -209,6 +211,8 @@ bool SBCCallProfile::readFromConfiguration(const string& name,
 	 sdpfilter_enabled?"en":"dis", FilterType2String(sdpfilter),
 	 sdpfilter_list.size());
 
+    INFO("SBC:      RTP relay %sabled\n", rtprelay_enabled?"en":"dis");
+
     INFO("SBC:      SST %sabled\n", sst_enabled?"en":"dis");
     INFO("SBC:      SIP auth %sabled\n", auth_enabled?"en":"dis");
     INFO("SBC:      call timer %sabled\n", call_timer_enabled?"en":"dis");
@@ -326,6 +330,7 @@ string SBCCallProfile::print() const {
   res += "prepaid_accmodule:    " + prepaid_accmodule + "\n";
   res += "prepaid_uuid:         " + prepaid_uuid + "\n";
   res += "prepaid_acc_dest:     " + prepaid_acc_dest + "\n";
+  res += "rtprelay_enabled:     " + string(rtprelay_enabled?"true":"false") + "\n";
 
   if (reply_translations.size()) {
     string reply_trans_codes;

apps/sbc/SBCCallProfile.h

@@ -88,7 +88,8 @@ struct SBCCallProfile {
 
   string refuse_with;
 
-  // todo: RTP forwarding mode
+  bool rtprelay_enabled;
+
   // todo: RTP transcoding mode
 
   SBCCallProfile()
@@ -99,7 +100,9 @@ struct SBCCallProfile {
     sst_enabled(false),
     auth_enabled(false),
     call_timer_enabled(false),
-    prepaid_enabled(false)
+    prepaid_enabled(false),
+    rtprelay_enabled(false)
+
   { }
 
   ~SBCCallProfile()

core/AmB2BSession.cpp

@@ -30,6 +30,7 @@
 #include "ampi/MonitoringAPI.h"
 #include "AmSipHeaders.h"
 #include "AmUtils.h"
+#include "AmRtpReceiver.h"
 
 #include <assert.h>
 
@@ -39,19 +40,27 @@
 
 AmB2BSession::AmB2BSession()
   : sip_relay_only(true),
-    b2b_mode(B2BMode_Transparent)
+    b2b_mode(B2BMode_Transparent),
+    rtp_relay_enabled(false)
 {
+  for (unsigned int i=0; i<MAX_RELAY_STREAMS; i++)
+    other_stream_fds[i] = 0;
 }
 
 AmB2BSession::AmB2BSession(const string& other_local_tag)
   : other_id(other_local_tag),
-    sip_relay_only(true)
+    sip_relay_only(true),
+    rtp_relay_enabled(false)
 {
+  for (unsigned int i=0; i<MAX_RELAY_STREAMS; i++)
+    other_stream_fds[i] = 0;
 }
 
-
 AmB2BSession::~AmB2BSession()
 {
+  if (rtp_relay_enabled)
+    clearRtpReceiverRelay();
+
   DBG("relayed_req.size() = %u\n",(unsigned int)relayed_req.size());
   DBG("recvd_req.size() = %u\n",(unsigned int)recvd_req.size());
 }
@@ -193,18 +202,112 @@ void AmB2BSession::onSipRequest(const AmSipRequest& req)
 
   B2BSipRequestEvent* r_ev = new B2BSipRequestEvent(req,fwd);
   
+  AmSdp filter_sdp;
+
   // filter relayed INVITE/UPDATE body
   if (fwd && b2b_mode != B2BMode_Transparent &&
       (req.method == SIP_METH_INVITE || req.method == SIP_METH_UPDATE ||
        req.method == SIP_METH_ACK)) {
     DBG("filtering body for request '%s' (c/t '%s')\n",
 	req.method.c_str(), req.content_type.c_str());
-    filterBody(r_ev->req.content_type, r_ev->req.body, a_leg);
+    // todo: handle filtering errors
+    filterBody(r_ev->req.content_type, r_ev->req.body, filter_sdp, a_leg);
+  }
+
+  if (rtp_relay_enabled &&
+      (req.method == SIP_METH_INVITE || req.method == SIP_METH_UPDATE ||
+       req.method == SIP_METH_ACK || req.method == SIP_METH_ACK)) {
+    updateRelayStreams(r_ev->req.content_type, r_ev->req.body, filter_sdp);
   }
 
   relayEvent(r_ev);
 }
 
+/** update RTP relay streams with address/port from SDP body */
+void AmB2BSession::updateRelayStreams(const string& content_type, const string& body,
+				      AmSdp& parser_sdp) {
+  if (body.empty() || (content_type != SIP_APPLICATION_SDP))
+    return;
+
+  if (parser_sdp.media.empty()) {
+    // SDP has not yet been parsed
+    parser_sdp.setBody(body.c_str());
+    if (parser_sdp.parse()) {
+      DBG("SDP parsing failed!\n");
+      return;
+    }
+  }
+
+  unsigned int media_index = 0;
+  for (std::vector<SdpMedia>::iterator it =
+	 parser_sdp.media.begin(); it != parser_sdp.media.end(); it++) {
+    if (media_index >= MAX_RELAY_STREAMS) {
+      ERROR("trying to relay SDP with more media lines than MAX_RELAY_STREAMS (%u)"
+	    "(consider increasing MAX_RELAY_STREAMS and rebuilding SEMS)\n",
+	    MAX_RELAY_STREAMS);
+      break;
+    }
+    string r_addr = it->conn.address;
+    if (r_addr.empty())
+      r_addr = parser_sdp.conn.address;
+
+    DBG("initializing RTP relay stream %u with remote <%s:%u>\n",
+	media_index, r_addr.c_str(), it->port);
+    relay_rtp_streams[media_index].setRAddr(r_addr, it->port);
+    media_index ++;
+  }
+}
+
+bool AmB2BSession::replaceConnectionAddress(const string& content_type,
+					    const string& body, string& r_body) {
+
+  if (body.empty() || (content_type != SIP_APPLICATION_SDP))
+    return false;
+
+  AmSdp parser_sdp;
+  parser_sdp.setBody(body.c_str());
+  if (parser_sdp.parse()) {
+    DBG("SDP parsing failed!\n");
+    return false;
+  }
+
+  // place advertisedIP() in connection address
+  if (!parser_sdp.conn.address.empty())
+    parser_sdp.conn.address = advertisedIP();
+
+  string replaced_ports;
+
+  unsigned int media_index = 0;
+  for (std::vector<SdpMedia>::iterator it =
+	 parser_sdp.media.begin(); it != parser_sdp.media.end(); it++) {
+    if (media_index >= MAX_RELAY_STREAMS) {
+      ERROR("trying to relay SDP with more media lines than MAX_RELAY_STREAMS (%u)"
+	    "(consider increasing MAX_RELAY_STREAMS and rebuilding SEMS)\n",
+	    MAX_RELAY_STREAMS);
+      break;
+    }
+    if (!it->conn.address.empty())
+      it->conn.address = advertisedIP();
+    try {
+      it->port = relay_rtp_streams[media_index].getLocalPort();
+      replaced_ports += (!media_index) ? int2str(it->port) : "/"+int2str(it->port);
+    } catch (const string& s) {
+      ERROR("setting port: '%s'\n", s.c_str());
+      return false; // TODO: or throw?
+    }
+
+    media_index++;
+  }
+
+  // regenerate SDP
+  parser_sdp.print(r_body);
+
+  DBG("replaced connection address in SDP with %s:%s.\n",
+      advertisedIP().c_str(), replaced_ports.c_str());
+
+  return true;
+}
+
 void AmB2BSession::onSipReply(const AmSipReply& reply,
 			      int old_dlg_status,
 			      const string& trans_method)
@@ -221,12 +324,21 @@ void AmB2BSession::onSipReply(const AmSipReply& reply,
     AmSipReply n_reply = reply;
     n_reply.cseq = t->second.cseq;
 
+    AmSdp filter_sdp;
+
     // filter relayed INVITE/UPDATE body
     if (b2b_mode != B2BMode_Transparent &&
 	(trans_method == SIP_METH_INVITE || trans_method == SIP_METH_UPDATE)) {
-      filterBody(n_reply.content_type, n_reply.body, a_leg);
+      filterBody(n_reply.content_type, n_reply.body, filter_sdp, a_leg);
     }
     
+    if (rtp_relay_enabled &&
+	(reply.code >= 180  && reply.code < 300) &&
+	(trans_method == SIP_METH_INVITE || trans_method == SIP_METH_UPDATE ||
+	 trans_method == SIP_METH_ACK || trans_method == SIP_METH_ACK)) {
+      updateRelayStreams(n_reply.content_type, n_reply.body, filter_sdp);
+    }
+
     relayEvent(new B2BSipReplyEvent(n_reply, true, t->second.method));
 
     if(reply.code >= 200) {
@@ -284,6 +396,10 @@ bool AmB2BSession::onOtherReply(const AmSipReply& reply)
 void AmB2BSession::terminateLeg()
 {
   setStopped();
+
+  if (rtp_relay_enabled)
+    clearRtpReceiverRelay();
+
   if ((dlg.getStatus() == AmSipDialog::Pending) 
       || (dlg.getStatus() == AmSipDialog::Connected))
     dlg.bye("", SIP_FLAGS_VERBATIM);
@@ -361,7 +477,15 @@ int AmB2BSession::sendEstablishedReInvite() {
   }
 
   DBG("sending re-INVITE with saved session description\n");
-  return dlg.reinvite("", established_content_type, established_body,
+
+  const string* body = &established_body;
+  string r_body;
+  if (rtp_relay_enabled &&
+      replaceConnectionAddress(established_content_type, *body, r_body)) {
+	body = &r_body;
+  }
+
+  return dlg.reinvite("", established_content_type, *body,
 		      SIP_FLAGS_VERBATIM);
 }
 
@@ -412,7 +536,17 @@ void AmB2BSession::relaySip(const AmSipRequest& req)
       }
     }
 
-    dlg.sendRequest(req.method, req.content_type, req.body, *hdrs, SIP_FLAGS_VERBATIM);
+    const string* body = &req.body;
+    string r_body;
+    if (rtp_relay_enabled &&
+	(req.method == SIP_METH_INVITE || req.method == SIP_METH_UPDATE ||
+	 req.method == SIP_METH_ACK || req.method == SIP_METH_ACK)) {
+      if (replaceConnectionAddress(req.content_type, *body, r_body)) {
+	body = &r_body;
+      }
+    }
+
+    dlg.sendRequest(req.method, req.content_type, *body, *hdrs, SIP_FLAGS_VERBATIM);
     // todo: relay error event back if sending fails
 
     if ((refresh_method != REFRESH_UPDATE) &&
@@ -462,9 +596,19 @@ void AmB2BSession::relaySip(const AmSipRequest& orig, const AmSipReply& reply)
     hdrs = &m_hdrs;
   }
 
+  const string* body = &reply.body;
+  string r_body;
+  if (rtp_relay_enabled &&
+      (orig.method == SIP_METH_INVITE || orig.method == SIP_METH_UPDATE ||
+       orig.method == SIP_METH_ACK || orig.method == SIP_METH_ACK)) {
+    if (replaceConnectionAddress(reply.content_type, *body, r_body)) {
+      body = &r_body;
+    }
+  }
+
   dlg.reply(orig,reply.code,reply.reason,
 	    reply.content_type,
-	    reply.body, *hdrs,SIP_FLAGS_VERBATIM);
+	    *body, *hdrs, SIP_FLAGS_VERBATIM);
 
   if ((refresh_method != REFRESH_UPDATE) &&
       (orig.method == SIP_METH_INVITE ||
@@ -475,20 +619,20 @@ void AmB2BSession::relaySip(const AmSipRequest& orig, const AmSipReply& reply)
 
 }
 
-int AmB2BSession::filterBody(string& content_type, string& body, bool is_a2b) {
+int AmB2BSession::filterBody(string& content_type, string& body, AmSdp& filter_sdp,
+			     bool is_a2b) {
   if (body.empty())
     return 0;
 
   if (content_type == SIP_APPLICATION_SDP) {
-    AmSdp f_sdp;
-    f_sdp.setBody(body.c_str());
-    int res = f_sdp.parse();
+    filter_sdp.setBody(body.c_str());
+    int res = filter_sdp.parse();
     if (0 != res) {
       DBG("SDP parsing failed!\n");
       return res;
     }
-    filterBody(f_sdp, is_a2b);
-    f_sdp.print(body);
+    filterBody(filter_sdp, is_a2b);
+    filter_sdp.print(body);
   }
 
   return 0;
@@ -499,6 +643,51 @@ int AmB2BSession::filterBody(AmSdp& sdp, bool is_a2b) {
   return 0;
 }
 
+void AmB2BSession::enableRtpRelay() {
+  DBG("enabled RTP relay mode for B2B call '%s'\n",
+      getLocalTag().c_str());
+  rtp_relay_enabled = true;
+}
+
+void AmB2BSession::disableRtpRelay() {
+  DBG("disabled RTP relay mode for B2B call '%s'\n",
+      getLocalTag().c_str());
+  rtp_relay_enabled = false;
+}
+
+void AmB2BSession::setupRelayStreams(AmB2BSession* other_session) {
+  if (!rtp_relay_enabled)
+    return;
+
+  if (NULL == other_session) {
+    ERROR("trying to setup relay for NULL b2b session\n");
+    return;
+  }
+
+  // link the other streams as our relay streams
+  for (unsigned int i=0; i<MAX_RELAY_STREAMS; i++) {
+    other_session->relay_rtp_streams[i].setRelayStream(&relay_rtp_streams[i]);
+    other_stream_fds[i] = other_session->relay_rtp_streams[i].getLocalSocket();
+    // set local IP (todo: option for other interface!)
+    relay_rtp_streams[i].setLocalIP(AmConfig::LocalIP);
+    relay_rtp_streams[i].enableRtpRelay();
+  }
+}
+
+void AmB2BSession::clearRtpReceiverRelay() {
+  for (unsigned int i=0; i<MAX_RELAY_STREAMS; i++) {
+    // clear the other call's RTP relay streams from RTP receiver
+    if (other_stream_fds[i]) {
+      AmRtpReceiver::instance()->removeStream(other_stream_fds[i]);
+      other_stream_fds[i] = 0;
+    }
+    // clear our relay streams from RTP receiver
+    if (relay_rtp_streams[i].hasLocalSocket()) {
+      AmRtpReceiver::instance()->removeStream(relay_rtp_streams[i].getLocalSocket());
+    }
+  }
+}
+
 // 
 // AmB2BCallerSession methods
 //
@@ -643,7 +832,8 @@ void AmB2BCallerSession::connectCallee(const string& remote_party,
   B2BConnectEvent* ev = new B2BConnectEvent(remote_party,remote_uri);
 
   if (b2b_mode == B2BMode_SDPFilter) {
-    filterBody(invite_req.content_type, invite_req.body, true);
+    AmSdp filter_sdp;
+    filterBody(invite_req.content_type, invite_req.body, filter_sdp, true);
   }
 
   ev->content_type = invite_req.content_type;
@@ -691,6 +881,13 @@ void AmB2BCallerSession::createCalleeSession() {
 		  "to",   callee_dlg.remote_party.c_str(),
 		  "ruri", callee_dlg.remote_uri.c_str());
 
+  try {
+    initializeRTPRelay(callee_session);
+  } catch (...) {
+    delete callee_session;
+    throw;
+  }
+
   callee_session->start();
 
   AmSessionContainer* sess_cont = AmSessionContainer::instance();
@@ -702,6 +899,20 @@ AmB2BCalleeSession* AmB2BCallerSession::newCalleeSession()
   return new AmB2BCalleeSession(this);
 }
 
+void AmB2BCallerSession::initializeRTPRelay(AmB2BCalleeSession* callee_session) {
+  if (!callee_session || !rtp_relay_enabled)
+    return;
+
+  callee_session->enableRtpRelay();
+  callee_session->setupRelayStreams(this);
+  setupRelayStreams(callee_session);
+
+  // bind caller session's relay_streams to a port
+  for (unsigned int i=0; i<MAX_RELAY_STREAMS; i++)
+    relay_rtp_streams[i].getLocalPort();
+
+}
+
 AmB2BCalleeSession::AmB2BCalleeSession(const string& other_local_tag)
   : AmB2BSession(other_local_tag)
 {
@@ -739,8 +950,17 @@ void AmB2BCalleeSession::onB2BEvent(B2BEvent* ev)
 	AmSipTransaction(SIP_METH_INVITE, co_ev->r_cseq, trans_ticket());
     }
 
+    const string* body = &co_ev->body;
+    string r_body;
+    if (rtp_relay_enabled) {
+      if (replaceConnectionAddress(co_ev->content_type, *body, r_body)) {
+	body = &r_body;
+      }
+    }
+
+
     if (dlg.sendRequest(SIP_METH_INVITE,
-			co_ev->content_type, co_ev->body,
+			co_ev->content_type, *body,
 			co_ev->hdrs, SIP_FLAGS_VERBATIM) < 0) {
 
       DBG("sending INVITE failed, relaying back 400 Bad Request\n");

core/AmB2BSession.h

@@ -32,6 +32,8 @@
 #include "AmSipDialog.h"
 #include "sip/hash.h"
 
+#define MAX_RELAY_STREAMS 3 // voice, video, rtt
+
 enum { B2BTerminateLeg,
        B2BConnectLeg,
        B2BSipRequest,
@@ -202,7 +204,8 @@ class AmB2BSession: public AmSession
   virtual bool onOtherReply(const AmSipReply& reply);
 
   /** filter body ( b2b_mode == SDPFilter */
-  virtual int filterBody(string& content_type, string& body, bool is_a2b);
+  virtual int filterBody(string& content_type, string& body,
+			 AmSdp& filter_sdp, bool is_a2b);
 
   /** filter SDP body ( b2b_mode == SDPFilter */
   virtual int filterBody(AmSdp& sdp, bool is_a2b);
@@ -212,9 +215,33 @@ class AmB2BSession: public AmSession
 
   virtual ~AmB2BSession();
 
+  /** flag to enable RTP relay mode */
+  bool rtp_relay_enabled;
+  /** RTP streams which receive from our side and are used
+      for relaying RTP from the other side */
+  AmRtpStream relay_rtp_streams[MAX_RELAY_STREAMS];
+  /** fd of the other streams' sockets (to remove from
+      RtpReceiver at end of relaying) */
+  int other_stream_fds[MAX_RELAY_STREAMS];
+  /** clear our and the other side's RTP streams from RTPReceiver */
+  void clearRtpReceiverRelay();
+  /** update remote connection in relay_streams */
+  void updateRelayStreams(const string& content_type, const string& body,
+			  AmSdp& parser_sdp);
+  /** replace connection with our address */
+  bool replaceConnectionAddress(const string& content_type, const string& body,
+				string& r_body);
+
  public:
   void set_sip_relay_only(bool r);
   B2BMode getB2BMode() const;
+
+  /** set RTP relay mode enabled */
+  void enableRtpRelay();
+  /** set RTP relay mode disabled */
+  void disableRtpRelay();
+  /** link RTP streams of other_session to our streams */
+  void setupRelayStreams(AmB2BSession* from_session);
 };
 
 class AmB2BCalleeSession;
@@ -273,6 +300,11 @@ class AmB2BCallerSession: public AmB2BSession
   AmSipRequest* getInviteReq() { return &invite_req; }
 
   void set_sip_relay_early_media_sdp(bool r);
+
+  /** initialize RTP relay mode, if rtp_relay_enabled
+      must be called *before* callee_session is started
+   */
+  void initializeRTPRelay(AmB2BCalleeSession* callee_session);
 };
 
 /** \brief Callee leg of a B2B session */

core/AmRtpStream.cpp

@@ -77,13 +77,14 @@ void AmRtpStream::setLocalIP(const string& ip)
 #endif
 }
 
-void AmRtpStream::setLocalPort()
+int AmRtpStream::hasLocalSocket() {
+  return l_sd;
+}
+
+int AmRtpStream::getLocalSocket()
 {
-  if(l_port)
-    return;
-  
-  assert(l_if >= 0);
-  assert(l_if < (int)AmConfig::Ifs.size());
+  if (l_sd)
+    return l_sd;
 
   int sd=0;
 #ifdef SUPPORT_IPV6
@@ -101,6 +102,21 @@ void AmRtpStream::setLocalPort()
     close(sd);
     throw string ("while setting socket non blocking.");
   }
+
+  l_sd = sd;
+  return l_sd;
+}
+
+void AmRtpStream::setLocalPort()
+{
+  if(l_port)
+    return;
+  
+  assert(l_if >= 0);
+  assert(l_if < (int)AmConfig::Ifs.size());
+
+  if (!getLocalSocket())
+    return;
   
   int retry = 10;
   unsigned short port = 0;
@@ -109,11 +125,11 @@ void AmRtpStream::setLocalPort()
     port = AmConfig::Ifs[l_if].getNextRtpPort();
 #ifdef SUPPORT_IPV6
     set_port_v6(&l_saddr,port);
-    if(!bind(sd,(const struct sockaddr*)&l_saddr,
+    if(!bind(l_sd,(const struct sockaddr*)&l_saddr,
 	     sizeof(struct sockaddr_storage)))
 #else	
       l_saddr.sin_port = htons(port);
-    if(!bind(sd,(const struct sockaddr*)&l_saddr,
+    if(!bind(l_sd,(const struct sockaddr*)&l_saddr,
 	     sizeof(struct sockaddr_in)))
 #endif
       {
@@ -122,24 +138,25 @@ void AmRtpStream::setLocalPort()
     else {
       DBG("bind: %s\n",strerror(errno));		
     }
-      
   }
 
+  int true_opt = 1;
   if (!retry){
     ERROR("could not find a free RTP port\n");
-    close(sd);
+    close(l_sd);
+    l_sd = 0;
     throw string("could not find a free RTP port");
   }
 
-  if(setsockopt(sd, SOL_SOCKET, SO_REUSEADDR,
+  if(setsockopt(l_sd, SOL_SOCKET, SO_REUSEADDR,
 		(void*)&true_opt, sizeof (true_opt)) == -1) {
 
     ERROR("%s\n",strerror(errno));
-    close(sd);
+    close(l_sd);
+    l_sd = 0;
     throw string ("while setting local address reusable.");
   }
-    
-  l_sd = sd;
+
   l_port = port;
   AmRtpReceiver::instance()->addStream(l_sd,this);
   DBG("local rtp port set to %i\n",l_port);
@@ -418,7 +435,9 @@ AmRtpStream::AmRtpStream(AmSession* _s, int _if)
     hold(false),
     receiving(true),
     monitor_rtp_timeout(true),
-    dtmf_sending_state(DTMF_SEND_NONE)
+    dtmf_sending_state(DTMF_SEND_NONE),
+    relay_stream(NULL),
+    relay_enabled(false)
 {
 
 #ifdef SUPPORT_IPV6
@@ -570,6 +589,14 @@ void AmRtpStream::bufferPacket(AmRtpPacket* p)
     return;
   }
 
+  if (relay_enabled) {
+    if (NULL != relay_stream) {
+      relay_stream->relay(p);
+    }
+    mem.freePacket(p);
+    return;
+  }
+
   receive_mut.lock();
   // free packet on double packet for TS received
   if(!(telephone_event_pt.get() && 
@@ -641,6 +668,11 @@ void AmRtpStream::bufferPacket(AmRtpPacket* p)
 void AmRtpStream::clearRTPTimeout(struct timeval* recv_time) {
  memcpy(&last_recv_time, recv_time, sizeof(struct timeval));
 }
+
+unsigned int AmRtpStream::bytes2samples(unsigned int) const {
+  ERROR("bytes2samples called on AmRtpStream\n");
+  return 0;
+}
  
 int AmRtpStream::nextPacket(AmRtpPacket*& p)
 {
@@ -683,6 +715,19 @@ int AmRtpStream::nextPacket(AmRtpPacket*& p)
   return 1;
 }
 
+void AmRtpStream::relay(AmRtpPacket* p) {
+  if (!l_port) // not yet initialized
+    return;
+
+  p->sequence = sequence++;
+  p->ssrc = l_ssrc;
+  p->setAddr(&r_saddr);
+
+  if(p->send(l_sd) < 0){
+    ERROR("while sending RTP packet.\n");
+  }
+}
+
 int AmRtpStream::getTelephoneEventRate()
 {
   int retval = 0;
@@ -699,6 +744,21 @@ void AmRtpStream::sendDtmf(int event, unsigned int duration_ms) {
   DBG("enqueued DTMF event %i duration %u\n", event, duration_ms);
 }
 
+void AmRtpStream::setRelayStream(AmRtpStream* stream) {
+  relay_stream = stream;
+  DBG("set relay stream [%p] for RTP instance [%p]\n",
+      stream, this);
+}
+
+void AmRtpStream::enableRtpRelay() {
+  relay_enabled = true;
+}
+
+void AmRtpStream::disableRtpRelay() {
+  relay_enabled = false;
+}
+
+
 PacketMem::PacketMem() {
   memset(used, 0, sizeof(used));
 }

core/AmRtpStream.h

@@ -135,6 +135,12 @@ protected:
   RtpEventQueue   rtp_ev_qu;
   AmMutex         receive_mut;
 
+  /** if relay_stream is initialized, received RTP is relayed there */
+  bool            relay_enabled;
+  /** pointer to relay stream.
+      NOTE: This may only be accessed in initialization
+      or by the AmRtpReceiver thread while relaying!  */
+  AmRtpStream*    relay_stream;
 
   /* get next packet in buffer */
   int nextPacket(AmRtpPacket*& p);
@@ -159,6 +165,8 @@ protected:
   unsigned int current_send_dtmf_ts;
   int send_dtmf_end_repeat;
 
+  void relay(AmRtpPacket* p);
+
 public:
 
   AmRtpPacket* newPacket();
@@ -194,6 +202,12 @@ public:
   /** Stops the stream and frees all resources. */
   virtual ~AmRtpStream();
 
+  /** returns the socket descriptor for local socket (initialized or not) */
+  int hasLocalSocket();
+
+  /** initializes and gets the socket descriptor for local socket */
+  int getLocalSocket();
+
   /**
    * This function must be called before setLocalPort, because
    * setLocalPort will bind the socket and it will be not
@@ -291,7 +305,17 @@ public:
    */
   void clearRTPTimeout(struct timeval* recv_time);
 
-  virtual unsigned int bytes2samples(unsigned int) const = 0;
+  virtual unsigned int bytes2samples(unsigned int) const;
+
+  /** set relay stream for  RTP relaying */
+  void setRelayStream(AmRtpStream* stream);
+
+  /** ensable RTP relaying through relay stream */
+  void enableRtpRelay();
+
+  /** disable RTP relaying through relay stream */
+  void disableRtpRelay();
+
 };
 
 /** \brief event fired on RTP timeout */