eof handling for demux -> decode, break channel to signal eof without returning error

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 {

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(_) => (),

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 {

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 => {