Move enqueue call and fixup buffer test

Author: ForeverASilver

src/internal_modules/roc_pipeline/config.cpp

@@ -96,7 +96,7 @@ bool ReceiverSessionConfig::deduce_defaults(audio::ProcessorMap& processor_map)
         return false;
     }
 
-    if(!prebuf_len) {
+    if (!prebuf_len) {
         prebuf_len = latency.target_latency;
     }
 

src/internal_modules/roc_pipeline/receiver_session_group.cpp

@@ -331,14 +331,16 @@ ReceiverSessionGroup::route_transport_packet_(const packet::PacketPtr& packet) {
     }
 
     if (sess) {
-        enqueue_prebuf_packet_(packet);
         // Session found, route packet to it.
         return sess->route_packet(packet);
     }
 
     // Session not found, auto-create session if possible.
     if (can_create_session_(packet)) {
         return create_session_(packet);
+    } else {
+        // else, put packet in pre session buffer
+        enqueue_prebuf_packet_(packet);
     }
 
     return status::StatusNoRoute;
@@ -508,7 +510,7 @@ void ReceiverSessionGroup::dequeue_prebuf_packet_(ReceiverSession& sess) {
     for (curr = prebuf_packets_.front(); curr; curr = next) {
         next = prebuf_packets_.nextof(*curr);
 
-        // if packet is too old, remote it from the queue
+        // if packet is too old, remove it from the queue
         core::nanoseconds_t received = curr->udp()->receive_timestamp;
         if (now - received > source_config_.session_defaults.prebuf_len) {
             prebuf_packets_.remove(*curr);

src/tests/roc_pipeline/test_receiver_source.cpp

@@ -6,6 +6,8 @@
  * file, You can obtain one at http://mozilla.org/MPL/2.0/.
  */
 
+#include "roc_core/log.h"
+#include "roc_fec/block_writer.h"
 #include "test_harness.h"
 #include "test_helpers/control_reader.h"
 #include "test_helpers/control_writer.h"
@@ -20,6 +22,7 @@
 #include "roc_pipeline/receiver_source.h"
 #include "roc_rtp/encoding_map.h"
 #include "roc_stat/mov_aggregate.h"
+#include <CppUTest/UtestMacros.h>
 
 // This file contains tests for ReceiverSource. ReceiverSource can be seen as a big
 // composite processor (consisting of chained smaller processors) that transforms
@@ -3400,64 +3403,37 @@ TEST(receiver_source, timestamp_mapping_remixing) {
 }
 
 TEST(receiver_source, packet_buffer) {
-    enum {
-        OutputRate = 48000,
-        PacketRate = 44100,
-        OutputChans = Chans_Stereo,
-        PacketChans = Chans_Mono
-    };
-
-    const audio::PcmSubformat OutputFormat = Format_S16_Ne;
-    const audio::PcmSubformat PacketFormat = Format_S16_Be;
-
-    init_with_specs(OutputRate, OutputChans, OutputFormat, PacketRate, PacketChans,
-                    PacketFormat);
+    init_with_defaults();
 
     ReceiverSource receiver(make_default_config(), processor_map, encoding_map,
                             packet_pool, packet_buffer_pool, frame_pool,
                             frame_buffer_pool, arena);
     LONGS_EQUAL(status::StatusOK, receiver.init_status());
 
     ReceiverSlot* slot = create_slot(receiver);
-    CHECK(slot);
 
-    packet::FifoQueue queue;
     packet::FifoQueue source_queue;
     packet::FifoQueue repair_queue;
 
     packet::IWriter* source_endpoint_writer = create_transport_endpoint(
         slot, address::Iface_AudioSource, address::Proto_RTP_RS8M_Source, dst_addr1);
 
-    packet::IWriter* repair_endpoint_writer = create_transport_endpoint(
-        slot, address::Iface_AudioRepair, address::Proto_RS8M_Repair, dst_addr2);
-
-    fec::BlockWriterConfig fec_config;
+    packet::IWriter* repair_endpoint_writer =
+        create_control_endpoint(slot, address::Iface_AudioRepair,
+                                address::Proto_RS8M_Repair, dst_addr2, repair_queue);
 
-    test::PacketWriter packet_writer(
-        arena, *source_endpoint_writer, *repair_endpoint_writer, encoding_map,
-        packet_factory, src_id1, src_addr1, dst_addr1, dst_addr2, PayloadType_Ch2,
-        packet::FEC_ReedSolomon_M8, fec_config);
+    test::PacketWriter packet_writer(arena, *source_endpoint_writer,
+                                     *repair_endpoint_writer, encoding_map,
+                                     packet_factory, src_id1, src_addr1, dst_addr1,
+                                     dst_addr2, PayloadType_Ch2, fec_scheme, fec_config);
 
     // setup reader
     test::FrameReader frame_reader(receiver, frame_factory);
 
-    packet_writer.write_packets(fec_config.n_source_packets, SamplesPerPacket,
-                                output_sample_spec);
-
-    for (int i = 0; i < ManyPackets; ++i) {
-        packet::PacketPtr pp;
-        LONGS_EQUAL(status::StatusOK, queue.read(pp, packet::ModeFetch));
-        CHECK(pp);
-
-        if (pp->flags() & packet::Packet::FlagAudio) {
-            UNSIGNED_LONGS_EQUAL(status::StatusOK, source_queue.write(pp));
-        }
-        if (pp->flags() & packet::Packet::FlagRepair) {
-            UNSIGNED_LONGS_EQUAL(status::StatusOK, repair_queue.write(pp));
-        }
-    }
+    packet_writer.write_packets(Latency / SamplesPerPacket, SamplesPerPacket,
+                                packet_sample_spec);
 
-    LONGS_EQUAL(status::StatusOK, receiver.refresh(frame_reader.refresh_ts(), NULL));
+    refresh_source(receiver, frame_reader.refresh_ts());
     frame_reader.read_nonzero_samples(SamplesPerFrame, output_sample_spec);
     UNSIGNED_LONGS_EQUAL(1, receiver.num_sessions());
 }

src/tools/roc_recv/main.cpp

@@ -8,7 +8,6 @@
 
 #include "roc_address/io_uri.h"
 #include "roc_address/network_uri.h"
-#include "roc_address/protocol_map.h"
 #include "roc_core/crash_handler.h"
 #include "roc_core/heap_arena.h"
 #include "roc_core/log.h"
@@ -314,9 +313,8 @@ bool build_receiver_config(const gengetopt_args_info& args,
     }
 
     if (args.prebuf_len_given) {
-        if (!core::parse_duration(
-                args.prebuf_len_arg, 
-                receiver_config.session_defaults.prebuf_len)) {
+        if (!core::parse_duration(args.prebuf_len_arg,
+                                  receiver_config.session_defaults.prebuf_len)) {
             roc_log(LogError, "invalid --prebuf-len: bad format");
             return false;
         }