src/server.rs

@@ -40,7 +40,10 @@ impl Server {
             .layer(TraceLayer::new_for_http());
 
         tracing::info!("listening on {addr}");
-        let listener = TcpListener::bind(addr).await?;
+        let listener = match TcpListener::bind(addr).await {
+            Ok(listen) => listen,
+            Err(err) => return Err(err),
+        };
         axum::serve(listener, app).await
     }
 }
@@ -50,21 +53,26 @@ async fn enqueue_file(
     TypedMultipart(req): TypedMultipart<ConvertRequest>,
 ) -> (StatusCode, Json<ConvertResponse>) {
     let task_id = Uuid::new_v4();
-    let input = Path::new(&server.work_dir).join(format!("{}.in.atranscoder", task_id));
-    let output = Path::new(&server.work_dir).join(format!("{}.out.atranscoder", task_id));
+    let input = Path::new(&server.work_dir).join(format!("{}.in.atranscoder", task_id.to_string()));
+    let output =
+        Path::new(&server.work_dir).join(format!("{}.out.atranscoder", task_id.to_string()));
 
     let file = req.file;
 
     match file.contents.persist(input.clone()) {
         Ok(_) => {
-            let input_path = match input.to_str() {
-                Some(path) => path,
-                None => return error_response("Invalid input path"),
-            };
-            let output_path = match output.to_str() {
-                Some(path) => path,
-                None => return error_response("Invalid output path"),
-            };
+            let input_path = input.to_str();
+            let output_path = output.to_str();
+
+            if input_path.is_none() || output_path.is_none() {
+                return (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json::from(ConvertResponse {
+                        id: None,
+                        error: Some(String::from("Input or output paths are not correct")),
+                    }),
+                );
+            }
 
             let task = Task::new(
                 task_id,
@@ -76,8 +84,8 @@ async fn enqueue_file(
                 req.sample_rate,
                 req.channel_layout,
                 req.upload_url,
-                input_path.to_string(),
-                output_path.to_string(),
+                input_path.unwrap().to_string(),
+                output_path.unwrap().to_string(),
             );
 
             // Enqueue the task to the thread pool
@@ -91,16 +99,12 @@ async fn enqueue_file(
                 }),
             )
         }
-        Err(_) => error_response("Cannot save the file"),
-    }
-}
-
-fn error_response(msg: &str) -> (StatusCode, Json<ConvertResponse>) {
-    (
+        Err(_) => (
             StatusCode::INTERNAL_SERVER_ERROR,
             Json::from(ConvertResponse {
-            id: None,
-            error: Some(msg.to_string()),
+                id: Some(task_id.to_string()),
+                error: Some(String::from("Cannot save the file")),
             }),
-    )
+        ),
+    }
 }

src/task.rs

@@ -1,6 +1,8 @@
 use crate::transcoder::{Transcoder, TranscoderParams};
 use ffmpeg_next::channel_layout::ChannelLayout;
-use ffmpeg_next::{format, Dictionary};
+use ffmpeg_next::format::context;
+use ffmpeg_next::format::context::Input;
+use ffmpeg_next::{format, Dictionary, Error};
 use tracing::{debug, error};
 
 pub struct Task {
@@ -59,15 +61,16 @@ impl Task {
             }
         };
 
-        let octx = if let Some(codec_opts) = self.codec_opts {
-            format::output_as_with(
+        let octx: Result<context::Output, ffmpeg_next::Error>;
+        if self.codec_opts.is_some() {
+            octx = format::output_as_with(
                 &self.output_path,
                 &self.format,
-                params_to_avdictionary(&codec_opts),
-            )
+                params_to_avdictionary(&self.codec_opts.unwrap_or_default()),
+            );
         } else {
-            format::output_as(&self.output_path, &self.format)
-        };
+            octx = format::output_as(&self.output_path, &self.format);
+        }
 
         let mut octx = match octx {
             Ok(val) => val,
@@ -93,7 +96,6 @@ impl Task {
                 },
             },
         );
