diff --git a/apps/sbc/SBC.cpp b/apps/sbc/SBC.cpp index bc9ed616..28ceabf7 100644 --- a/apps/sbc/SBC.cpp +++ b/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(); diff --git a/apps/sbc/SBCCallProfile.cpp b/apps/sbc/SBCCallProfile.cpp index c794be3b..9478b2c4 100644 --- a/apps/sbc/SBCCallProfile.cpp +++ b/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 = ""; 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; diff --git a/apps/sbc/SBCCallProfile.h b/apps/sbc/SBCCallProfile.h index 2519214a..fc8ccee7 100644 --- a/apps/sbc/SBCCallProfile.h +++ b/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() diff --git a/core/AmB2BSession.cpp b/core/AmB2BSession.cpp index 6f51c632..d637c9e6 100644 --- a/core/AmB2BSession.cpp +++ b/core/AmB2BSession.cpp @@ -30,6 +30,7 @@ #include "ampi/MonitoringAPI.h" #include "AmSipHeaders.h" #include "AmUtils.h" +#include "AmRtpReceiver.h" #include @@ -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; ireq.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::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::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; irelay_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; iremoveStream(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; ir_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"); diff --git a/core/AmB2BSession.h b/core/AmB2BSession.h index dfbb242b..d5659bd0 100644 --- a/core/AmB2BSession.h +++ b/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 */ diff --git a/core/AmRtpStream.cpp b/core/AmRtpStream.cpp index 155b4800..61341205 100644 --- a/core/AmRtpStream.cpp +++ b/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)); } diff --git a/core/AmRtpStream.h b/core/AmRtpStream.h index c9829347..a3dcb3b2 100644 --- a/core/AmRtpStream.h +++ b/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 */