drop excess frames on loopback channel

git-svn-id: http://svn.freeswitch.org/svn/freeswitch/trunk@13156 d0543943-73ff-0310-b7d9-9358b9ac24b2

diff --git a/src/mod/endpoints/mod_loopback/mod_loopback.c b/src/mod/endpoints/mod_loopback/mod_loopback.c
index d0905636da85ec8306fdd48e96dbfb7e056818b7..d8a18dd737c4c6d8ca5d744f4c1569051acf4698 100644 (file)
--- a/src/mod/endpoints/mod_loopback/mod_loopback.c
+++ b/src/mod/endpoints/mod_loopback/mod_loopback.c
@@ -187,7 +187,7 @@ static switch_status_t tech_init(private_t *tech_pvt, switch_core_session_t *ses
                switch_mutex_init(&tech_pvt->flag_mutex, SWITCH_MUTEX_NESTED, switch_core_session_get_pool(session));
                switch_mutex_init(&tech_pvt->mutex, SWITCH_MUTEX_NESTED, switch_core_session_get_pool(session));
                switch_core_session_set_private(session, tech_pvt);
-               switch_queue_create(&tech_pvt->frame_queue, 50000, switch_core_session_get_pool(session));
+               switch_queue_create(&tech_pvt->frame_queue, 3, switch_core_session_get_pool(session));
                tech_pvt->session = session;
                tech_pvt->channel = switch_core_session_get_channel(session);
        }
@@ -503,7 +503,7 @@ static switch_status_t channel_on_consume_media(switch_core_session_t *session)
 
        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "CHANNEL CONSUME_MEDIA\n");
 
-       return SWITCH_STATUS_FALSE;
+       return SWITCH_STATUS_SUCCESS;
 }
 
 static switch_status_t channel_send_dtmf(switch_core_session_t *session, const switch_dtmf_t *dtmf)
@@ -548,9 +548,7 @@ static switch_status_t channel_read_frame(switch_core_session_t *session, switch
                goto end;
        }
 
-       if (!switch_queue_size(tech_pvt->frame_queue)) {
-               switch_core_timer_next(&tech_pvt->timer);
-       }
+       switch_core_timer_next(&tech_pvt->timer);
        
        if (switch_queue_trypop(tech_pvt->frame_queue, &pop) == SWITCH_STATUS_SUCCESS && pop) {
                if (tech_pvt->write_frame) {
@@ -669,8 +667,11 @@ static switch_status_t channel_write_frame(switch_core_session_t *session, switc
                if (switch_frame_dup(frame, &clone) != SWITCH_STATUS_SUCCESS) {
                        abort();
                }
+
+               if (switch_queue_trypush(tech_pvt->other_tech_pvt->frame_queue, clone) != SWITCH_STATUS_SUCCESS) {
+                       switch_frame_free(&clone);
+               }
                
-               switch_queue_push(tech_pvt->other_tech_pvt->frame_queue, clone);
                switch_set_flag_locked(tech_pvt->other_tech_pvt, TFLAG_WRITE);
                status = SWITCH_STATUS_SUCCESS;
        }
@@ -727,10 +728,10 @@ static switch_status_t channel_receive_message(switch_core_session_t *session, s
                                switch_frame_t *frame = (switch_frame_t *) pop;
                                switch_frame_free(&frame);
                        }
-                       /* avoid endless loop by not forwarding it when it came from this same place */
-                       if (tech_pvt->other_session && strcmp(msg->_file, __FILE__)) {
-                               /* pass message over to the other leg */
-                               switch_core_session_receive_message(tech_pvt->other_session, msg);
+
+                       while (switch_queue_trypop(tech_pvt->other_tech_pvt->frame_queue, &pop) == SWITCH_STATUS_SUCCESS && pop) {
+                               switch_frame_t *frame = (switch_frame_t *) pop;
+                               switch_frame_free(&frame);
                        }
                }
                break;

diff --git a/src/switch_rtp.c b/src/switch_rtp.c
index 070062ef100c8020c531b1f9ef5bc9f1f8692ca6..1a1693049a17080d9d9f8cea11c667d836d89320 100644 (file)
--- a/src/switch_rtp.c
+++ b/src/switch_rtp.c
@@ -31,6 +31,7 @@
  *
  */
 //#define DEBUG_2833
+#define RTP_DEBUG_WRITE_DELTA
 #include <switch.h>
 #include <switch_stun.h>
 #undef PACKAGE_NAME
@@ -203,6 +204,11 @@ struct switch_rtp {
        uint32_t cng_count;
        switch_rtp_bug_flag_t rtp_bugs;
        switch_rtp_stats_t stats;
+
+#ifdef RTP_DEBUG_WRITE_DELTA
+       switch_time_t send_time;
+#endif
+
 };
 
 static int global_init = 0;
@@ -2229,6 +2235,18 @@ static int rtp_common_write(switch_rtp_t *rtp_session,
                        bytes = sbytes;
                }
 
+#undef RTP_DEBUG_WRITE_DELTA
+#ifdef RTP_DEBUG_WRITE_DELTA
+               {
+                       switch_time_t now = switch_time_now();
+                       int delta = (int) (now - rtp_session->send_time) / 1000;
+                       //assert(delta);
+                       //printf("WRITE %d delta %d\n", (int)bytes, delta);
+                       rtp_session->send_time = now;
+               }
+#endif
+
+
                if (switch_socket_sendto(rtp_session->sock_output, rtp_session->remote_addr, 0, (void *) send_msg, &bytes) != SWITCH_STATUS_SUCCESS) {
                        rtp_session->seq--;
                        ret = -1;