rtp: improve reverse stream detection. #135

src/curses/ui_call_flow.c

@@ -267,8 +267,9 @@ call_flow_draw_columns(ui_t *ui)
     if (!setting_disabled(SETTING_CF_MEDIA)) {
         while ((call = call_group_get_next(info->group, call)) ) {
             streams = vector_iterator(call->streams);
+
             while ((stream = vector_iterator_next(&streams))) {
-                if (stream_get_count(stream)) {
+                if (stream->type == PACKET_RTP && stream_get_count(stream)) {
                     addr = stream->src;
                     addr.port = 0;
                     call_flow_column_add(ui, NULL, addr);

src/rtp.c

@@ -201,12 +201,43 @@ rtp_check_packet(packet_t *packet)
         if (!(stream_is_complete(stream))) {
             stream_complete(stream, src);
             stream_set_format(stream, format);
+
+            /**
+             * TODO This is a mess. Rework required
+             *
+             * This logic tries to handle a common problem when SDP address and RTP address
+             * doesn't match. In some cases one endpoint waits until RTP data is sent to its
+             * configured port in SDP and replies its RTP to the source ignoring what the other
+             * endpoint has configured in its SDP.
+             *
+             * For such cases, we create streams 'on the fly', when a stream is completed (the
+             * first time its source address is filled), a new stream is created with the
+             * opposite src and dst.
+             *
+             * BUT, there are some cases when this 'reverse' stream should not be created:
+             *  - When there already exists a stream with that setup
+             *  - When there exists an incomplete stream with that destination (and still no source)
+             *  - ...
+             *
+             */
+
             // Check if an stream in the opposite direction exists
             if (!(reverse = rtp_find_call_stream(stream->media->msg->call, stream->dst, stream->src))) {
                 reverse = stream_create(stream->media, stream->src, PACKET_RTP);
                 stream_complete(reverse, stream->dst);
                 stream_set_format(reverse, format);
                 call_add_stream(msg_get_call(stream->media->msg), reverse);
+            } else {
+                // If the reverse stream has other source configured
+                if (reverse->src.port && !addressport_equals(stream->src, reverse->src)) {
+                    if (!(reverse = rtp_find_call_exact_stream(stream->media->msg->call, stream->dst, stream->src))) {
+                        // Create a new reverse stream
+                        reverse = stream_create(stream->media, stream->src, PACKET_RTP);
+                        stream_complete(reverse, stream->dst);
+                        stream_set_format(reverse, format);
+                        call_add_stream(msg_get_call(stream->media->msg), reverse);
+                    }
+                }
             }
         }
 
@@ -390,14 +421,29 @@ rtp_find_call_stream(struct sip_call *call, address_t src, address_t dst)
         }
     }
 
-    // Try to look for an incomplete stream with this destination
+    // Try to look for a complete stream with this destination
     if (src.port) {
-        vector_iterator_set_last(&it);
-        while ((stream = vector_iterator_prev(&it))) {
-            if (addressport_equals(src, stream->src) &&
-                addressport_equals(dst, stream->dst)) {
-                return stream;
-            }
+        return rtp_find_call_exact_stream(call, src, dst);
+    }
+
+    // Nothing found
+    return NULL;
+}
+
+rtp_stream_t *
+rtp_find_call_exact_stream(struct sip_call *call, address_t src, address_t dst)
+{
+    rtp_stream_t *stream;
+    vector_iter_t it;
+
+    // Create an iterator for call streams
+    it = vector_iterator(call->streams);
+
+    vector_iterator_set_last(&it);
+    while ((stream = vector_iterator_prev(&it))) {
+        if (addressport_equals(src, stream->src) &&
+            addressport_equals(dst, stream->dst)) {
+            return stream;
         }
     }
 

src/rtp.h

@@ -304,6 +304,9 @@ rtp_find_stream(address_t src, address_t dst);
 rtp_stream_t *
 rtp_find_call_stream(struct sip_call *call, address_t src, address_t dst);
 
+rtp_stream_t *
+rtp_find_call_exact_stream(struct sip_call *call, address_t src, address_t dst);
+
 /**
  * @brief Check if a message is older than other
  *

src/sip.c

@@ -585,7 +585,7 @@ sip_parse_msg_media(sip_msg_t *msg, const u_char *payload)
     }
 
     address_t dst, src = { };
-    rtp_stream_t *rtp_stream = NULL, *rtcp_stream = NULL;
+    rtp_stream_t *rtp_stream = NULL, *rtcp_stream = NULL, *msg_rtp_stream = NULL;
     char media_type[MEDIATYPELEN] = { };
     char media_format[30] = { };
     uint32_t media_fmt_pref;
@@ -609,6 +609,7 @@ sip_parse_msg_media(sip_msg_t *msg, const u_char *payload)
             if (sscanf(line, "m=%s %hu RTP/%*s %u", media_type, &dst.port, &media_fmt_pref) == 3) {
 
                 // Add streams from previous 'm=' line to the call
+                ADD_STREAM(msg_rtp_stream);
                 ADD_STREAM(rtp_stream);
                 ADD_STREAM(rtcp_stream);
 
@@ -626,6 +627,11 @@ sip_parse_msg_media(sip_msg_t *msg, const u_char *payload)
                      */
                     // Create a new stream with this destination address:port
 
+                    // Create RTP stream with source of message as destination address
+                    msg_rtp_stream = stream_create(media, dst, PACKET_RTP);
+                    msg_rtp_stream->dst = msg->packet->src;
+                    msg_rtp_stream->dst.port = dst.port;
+
                     // Create RTP stream
                     rtp_stream = stream_create(media, dst, PACKET_RTP);
 
@@ -661,6 +667,7 @@ sip_parse_msg_media(sip_msg_t *msg, const u_char *payload)
     }
 
     // Add streams from last 'm=' line to the call
+    ADD_STREAM(msg_rtp_stream);
     ADD_STREAM(rtp_stream);
     ADD_STREAM(rtcp_stream);