Refactor output types

Author: wkozyra95

compositor_api/src/output/whip.rs

@@ -9,7 +9,7 @@ use crate::*;
 #[serde(deny_unknown_fields)]
 pub struct WhipOutput {
     /// WHIP server endpoint
-    pub endpoint_url: String,
+    pub endpoint_url: Arc<str>,
     // Bearer token
     pub bearer_token: Option<Arc<str>>,
     /// Video track configuration.

compositor_pipeline/src/error.rs

@@ -6,7 +6,7 @@ use compositor_render::{
     InputId, OutputId,
 };
 
-use crate::pipeline::{decoder::AacDecoderError, output::whip, VideoCodec};
+use crate::pipeline::{decoder::AacDecoderError, output::whip, AudioCodec, VideoCodec};
 use fdk_aac_sys as fdk;
 
 #[derive(Debug, thiserror::Error)]
@@ -42,9 +42,6 @@ pub enum RegisterOutputError {
     #[error("Failed to register output stream. Stream \"{0}\" is already registered.")]
     AlreadyRegistered(OutputId),
 
-    #[error("Encoder error while registering output stream for stream \"{0}\".")]
-    EncoderError(OutputId, #[source] EncoderInitError),
-
     #[error("Output initialization error while registering output for stream \"{0}\".")]
     OutputError(OutputId, #[source] OutputInitError),
 
@@ -85,9 +82,15 @@ pub enum UnregisterOutputError {
 
 #[derive(Debug, thiserror::Error)]
 pub enum OutputInitError {
-    #[error("An unsupported codec was requested: {0:?}.")]
+    #[error("Failed to initialize encoder.")]
+    EncoderError(#[from] EncoderInitError),
+
+    #[error("An unsupported video codec was requested: {0:?}.")]
     UnsupportedVideoCodec(VideoCodec),
 
+    #[error("An unsupported audio codec was requested: {0:?}.")]
+    UnsupportedAudioCodec(AudioCodec),
+
     #[error(transparent)]
     SocketError(#[from] std::io::Error),
 
@@ -204,7 +207,6 @@ impl From<&RegisterInputError> for PipelineErrorInfo {
 }
 
 const OUTPUT_STREAM_ALREADY_REGISTERED: &str = "OUTPUT_STREAM_ALREADY_REGISTERED";
-const ENCODER_ERROR: &str = "OUTPUT_STREAM_ENCODER_ERROR";
 const OUTPUT_ERROR: &str = "OUTPUT_STREAM_OUTPUT_ERROR";
 const UNSUPPORTED_RESOLUTION: &str = "UNSUPPORTED_RESOLUTION";
 const NO_VIDEO_OR_AUDIO_FOR_OUTPUT: &str = "NO_VIDEO_OR_AUDIO_FOR_OUTPUT";
@@ -216,15 +218,9 @@ impl From<&RegisterOutputError> for PipelineErrorInfo {
             RegisterOutputError::AlreadyRegistered(_) => {
                 PipelineErrorInfo::new(OUTPUT_STREAM_ALREADY_REGISTERED, ErrorType::UserError)
             }
-
-            RegisterOutputError::EncoderError(_, _) => {
-                PipelineErrorInfo::new(ENCODER_ERROR, ErrorType::ServerError)
-            }
-
             RegisterOutputError::OutputError(_, _) => {
                 PipelineErrorInfo::new(OUTPUT_ERROR, ErrorType::ServerError)
             }
-
             RegisterOutputError::UnsupportedResolution(_) => {
                 PipelineErrorInfo::new(UNSUPPORTED_RESOLUTION, ErrorType::UserError)
             }

compositor_pipeline/src/pipeline.rs

@@ -23,10 +23,14 @@ use crossbeam_channel::{bounded, Receiver};
 use glyphon::fontdb;
 use input::InputInitInfo;
 use input::RawDataInputOptions;
+use output::encoded_data::EncodedDataOutput;
+use output::new_external_output;
+use output::raw_data::RawDataOutput;
 use output::EncodedDataOutputOptions;
 use output::OutputOptions;
 use output::RawDataOutputOptions;
 use pipeline_output::register_pipeline_output;
+use pipeline_output::OutputInfo;
 use tokio::runtime::Runtime;
 use tokio::sync::oneshot;
 use tracing::{error, info, trace, warn};
@@ -311,9 +315,9 @@ impl Pipeline {
         register_pipeline_output(
             pipeline,
             output_id,
-            &register_options.output_options,
             register_options.video,
             register_options.audio,
+            |ctx, output_id| new_external_output(ctx, output_id, register_options.output_options),
         )
     }
 
@@ -325,9 +329,13 @@ impl Pipeline {
         register_pipeline_output(
             pipeline,
             output_id,
-            &register_options.output_options,
             register_options.video,
             register_options.audio,
+            |ctx, output_id| {
+                let (output, result) =
+                    EncodedDataOutput::new(output_id, ctx, register_options.output_options)?;
+                Ok((Box::new(output), result))
+            },
         )
     }
 
@@ -339,9 +347,12 @@ impl Pipeline {
         register_pipeline_output(
             pipeline,
             output_id,
-            &register_options.output_options,
             register_options.video,
             register_options.audio,
+            |_ctx, _output_id| {
+                let (output, result) = RawDataOutput::new(register_options.output_options)?;
+                Ok((Box::new(output), result))
+            },
         )
     }
 
@@ -398,7 +409,13 @@ impl Pipeline {
             return Err(RequestKeyframeError::OutputNotRegistered(output_id.clone()));
         };
 
-        output.output.request_keyframe(output_id)
+        match output.output.video() {
+            Some(video) => video
+                .keyframe_request_sender
+                .send(())
+                .map_err(|_| RequestKeyframeError::KeyframesUnsupported(output_id.clone())),
+            None => Err(RequestKeyframeError::NoVideoOutput(output_id.clone())),
+        }
     }
 
     pub fn register_font(&self, font_source: fontdb::Source) {
@@ -443,17 +460,18 @@ impl Pipeline {
             }
         }
 
-        let (Some(resolution), Some(frame_format)) = (
-            output.output.resolution(),
-            output.output.output_frame_format(),
-        ) else {
+        let Some(video_output) = output.output.video() else {
             return Err(UpdateSceneError::AudioVideoNotMatching(output_id));
         };
 
         info!(?output_id, "Update scene {:#?}", scene_root);
 
-        self.renderer
-            .update_scene(output_id, resolution, frame_format, scene_root)
+        self.renderer.update_scene(
+            output_id,
+            video_output.resolution,
+            video_output.frame_format,
+            scene_root,
+        )
     }
 
     fn update_audio(
@@ -500,8 +518,15 @@ impl Pipeline {
         self.inputs.iter()
     }
 
-    pub fn outputs(&self) -> impl Iterator<Item = (&OutputId, &PipelineOutput)> {
-        self.outputs.iter()
+    pub fn outputs(&self) -> impl Iterator<Item = (&OutputId, OutputInfo)> {
+        self.outputs.iter().map(|(id, output)| {
+            (
+                id,
+                OutputInfo {
+                    kind: output.output.kind(),
+                },
+            )
+        })
     }
 }