/*!
 * DMA-BUF streaming implementations. 
 */

// TODO: remove code duplication between both implementations. Especially concerning IO: starting and stopping and enqueueing and dequeueing buffers.

use std::convert::TryInto;
use std::os::fd::AsRawFd;
use std::time::Duration;
use std::{io, mem, sync::Arc};

use dma_buf::DmaBuf;

use crate::buffer::{Metadata, Type};
use crate::device::{Device, Handle};
use crate::io::traits::{CaptureStream, Stream as StreamTrait};
use crate::io::dmabuf::arena::Arena;
use crate::memory::Memory;
use crate::v4l2;
use crate::v4l_sys::*;

use super::arena::{self, DmaBufProtected};


/// Stream core responsible for DMA-BUF calls
struct StreamIo {
    handle: Arc<Handle>,
    buf_type: Type,
    streaming: bool,
}

impl StreamIo {
    fn buffer_desc(&self) -> v4l2_buffer {
        v4l2_buffer {
            type_: self.buf_type as u32,
            memory: Memory::UserPtr as u32,
            ..unsafe { mem::zeroed() }
        }
    }

    // TODO: deduplicate with the simple Stream implementation
    fn enqueue(&mut self, index: usize, buf: &DmaBuf) -> io::Result<()> {
        let mut v4l2_buf = v4l2_buffer {
            index: index as u32,
            memory: Memory::DmaBuf as u32,
            m: v4l2_buffer__bindgen_ty_1 {
                fd: buf.as_raw_fd(),
            },
            ..self.buffer_desc()
        };
        unsafe {
            v4l2::ioctl(
                self.handle.fd(),
                v4l2::vidioc::VIDIOC_QBUF,
                &mut v4l2_buf as *mut _ as *mut std::os::raw::c_void,
            )?;
        }

        Ok(())
    }

    // TODO: deduplicate with the simple Stream implementation
    fn dequeue(&mut self) -> io::Result<(usize, Metadata)> {
        let mut v4l2_buf = self.buffer_desc();

        if self.handle.poll(libc::POLLIN, -1)? == 0 {
            // This condition can only happen if there was a timeout.
            // A timeout is only possible if the `timeout` value is non-zero, meaning we should
            // propagate it to the caller.
            return Err(io::Error::new(io::ErrorKind::TimedOut, "VIDIOC_DQBUF"));
        }

        unsafe {
            v4l2::ioctl(
                self.handle.fd(),
                v4l2::vidioc::VIDIOC_DQBUF,
                &mut v4l2_buf as *mut _ as *mut std::os::raw::c_void,
            )?;
        }

        Ok((
            v4l2_buf.index as usize,
            Metadata {
                bytesused: v4l2_buf.bytesused,
                flags: v4l2_buf.flags.into(),
                field: v4l2_buf.field,
                timestamp: v4l2_buf.timestamp.into(),
                sequence: v4l2_buf.sequence,
            },
        ))
    }
}


// TODO: deduplicate with the simple Stream implementation
impl StreamTrait for StreamIo {
    type Item = ();

    fn start(&mut self) -> io::Result<()> {
        unsafe {
            let mut typ = self.buf_type as u32;
            v4l2::ioctl(
                self.handle.fd(),
                v4l2::vidioc::VIDIOC_STREAMON,
                &mut typ as *mut _ as *mut std::os::raw::c_void,
            )?;
        }

        self.streaming = true;
        Ok(())
    }

    fn stop(&mut self) -> io::Result<()> {
        unsafe {
            let mut typ = self.buf_type as u32;
            v4l2::ioctl(
                self.handle.fd(),
                v4l2::vidioc::VIDIOC_STREAMOFF,
                &mut typ as *mut _ as *mut std::os::raw::c_void,
            )?;
        }

        self.streaming = false;
        Ok(())
    }
}

// TODO: deduplicate with the simple Stream implementation
impl Drop for StreamIo {
    fn drop(&mut self) {
        if let Err(e) = self.stop() {
            if let Some(code) = e.raw_os_error() {
                // ENODEV means the file descriptor wrapped in the handle became invalid, most
                // likely because the device was unplugged or the connection (USB, PCI, ..)
                // broke down. Handle this case gracefully by ignoring it.
                if code == 19 {
                    /* ignore */
                    return;
                }
            }

            panic!("{:?}", e)
        }
    }
}

/// Stream of user buffers. Exposes the borrowing API
pub struct Stream {
    handle: Arc<Handle>,
    arena: Arena<DmaBuf>,
    arena_index: usize,
    buf_type: Type,
    buf_meta: Vec<Metadata>,
    timeout: Option<i32>,

    active: bool,
}

impl Stream {
    /// Returns a stream for frame capturing
    ///
    /// # Arguments
    ///
    /// * `dev` - Device ref to get its file descriptor
    /// * `buf_type` - Type of the buffers
    ///
    /// # Example
    ///
    /// ```
    /// use v4l::buffer::Type;
    /// use v4l::device::Device;
    /// use v4l::io::userptr::Stream;
    ///
    /// let dev = Device::new(0);
    /// if let Ok(dev) = dev {
    ///     let stream = Stream::new(&dev, Type::VideoCapture);
    /// }
    /// ```
    pub fn new(dev: &Device, buf_type: Type) -> io::Result<Self> {
        Stream::with_buffers(dev, buf_type, 4)
    }