-
         let mut transcoder = match transcoder {
             Ok(val) => val,
             Err(err) => {
@@ -101,57 +103,29 @@ impl Task {
                 return;
             }
         };
-
         octx.set_metadata(ictx.metadata().to_owned());
-
-        if let Err(err) = octx.write_header() {
-            error!("couldn't start transcoding: {:?}", err);
-            return;
-        }
+        octx.write_header()
+            .unwrap_or_else(|err| error!("couldn't start transcoding: {:?}", err));
 
         for (stream, mut packet) in ictx.packets() {
             if stream.index() == transcoder.stream {
                 packet.rescale_ts(stream.time_base(), transcoder.in_time_base);
-
-                if let Err(err) = transcoder.send_packet_to_decoder(&packet) {
-                    error!("error sending packet to decoder: {:?}", err);
-                    return;
-                }
-
-                transcoder.receive_and_process_decoded_frames(&mut octx)
-                    .unwrap_or_else(|err| error!("failure during processing decoded frames: {:?}", err));
+                transcoder.send_packet_to_decoder(&packet);
+                transcoder.receive_and_process_decoded_frames(&mut octx);
             }
         }
 
-        if let Err(err) = transcoder.send_eof_to_decoder() {
-            error!("error sending EOF to decoder: {:?}", err);
-            return;
-        }
+        transcoder.send_eof_to_decoder();
+        transcoder.receive_and_process_decoded_frames(&mut octx);
 
-        if let Err(err) = transcoder.receive_and_process_decoded_frames(&mut octx) {
-            error!("error receiving and processing decoded frames: {:?}", err);
-            return;
-        }
+        transcoder.flush_filter();
+        transcoder.get_and_process_filtered_frames(&mut octx);
 
-        if let Err(err) = transcoder.flush_filter() {
-            error!("couldn't flush filter: {:?}", err);
-            return;
-        }
+        transcoder.send_eof_to_encoder();
+        transcoder.receive_and_process_encoded_packets(&mut octx);
 
-        transcoder.get_and_process_filtered_frames(&mut octx)
-            .unwrap_or_else(|err| error!("failure during processing filtered frames: {:?}", err));
-
-        if let Err(err) = transcoder.send_eof_to_encoder() {
-            error!("couldn't send EOF to encoder: {:?}", err);
-            return;
-        }
-
-        transcoder.receive_and_process_encoded_packets(&mut octx)
-            .unwrap_or_else(|err| error!("failure during transcoding: {:?}", err));
-
-        if let Err(err) = octx.write_trailer() {
-            error!("couldn't finish transcoding: {:?}", err);
-        }
+        octx.write_trailer()
+            .unwrap_or_else(|err| error!("couldn't finish transcoding: {:?}", err));
     }
 }
 

src/transcoder.rs

@@ -4,9 +4,11 @@ use std::error::Error;
 use std::sync::Arc;
 
 use ffmpeg::{codec, filter, format, frame, media};
