gh-217 Add packet removal loop to dequeue function

src/internal_modules/roc_pipeline/receiver_session_group.cpp

@@ -10,7 +10,6 @@
 #include "roc_address/socket_addr_to_str.h"
 #include "roc_core/log.h"
 #include "roc_core/panic.h"
-#include <iostream>
 
 namespace roc {
 namespace pipeline {
@@ -201,10 +200,14 @@ void ReceiverSessionGroup::enqueue_prebuf_packet_(const packet::PacketPtr& packe
     prebuf_packets_.push_back(*packet_ptr.get());
 
     core::nanoseconds_t now = core::timestamp(core::ClockMonotonic);
-    core::nanoseconds_t received = prebuf_packets_.front()->udp()->recieve_timestamp;
 
-    if (now - received > receiver_config_.default_session.prebuf_len) {
-        prebuf_packets_.remove(*prebuf_packets_.front());
+    while (prebuf_packets_.size() > 0) {
+        core::nanoseconds_t received = prebuf_packets_.front()->udp()->recieve_timestamp;
+        if (now - received > receiver_config_.default_session.prebuf_len) {
+            prebuf_packets_.remove(*prebuf_packets_.front());
+        } else {
+            break;
+        }
     }
 }
 
@@ -215,9 +218,19 @@ void ReceiverSessionGroup::dequeue_prebuf_packets_(ReceiverSession& sess) {
         return;
     }
 
+    core::nanoseconds_t now = core::timestamp(core::ClockMonotonic);
+
     for (curr = prebuf_packets_.front(); curr; curr = next) {
         next = prebuf_packets_.nextof(*curr);
 
+        // if packet is too old, remove it from the queue
+        core::nanoseconds_t received = curr->udp()->recieve_timestamp;
+        if (now - received > receiver_config_.default_session.prebuf_len) {
+            prebuf_packets_.remove(*curr);
+            continue;
+        }
+
+        // if session handles the packet, remove it from the queue
         if (sess.handle(curr)) {
             prebuf_packets_.remove(*curr);
         }