    pub fn with_buffers(dev: &Device, buf_type: Type, buf_count: u32) -> io::Result<Self> {
        let mut arena = Arena::new(dev.handle(), buf_type);
        let count = arena.allocate(buf_count)?;
        let mut buf_meta = Vec::new();
        buf_meta.resize(count as usize, Metadata::default());

        Ok(Stream {
            handle: dev.handle(),
            arena,
            arena_index: 0,
            buf_type,
            buf_meta,
            active: false,
            timeout: None,
        })
    }

    /// Returns the raw device handle
    pub fn handle(&self) -> Arc<Handle> {
        self.handle.clone()
    }

    /// Sets a timeout of the v4l file handle.
    pub fn set_timeout(&mut self, duration: Duration) {
        self.timeout = Some(duration.as_millis().try_into().unwrap());
    }

    /// Clears the timeout of the v4l file handle.
    pub fn clear_timeout(&mut self) {
        self.timeout = None;
    }

    fn buffer_desc(&self) -> v4l2_buffer {
        v4l2_buffer {
            type_: self.buf_type as u32,
            memory: Memory::UserPtr as u32,
            ..unsafe { mem::zeroed() }
        }
    }
}


impl<'a> Stream where Self: CaptureStream<'a> {
    /// Waits for a buffer to be ready
    pub fn wait(&self) -> Result<(), io::Error>{
        if self.handle.poll(libc::POLLIN, -1)? == 0 {
            // This condition can only happen if there was a timeout.
            // A timeout is only possible if the `timeout` value is non-zero, meaning we should
            // propagate it to the caller.
            return Err(io::Error::new(io::ErrorKind::TimedOut, "Blocking poll"));
        }
        Ok(())
    }
    
    /// Waits for a buffer to be ready
    pub fn is_ready(&self) -> Result<bool, io::Error>{
        Ok(self.handle.poll(libc::POLLIN, 0)? > 0)
    }
}

impl Drop for Stream {
    fn drop(&mut self) {
        if let Err(e) = self.stop() {
            if let Some(code) = e.raw_os_error() {
                // ENODEV means the file descriptor wrapped in the handle became invalid, most
                // likely because the device was unplugged or the connection (USB, PCI, ..)
                // broke down. Handle this case gracefully by ignoring it.
                if code == 19 {
                    /* ignore */
                    return;
                }
            }

            panic!("{:?}", e)
        }
    }
}

impl StreamTrait for Stream {
    type Item = DmaBuf;

    fn start(&mut self) -> io::Result<()> {
        unsafe {
            let mut typ = self.buf_type as u32;
            v4l2::ioctl(
                self.handle.fd(),
                v4l2::vidioc::VIDIOC_STREAMON,
                &mut typ as *mut _ as *mut std::os::raw::c_void,
            )?;
        }

        self.active = true;
        Ok(())
    }

    fn stop(&mut self) -> io::Result<()> {
        unsafe {
            let mut typ = self.buf_type as u32;
            v4l2::ioctl(
                self.handle.fd(),
                v4l2::vidioc::VIDIOC_STREAMOFF,
                &mut typ as *mut _ as *mut std::os::raw::c_void,
            )?;
        }

        self.active = false;
        Ok(())
    }
}

impl<'a> CaptureStream<'a> for Stream {
    fn queue(&mut self, index: usize) -> io::Result<()> {
        let buf = &mut self.arena.get_dmabuf_mut(index).map_err(io::Error::other)?;
        let mut v4l2_buf = v4l2_buffer {
            index: index as u32,
            memory: Memory::DmaBuf as u32,
            m: v4l2_buffer__bindgen_ty_1 {
                fd: buf.as_raw_fd(),
            },
            ..self.buffer_desc()
        };
        unsafe {
            v4l2::ioctl(
                self.handle.fd(),
                v4l2::vidioc::VIDIOC_QBUF,
                &mut v4l2_buf as *mut _ as *mut std::os::raw::c_void,
            )?;
        }

        Ok(())
    }

    fn dequeue(&mut self) -> io::Result<usize> {
        let mut v4l2_buf = self.buffer_desc();

        if self.handle.poll(libc::POLLIN, self.timeout.unwrap_or(-1))? == 0 {
            // This condition can only happen if there was a timeout.
            // A timeout is only possible if the `timeout` value is non-zero, meaning we should
            // propagate it to the caller.
            return Err(io::Error::new(io::ErrorKind::TimedOut, "VIDIOC_DQBUF"));
        }

        unsafe {
            v4l2::ioctl(
                self.handle.fd(),
                v4l2::vidioc::VIDIOC_DQBUF,
                &mut v4l2_buf as *mut _ as *mut std::os::raw::c_void,
            )?;
        }
        self.arena_index = v4l2_buf.index as usize;

        self.buf_meta[self.arena_index] = Metadata {
            bytesused: v4l2_buf.bytesused,
            flags: v4l2_buf.flags.into(),
            field: v4l2_buf.field,
            timestamp: v4l2_buf.timestamp.into(),
            sequence: v4l2_buf.sequence,
        };

        Ok(self.arena_index)
    }

