[spo] roc-streaminggh-731: Add tests for new pipeline features

Sponsored-by: waspd

src/internal_modules/roc_audio/fanout.cpp

@@ -17,8 +17,7 @@ namespace audio {
 Fanout::Fanout(const SampleSpec& sample_spec)
     : sample_spec_(sample_spec)
     , init_status_(status::NoStatus) {
-    roc_panic_if_msg(!sample_spec_.is_valid() || !sample_spec_.is_raw(),
-                     "fanout: required valid sample spec with raw format: %s",
+    roc_panic_if_msg(!sample_spec_.is_valid(), "fanout: required valid sample spec: %s",
                      sample_spec_to_str(sample_spec_).c_str());
 
     init_status_ = status::StatusOK;

src/internal_modules/roc_pipeline/receiver_session.cpp

@@ -233,14 +233,20 @@ ReceiverSession::ReceiverSession(const ReceiverSessionConfig& session_config,
         frm_reader = resampler_reader_.get();
     }
 
-    latency_monitor_.reset(new (latency_monitor_) audio::LatencyMonitor(
-        *frm_reader, *source_queue_, *depacketizer_, *source_meter_, fec_reader_.get(),
-        resampler_reader_.get(), session_config.latency, pkt_encoding->sample_spec,
-        common_config.output_sample_spec));
-    if ((init_status_ = latency_monitor_->init_status()) != status::StatusOK) {
-        return;
+    {
+        const audio::SampleSpec inout_spec(
+            common_config.output_sample_spec.sample_rate(), audio::Sample_RawFormat,
+            common_config.output_sample_spec.channel_set());
+
+        latency_monitor_.reset(new (latency_monitor_) audio::LatencyMonitor(
+            *frm_reader, *source_queue_, *depacketizer_, *source_meter_,
+            fec_reader_.get(), resampler_reader_.get(), session_config.latency,
+            pkt_encoding->sample_spec, inout_spec));
+        if ((init_status_ = latency_monitor_->init_status()) != status::StatusOK) {
+            return;
+        }
+        frm_reader = latency_monitor_.get();
     }
-    frm_reader = latency_monitor_.get();
 
     // Top-level frame reader that is added to mixer.
     frame_reader_ = frm_reader;
@@ -398,7 +404,8 @@ status::StatusCode ReceiverSession::read(audio::Frame& frame,
     // Return StatusEnd to be excluded from mixing.
     // We don't return error from read() because we don't want the whole
     // receiver to fail, we just need to remove one session.
-    if (code != status::StatusOK && code != status::StatusEnd) {
+    if (code != status::StatusOK && code != status::StatusPart
+        && code != status::StatusDrain) {
         fail_status_ = code;
         return status::StatusEnd;
     }

src/internal_modules/roc_pipeline/receiver_source.cpp

@@ -32,12 +32,18 @@ ReceiverSource::ReceiverSource(const ReceiverSourceConfig& source_config,
 
     audio::IFrameReader* frm_reader = NULL;
 
-    mixer_.reset(new (mixer_) audio::Mixer(
-        frame_factory_, arena, source_config.common.output_sample_spec, true));
-    if ((init_status_ = mixer_->init_status()) != status::StatusOK) {
-        return;
+    {
+        const audio::SampleSpec inout_spec(
+            source_config_.common.output_sample_spec.sample_rate(),
+            audio::Sample_RawFormat,
+            source_config_.common.output_sample_spec.channel_set());
+
+        mixer_.reset(new (mixer_) audio::Mixer(frame_factory_, arena, inout_spec, true));
+        if ((init_status_ = mixer_->init_status()) != status::StatusOK) {
+            return;
+        }
+        frm_reader = mixer_.get();
     }
-    frm_reader = mixer_.get();
 
     if (!source_config_.common.output_sample_spec.is_raw()) {
         const audio::SampleSpec in_spec(

src/internal_modules/roc_pipeline/sender_session.cpp

@@ -203,13 +203,19 @@ SenderSession::create_transport_pipeline(SenderEndpoint* source_endpoint,
         frm_writer = resampler_writer_.get();
     }
 
-    feedback_monitor_.reset(new (feedback_monitor_) audio::FeedbackMonitor(
-        *frm_writer, *packetizer_, resampler_writer_.get(), sink_config_.feedback,
-        sink_config_.latency, sink_config_.input_sample_spec));
-    if ((status = feedback_monitor_->init_status()) != status::StatusOK) {
-        return status;
+    {
+        const audio::SampleSpec inout_spec(sink_config_.input_sample_spec.sample_rate(),
+                                           audio::Sample_RawFormat,
+                                           sink_config_.input_sample_spec.channel_set());
+
+        feedback_monitor_.reset(new (feedback_monitor_) audio::FeedbackMonitor(
+            *frm_writer, *packetizer_, resampler_writer_.get(), sink_config_.feedback,
+            sink_config_.latency, inout_spec));
+        if ((status = feedback_monitor_->init_status()) != status::StatusOK) {
+            return status;
+        }
+        frm_writer = feedback_monitor_.get();
     }
-    frm_writer = feedback_monitor_.get();
 
     // Top-level frame writer that is added to fanout.
     frame_writer_ = frm_writer;

src/internal_modules/roc_pipeline/sender_sink.cpp

@@ -33,11 +33,17 @@ SenderSink::SenderSink(const SenderSinkConfig& sink_config,
 
     audio::IFrameWriter* frm_writer = NULL;
 
-    fanout_.reset(new (fanout_) audio::Fanout(sink_config_.input_sample_spec));
-    if ((init_status_ = fanout_->init_status()) != status::StatusOK) {
-        return;
+    {
+        const audio::SampleSpec inout_spec(sink_config_.input_sample_spec.sample_rate(),
+                                           audio::Sample_RawFormat,
+                                           sink_config_.input_sample_spec.channel_set());
+
+        fanout_.reset(new (fanout_) audio::Fanout(inout_spec));
+        if ((init_status_ = fanout_->init_status()) != status::StatusOK) {
+            return;
+        }
+        frm_writer = fanout_.get();
     }
-    frm_writer = fanout_.get();
 
     if (!sink_config_.input_sample_spec.is_raw()) {
         const audio::SampleSpec out_spec(sink_config_.input_sample_spec.sample_rate(),