gd-extension-rs-ffmpeg/src/playback/decoder.rs

use std::ptr::null_mut;
use std::sync::Condvar;
use std::thread::ThreadId;
use libc::{EAGAIN, endservent};
use rsmpeg::avcodec::{AVCodecContext, AVPacket};
use rsmpeg::avutil::{av_rescale_q, AVFrame};
use rsmpeg::error::RsmpegError;
use rsmpeg::error::RsmpegError::AVError;
use rsmpeg::ffi::{AV_NOPTS_VALUE, AVERROR_EOF};
use crate::playback::DECODER_REORDER_PTS;
use crate::playback::packet::MutexPacketQueue;
use rsmpeg::ffi::{ AVRational, avcodec_flush_buffers };


pub struct Decoder {
    pkt : Box<AVPacket>,
    queue : Box<MutexPacketQueue>,
    avctx : Box<AVCodecContext>,
    pkt_serial : Box<i64>,
    finished: i64,
    packet_pending: bool,
    empty_queue_cond: Box<Condvar>,
    start_pts : i64,
    start_pts_tb : AVRational,
    next_pts : i64,
    next_pts_tb: AVRational,
    decoder_tid: Box<ThreadId>,
}

impl Decoder {
    fn new(avctx: AVCodecContext, queue: MutexPacketQueue, tid: ThreadId) -> Self {
        Decoder {
            pkt: Box::new(Default::default()),
            queue: Box::new(queue),
            avctx: Box::new(avctx),
            pkt_serial: Box::new(0),
            finished: 0,
            packet_pending: false,
            empty_queue_cond: Box::new(Default::default()),
            start_pts: 0,
            start_pts_tb: AVRational {num: 1, den : 1},
            next_pts: 0,
            next_pts_tb: AVRational {num : 1, den : 1},
            decoder_tid: Box::new(tid),
        }
    }

    pub fn decode_frame(&mut self, av_frame: &mut AVFrame, sub: &mut rsmpeg::ffi::AVSubtitle) ->  Result<(),RsmpegError> { // TODO Recheck
        let mut ret = Err(AVError(EAGAIN));

        loop {
            if self.queue.data.serial == *self.pkt_serial {
                loop {
                    if self.queue.data.abort_request {
                        return Err(AVError(-1));
                    }

                    let codec_type = self.avctx.extract_codecpar().codec_type();
                    ret = if codec_type.is_video() {
                        let ret_receive = self.avctx.receive_frame();
                        if let Ok(mut frame) = ret_receive {
                            if ! DECODER_REORDER_PTS {
                                frame.set_pts(frame.best_effort_timestamp);
                            } else {
                                frame.set_pts(frame.pkt_dts);
                            }
                            *av_frame =  frame;
                            Ok(())
                        } else {
                            Err(ret_receive.unwrap_err())
                        }
                    }  else if codec_type.is_audio() {
                        let ret_receive = self.avctx.receive_frame();
                        if let Ok(mut frame) = ret_receive {
                            let tb = AVRational {num: 1, den: frame.sample_rate};
                            if frame.pts != AV_NOPTS_VALUE {
                                frame.set_pts(av_rescale_q(frame.pts, self.avctx.pkt_timebase, tb))
                            } else if self.next_pts != AV_NOPTS_VALUE  {
                                frame.set_pts(av_rescale_q(self.next_pts, self.next_pts_tb, tb));
                            }
                            if frame.pts != AV_NOPTS_VALUE {
                                self.next_pts = frame.pts + frame.nb_samples as i64;
                                self.next_pts_tb = tb;
                            }
                            *av_frame = frame;
                            Ok(())
                        } else {
                            Err(ret_receive.unwrap_err())
                        }
                    } else {
                        ret
                    };

                    if Some(&AVError(AVERROR_EOF)) == ret.as_ref().err(){
                        self.finished = *self.pkt_serial;
                        unsafe { avcodec_flush_buffers(self.avctx.as_mut_ptr())};
                        return Ok(()); // TODO
                    }
                    if let Some(&AVError(x)) = ret.as_ref().err()  {
                        if x >= 0 {
                            return Err(AVError(1))
                        }
                    }
                    if Some(&AVError(EAGAIN)) == ret.as_ref().err() {
                        break;
                    }
                }
            }

            loop {
                {
                    let mut pq = self.queue.mutex.lock().unwrap();
                    while self.queue.data.queue.len() == 0 {
                        pq = self.empty_queue_cond.wait(pq).unwrap();
                    }
                }

                if ! self.packet_pending {
                    let old_serial = *self.pkt_serial;
                    // if packet_queue_get(&d.queue, &mut d.pkt, true, Some(&mut d.pkt_serial)) < 0 {// TODO
                    if let Some(AVError(x)) = &mut self.queue.get( &mut self.pkt, true, Some(&mut self.pkt_serial)).err(){// TODO
                        if (*x as i32) < 0 {
                            return Err(AVError(-1));
                        }
                    }
                    if old_serial != *self.pkt_serial {
                        unsafe { avcodec_flush_buffers(self.avctx.as_mut_ptr())};
                        self.finished = 0;
                        self.next_pts = self.start_pts;
                        self.next_pts_tb = self.start_pts_tb;
                    }
                }


                {
                    if self.queue.data.serial == *self.pkt_serial {
                        break;
                    }
                }
                //
                // TODO av_packet_unref(d->pkt);
            }

            if self.avctx.extract_codecpar().codec_type().is_subtitle() {
                let got_frame = 0;

                let mut ret = self.avctx.decode_subtitle(Some(&mut *self.pkt)); // TODO no return ?
                if ret.is_err() {
                    ret = Err(AVError(EAGAIN));
                } else {

                    if ret.as_ref().unwrap().is_some() && self.pkt.data != null_mut() {
                        // *sub = *ret.unwrap().unwrap(); FIXME
                        self.packet_pending = true;
                    }
                    ret = if ret.as_ref().unwrap().is_some() { Ok(None) } else {
                        if self.pkt.data != null_mut() {
                            Err(AVError(EAGAIN))
                        } else {
                            Err(AVError(AVERROR_EOF))
                        }
                    };
                    //av_packet_unref(d->pkt);
                }
            } else {
                if self.avctx.send_packet(Some(&self.pkt)).err() == Some(AVError(EAGAIN)) {
                    //av_log(&d.avctx, AV_LOG_ERROR,
                    //       "Receive_frame and send_packet both returned EAGAIN, which is an API violation.\n");
                    self.packet_pending = true;
                } else {
                    //av_packet_unref(d->pkt);
                }
            }
        }
    }


}