From 3b71f9babdd924e72ff5662ee89dfe5dd45a65d0 Mon Sep 17 00:00:00 2001
From: Muaz Ahmad <mahmad2000@protonmail.com>
Date: Wed, 11 Oct 2023 13:02:00 +0500
Subject: [PATCH] eof handling for demux -> decode, break channel to signal eof
 without returning error

---
 src/decode/codecs/audio.rs |  4 ++++
 src/decode/codecs/mod.rs   |  4 +++-
 src/decode/codecs/video.rs |  5 +++++
 src/demux/input.rs         | 17 ++++++++++++++++-
 4 files changed, 28 insertions(+), 2 deletions(-)

diff --git a/src/decode/codecs/audio.rs b/src/decode/codecs/audio.rs
index 09a270d..375f1a3 100644
--- a/src/decode/codecs/audio.rs
+++ b/src/decode/codecs/audio.rs
@@ -13,6 +13,10 @@ pub struct AACDecoder {
 }
 
 impl Decoder for Arc<Mutex<AACDecoder>> {
+    fn close_clean(&self) {
+        let mut self_ptr = self.lock().unwrap();
+        drop(self_ptr.cmd.stdin.take().unwrap());
+    }
     fn write_nalu(&self, nalu: util::NALUPacket) -> Result<(), Box<dyn Error + Send + Sync>> {
         let mut self_ptr = self.lock().unwrap();
         let mut pipe = match &self_ptr.cmd.stdin {
diff --git a/src/decode/codecs/mod.rs b/src/decode/codecs/mod.rs
index a1b2ad7..a081da3 100644
--- a/src/decode/codecs/mod.rs
+++ b/src/decode/codecs/mod.rs
@@ -3,16 +3,18 @@ pub mod audio;
 
 use std::sync::mpsc;
 use std::error::Error;
+use std::thread;
 
 use crate::util;
 
 pub trait Decoder {
+    fn close_clean(&self);
     fn write_nalu(&self, nalu: util::NALUPacket) -> Result<(), Box<dyn Error + Send + Sync>>;
     fn write_loop(&self, chan_in: mpsc::Receiver<util::NALUPacket>, err_in: mpsc::Sender<Box<dyn Error + Send + Sync>>) {
         loop {
             let nalu = match chan_in.recv() {
                 Ok(x) => x,
-                Err(err) => {util::thread_freeze(err_in, Box::new(err)); return}
+                Err(err) => {self.close_clean(); thread::park(); return}
             };
             match self.write_nalu(nalu) {
                 Ok(_) => (),
diff --git a/src/decode/codecs/video.rs b/src/decode/codecs/video.rs
index e49118b..70ea6d3 100644
--- a/src/decode/codecs/video.rs
+++ b/src/decode/codecs/video.rs
@@ -13,6 +13,11 @@ pub struct H264Decoder {
 }
 
 impl Decoder for Arc<Mutex<H264Decoder>> {
+    fn close_clean(&self) {
+        let mut self_ptr = self.lock().unwrap();
+        drop(self_ptr.cmd.stdin.take().unwrap());
+    }
+
     fn write_nalu(&self, nalu: util::NALUPacket) -> Result<(), Box<dyn Error + Send + Sync>> {
         let mut self_ptr = self.lock().unwrap();
         let mut pipe = match &self_ptr.cmd.stdin {
diff --git a/src/demux/input.rs b/src/demux/input.rs
index 58e2640..51990b4 100644
--- a/src/demux/input.rs
+++ b/src/demux/input.rs
@@ -2,6 +2,7 @@ use std::error::Error;
 use std::io;
 use std::io::Read;
 use std::sync::{mpsc, Arc};
+use std::thread;
 
 use crate::util;
 use crate::demux::flv;
@@ -13,7 +14,21 @@ pub trait FileReader {
         loop {
             let nalu = match self.read_nalu(&metadata) {
                 Ok(x) => x,
-                Err(err) => {util::thread_freeze(err_in, err); return}
+                Err(err) => {
+                    if let Some(e) = err.downcast_ref::<util::DemuxerError>() {
+                        match e {
+                            util::DemuxerError::EOF => {
+                                drop(v_in);
+                                drop(a_in);
+                                thread::park();
+                                return
+                            },
+                            _ => ()
+                        }
+                    };
+                    util::thread_freeze(err_in, err);
+                    return
+                }
             };
             match nalu.packet_type {
                 util::NALUPacketType::Audio => {