-use ffmpeg_next::error::EAGAIN;
+use ffmpeg_next::codec::Audio;
+use ffmpeg_next::Codec;
 
 pub struct Transcoder {
+    params: Arc<TranscoderParams>,
     pub(crate) stream: usize,
     filter: filter::Graph,
     decoder: codec::decoder::Audio,
@@ -32,18 +34,27 @@ impl Transcoder {
         let input = ictx
             .streams()
             .best(media::Type::Audio)
-            .ok_or("could not find best audio stream")?;
-
+            .expect("could not find best audio stream");
         let context = codec::context::Context::from_parameters(input.parameters())?;
-        let mut decoder = context.decoder().audio().map_err(|err| {
-            format!("couldn't find decoder for input file: {}", err)
-        })?;
-
-        let codec = ffmpeg::encoder::find_by_name(&*params.codec)
-            .ok_or_else(|| format!("couldn't find codec with name: {}", params.codec))?
-            .audio()?;
-
-        let global = octx.format().flags().contains(format::flag::Flags::GLOBAL_HEADER);
+        let mut decoder = match context.decoder().audio() {
+            Ok(val) => val,
+            Err(err) => {
+                return Err(
+                    format!("couldn't find decoder for input file: {}", err.to_string()).into(),
+                )
+            }
+        };
+        let codec = match ffmpeg::encoder::find_by_name(&*params.codec) {
+            None => return Err(format!("couldn't find codec with name: {}", params.codec).into()),
+            Some(val) => match val.audio() {
+                Ok(val) => val,
+                Err(err) => return Err(err.into()),
+            },
+        };
+        let global = octx
+            .format()
+            .flags()
+            .contains(format::flag::Flags::GLOBAL_HEADER);
 
         decoder.set_parameters(input.parameters())?;
 
@@ -63,36 +74,55 @@ impl Transcoder {
 
         encoder.set_rate(sample_rate);
         encoder.set_channel_layout(params.channel_layout);
-
         #[cfg(not(feature = "ffmpeg_7_0"))]
+        {
             encoder.set_channels(params.channel_layout.channels());
-
+        }
         encoder.set_format(
             codec
                 .formats()
-                .ok_or_else(|| format!("failed to get supported formats for codec: {}", codec.name()))?
+                .expect(
+                    format!(
+                        "failed to get supported formats for codec: {}",
+                        codec.name()
+                    )
+                    .as_str(),
+                )
                 .next()
-                .ok_or("no supported formats found for codec")?,
+                .unwrap(),
         );
 
-        encoder.set_bit_rate(if params.bit_rate > 0 { params.bit_rate } else { decoder.bit_rate() });
-        encoder.set_max_bit_rate(if params.max_bit_rate > 0 { params.max_bit_rate } else { decoder.max_bit_rate() });
+        if params.bit_rate > 0 {
+            encoder.set_bit_rate(params.bit_rate);
+        } else {
+            encoder.set_bit_rate(decoder.bit_rate());
+        }
+
+        if params.max_bit_rate > 0 {
+            encoder.set_max_bit_rate(params.bit_rate);
+        } else {
+            encoder.set_max_bit_rate(decoder.max_bit_rate());
+        }
+
         encoder.set_time_base((1, sample_rate));
         output.set_time_base((1, sample_rate));
 
-        let in_time_base = decoder.time_base();
         let encoder = encoder.open_as(codec)?;
         output.set_parameters(&encoder);
 
-        let filter = filter_graph("anull", &decoder, &encoder)?;
+        let filter = filter("anull", &decoder, &encoder)?;
+
+        let in_time_base = decoder.time_base();
+        let out_time_base = output.time_base();
 
         Ok(Transcoder {
             stream: input.index(),
+            params: Arc::new(params),
             filter,
             decoder,
             encoder,
             in_time_base,
-            out_time_base: output.time_base(),
+            out_time_base,
         })
     }
 
@@ -100,8 +130,8 @@ impl Transcoder {
         self.encoder.send_frame(frame)
     }
 
-    pub(crate) fn send_eof_to_encoder(&mut self) -> Result<(), ffmpeg::Error> {
-        self.encoder.send_eof()
+    pub(crate) fn send_eof_to_encoder(&mut self) {
+        self.encoder.send_eof().unwrap();
     }
 
     pub(crate) fn receive_and_process_encoded_packets(
@@ -113,19 +143,20 @@ impl Transcoder {
             encoded.set_stream(0);
             encoded.rescale_ts(self.in_time_base, self.out_time_base);
 
-            if let Err(err) = encoded.write_interleaved(octx) {
-                return Err(err.into());
+            match encoded.write_interleaved(octx) {
+                Err(err) => return Err(err.into()),
+                Ok(_) => (),
             }
         }
         Ok(())
     }
 
-    fn add_frame_to_filter(&mut self, frame: &ffmpeg::Frame) -> Result<(), ffmpeg::Error> {
-        self.filter.get("in").unwrap().source().add(frame)
+    fn add_frame_to_filter(&mut self, frame: &ffmpeg::Frame) {
+        self.filter.get("in").unwrap().source().add(frame).unwrap();
     }
 
-    pub(crate) fn flush_filter(&mut self) -> Result<(), ffmpeg::Error> {
-        self.filter.get("in").unwrap().source().flush()
+    pub(crate) fn flush_filter(&mut self) {
+        self.filter.get("in").unwrap().source().flush().unwrap();
     }
 
     pub(crate) fn get_and_process_filtered_frames(
@@ -134,17 +165,23 @@ impl Transcoder {
     ) -> Result<(), Box<dyn Error>> {
         let mut filtered = frame::Audio::empty();
         loop {
-            let mut ctx = self.filter.get("out").ok_or("cannot get context from filter")?;
+            let mut ctx: ffmpeg::filter::Context = match self.filter.get("out") {
+                None => return Err(Box::from("cannot get context from filter")),
+                Some(val) => val,
+            };
 
-            if let Err(err) = ctx.sink().frame(&mut filtered) {
-                if err != ffmpeg::Error::Eof {
-                    return Err(err.into());
-                }
-                return Ok(());
+            if !ctx.sink().frame(&mut filtered).is_ok() {
+                return Err(Box::from("frame is suddenly invalid, stopping..."));
             }
 
-            self.send_frame_to_encoder(&filtered)?;
-            self.receive_and_process_encoded_packets(octx)?;
+            match self.send_frame_to_encoder(&filtered) {
+                Err(err) => return Err(err.into()),
+                Ok(_) => (),
+            };
+            match self.receive_and_process_encoded_packets(octx) {
+                Err(err) => return Err(err.into()),
+                Ok(_) => (),
+            }
         }
     }
 
@@ -152,11 +189,17 @@ impl Transcoder {
         &mut self,
         packet: &ffmpeg::Packet,
     ) -> Result<(), Box<dyn Error>> {
-        self.decoder.send_packet(packet).map_err(|err| err.into())
+        match self.decoder.send_packet(packet) {
+            Err(err) => return Err(err.into()),
+            Ok(_) => Ok(()),
+        }
     }
 
     pub(crate) fn send_eof_to_decoder(&mut self) -> Result<(), Box<dyn Error>> {
-        self.decoder.send_eof().map_err(|err| err.into())
+        match self.decoder.send_eof() {
+            Err(err) => return Err(err.into()),
+            Ok(_) => Ok(()),
+        }
     }
 
     pub(crate) fn receive_and_process_decoded_frames(
@@ -167,20 +210,18 @@ impl Transcoder {
         while self.decoder.receive_frame(&mut decoded).is_ok() {
             let timestamp = decoded.timestamp();
             decoded.set_pts(timestamp);
-            self.add_frame_to_filter(&decoded)?;
+            self.add_frame_to_filter(&decoded);
 
-            if let Err(mut err) = self.get_and_process_filtered_frames(octx) {
-                let expected = ffmpeg::Error::Other { errno: EAGAIN };
-                if err.downcast_mut::<ffmpeg::error::Error>().ok_or(ffmpeg::Error::Bug) == Err(expected) {
-                    continue
-                }
+            match self.get_and_process_filtered_frames(octx) {
+                Err(err) => return Err(err.into()),
+                Ok(_) => (),
             }
         }
         Ok(())
     }
 }
 
-fn filter_graph(
+fn filter(
     spec: &str,
     decoder: &codec::decoder::Audio,
     encoder: &codec::encoder::Audio,
@@ -198,10 +239,13 @@ fn filter_graph(
     filter.add(&filter::find("abuffer").unwrap(), "in", &args)?;
     filter.add(&filter::find("abuffersink").unwrap(), "out", "")?;
 
+    {
         let mut out = filter.get("out").unwrap();
+
         out.set_sample_format(encoder.format());
         out.set_channel_layout(encoder.channel_layout());
         out.set_sample_rate(encoder.rate());
+    }
 
     filter.output("in", 0)?.input("out", 0)?.parse(spec)?;
     filter.validate()?;
@@ -209,8 +253,15 @@ fn filter_graph(
     println!("{}", filter.dump());
 
     if let Some(codec) = encoder.codec() {
-        if !codec.capabilities().contains(codec::capabilities::Capabilities::VARIABLE_FRAME_SIZE) {
-            filter.get("out").unwrap().sink().set_frame_size(encoder.frame_size());
+        if !codec
+            .capabilities()
+            .contains(codec::capabilities::Capabilities::VARIABLE_FRAME_SIZE)
+        {
+            filter
+                .get("out")
+                .unwrap()
+                .sink()
+                .set_frame_size(encoder.frame_size());
         }
     }