    fn next(&'a mut self) -> io::Result<(&Self::Item, &Metadata, bool)> {
        if !self.active {
            // Enqueue all buffers once on stream start
            for index in 0..self.arena.len() {
                self.queue(index)?;
            }

            self.start()?;
        } else {
            self.queue(self.arena_index)?;
        }

        self.arena_index = self.dequeue()?;

        let ready = self.is_ready()?;

        let buf = self.arena.get_dmabuf(self.arena_index).map_err(io::Error::other)?;
        let meta = &self.buf_meta[self.arena_index];
        Ok((buf, meta, ready))
    }
}


/// A stream using the manual buffer management API, where owned buffers are handed to the user, to be enqueued and dequeued explicitly, rather than only borrowed.
pub struct StreamManual {
    io: StreamIo,
    /// Bufers in the arena are always locked for writing.
    arena: arena::ManuallyManaged,
}

impl StreamManual {
    /// Returns an advanced, buffer-owning stream for frame capturing.
    ///
    /// Currently allows only one buffer to be in flight.
    ///
    /// # Arguments
    ///
    /// * `dev` - Device ref to get its file descriptor
    /// * `buf_type` - Type of the buffers
    /// * `buf_count` - the number of buffers to allocate. Too many buffers will hog memory, too few will give less processing time for each buffer (although that doesn't matter with this API). A safe number is 4.
    ///
    /// # Example
    ///
    /// ```
    /// use v4l::buffer::Type;
    /// use v4l::device::Device;
    /// use v4l::io::dmabuf::StreamManual;
    ///
    /// let dev = Device::new(0);
    /// if let Ok(dev) = dev {
    ///     let stream = StreamManual::new(&dev, Type::VideoCapture, 4);
    /// }
    /// ```
    pub fn new(dev: &Device, buf_type: Type, buf_count: u32) -> io::Result<Self> {
        let mut arena = arena::ManuallyManaged::new(dev.handle(), buf_type);
        arena.allocate(buf_count)?;
        
        Ok(Self {
            io: StreamIo {
                streaming: false,
                handle: dev.handle(),
                buf_type,
            },
            arena,
        })
    }

    /// Returns an uninitialized buffer for immediate queueing in cycle_buffer and starts the stream.
    pub fn start(&mut self) -> io::Result<DmaBufProtected> {
        if self.io.streaming {
            return Err(io::Error::other("Already started"));
        }
        // Enqueue all buffers but the one given to the user.
        // The user is expected to enqueue it immediately.
        for i in 1..self.arena.len() {
            self.enqueue(i)?;
        }

        self.io.start()?;

        let buf = self.arena.take_buffer(0)
            .map_err(io::Error::other)?;

        Ok(buf)
    }

    /// Stops the stream and deallocates buffers.
    pub fn finish(mut self, buf: DmaBufProtected) -> Result<(), (io::Error, DmaBufProtected, Self)> {
        if !self.io.streaming {
            // Where did the user get the buffer anyway?
            // Must be one from another device.
            return Err((io::Error::other("Stream wasn't started"), buf, self));
        }
        
        match self.io.stop() {
            Ok(()) => {
                self.arena.replace_buffer(buf)
                    .map(|_| ())
                    .map_err(|(buf, e)| (io::Error::other(e), buf, self))
            },
            Err(e) => Err((e, buf, self)),
        }
    }

    /// Returns a buffer and removes a new buffer once it's ready.
    ///
    /// This will lock the buffer passed in for writing, so the buffer must be unlocked in a timely fashion. Watch out for deadlocks!
    ///
    /// FIXME: if this fails, the buffer is swallowed and streaming can't continue due to no free buffers.
    pub fn cycle_buffer(&mut self, buf: DmaBufProtected) -> io::Result<(DmaBufProtected, Metadata)> {
        if !self.io.streaming {
            return Err(io::Error::other("Streaming wasn't started"));
        }
        let index = self.arena.replace_buffer(buf).map_err(|(_b, e)| io::Error::other(e))?;

        self.enqueue(index)?;

        let (index, meta) = self.dequeue()?;

        // The index used to access the buffer elements is given to us by v4l2, so we assume it
        // will always be valid.
        let buf = self.arena.take_buffer(index).map_err(io::Error::other)?;
        Ok((buf, meta))
    }

    fn enqueue(&mut self, index: usize) -> io::Result<()> {
        let buf = self.arena.get_dmabuf_mut(index)
            .map_err(io::Error::other)?;
        self.io.enqueue(index, buf)
    }

    fn dequeue(&mut self) -> io::Result<(usize, Metadata)> {
        self.io.dequeue()
    }
}