use crate::CodecLayer;
use crate::WriteBufStorage;
use crate::protocol::client::ClientProtocol;
use dope_core::driver::{DriverOps, PoolDriver};
use dope_core::profile::Profile;
use dope_transport::Transport;

use super::super::Pool;
use super::super::core::PushError;
use super::super::slot::{ConnFlags, CoreSlot};
use super::Client;
use super::connect_source::ConnectSource;
use super::state::CLIENT_BYTES_CAP;
use crate::SlotId;

impl<P, T, S, F, B> Pool<Client<P, T, S>, F, B>
where
    P: ClientProtocol,
    T: Transport,
    T::Addr: Clone,
    S: ConnectSource<T>,
    F: Profile,
    B: PoolDriver,
{
    #[inline]
    pub(super) fn try_submit_outbound(&mut self, driver: &mut B::Driver, slot_idx: usize) {
        if !self.core.slots[slot_idx]
            .conn
            .flags
            .contains(ConnFlags::LIVE)
        {
            return;
        }
        if self.core.slots[slot_idx]
            .conn
            .flags
            .contains(ConnFlags::CLOSE_AFTER)
        {
            return;
        }
        if <P::CodecLayer as CodecLayer>::IS_PASSTHROUGH {
            self.try_submit_outbound_plain(driver, slot_idx);
        } else {
            self.try_submit_outbound_codec(driver, slot_idx);
        }
    }

    #[inline]
    fn try_submit_outbound_plain(&mut self, driver: &mut B::Driver, slot_idx: usize) {
        if self.core.slots[slot_idx].io.write.has_in_flight {
            return;
        }
        let conn_fd;
        let token_epoch: Option<(B::Token, u32)>;
        let msg_ptr: *const libc::msghdr;
        let close;
        {
            let external = self.core.slot_id_base + slot_idx;
            let entry = &mut self.core.slots[slot_idx];
            conn_fd = entry.conn.fd;
            let user = entry.user.get_mut();
            match user.session.fill_msghdr(CLIENT_BYTES_CAP) {
                Some(m) => {
                    msg_ptr = m as *const _;
                    token_epoch = Some(entry.io.write.next_token::<B::Token>(external));
                    close = false;
                }
                None => {
                    msg_ptr = std::ptr::null();
                    token_epoch = None;
                    close = self.direction.protocol.wants_close(user.session.state());
                }
            }
        }
        if let Some((token, epoch)) = token_epoch {
            if unsafe { driver.submit_send_msg_tagged(token, conn_fd, msg_ptr) } {
                self.core.slots[slot_idx].io.write.mark_inflight(epoch);
            }
        } else if close {
            self.disconnect_and_free(driver, slot_idx);
        }
    }

    #[cold]
    fn try_submit_outbound_codec(&mut self, driver: &mut B::Driver, slot_idx: usize) {
        let cap = self.core.slots[slot_idx].buf.write_buf.as_slice().len();
        let head = self.core.slots[slot_idx].conn.write_head as usize;
        let room = cap.saturating_sub(head);

        if room > 0 {
            let CoreSlot {
                conn, buf, user, ..
            } = &mut self.core.slots[slot_idx];
            let user = user.get_mut();
            let req_cap = room.min(CLIENT_BYTES_CAP);
            if let Some(msg) = user.session.fill_msghdr(req_cap) {
                let iov_count = msg.msg_iovlen as usize;

                let iov_slice: &[libc::iovec] =
                    unsafe { std::slice::from_raw_parts(msg.msg_iov, iov_count) };
                let mut copied = 0u32;
                for iov in iov_slice {
                    let len = iov.iov_len;
                    if (head + copied as usize) + len > cap {
                        break;
                    }

                    let src = unsafe { std::slice::from_raw_parts(iov.iov_base.cast::<u8>(), len) };
                    let off = head + copied as usize;
                    buf.write_buf.as_mut_slice()[off..off + len].copy_from_slice(src);
                    copied += len as u32;
                }
                conn.write_head = (head + copied as usize).min(u16::MAX as usize) as u16;
                user.session.ack_send(copied);
            }
        }

        let close = {
            let user = self.core.slots[slot_idx].user.get();
            self.direction.protocol.wants_close(user.session.state())
        };
        if close {
            self.core.slots[slot_idx]
                .conn
                .flags
                .set(ConnFlags::CLOSE_AFTER, true);
        }

        self.core.encrypt_write_buf(slot_idx);
        self.core.try_submit_codec(driver, slot_idx);
    }

    pub fn poke_outbound(
        &mut self,
        driver: &mut B::Driver,
        conn_id: SlotId,
    ) -> Result<(), PushError> {
        let slot_idx = self.core.resolve_live_slot(conn_id)?;
        self.try_submit_outbound(driver, slot_idx);
        Ok(())
    }

    pub fn pump_outbound(&mut self, driver: &mut B::Driver) -> usize {
        let mut count = 0usize;
        for slot_idx in 0..self.core.slots.len() {
            if !self.core.slots[slot_idx]
                .conn
                .flags
                .contains(ConnFlags::LIVE)
            {
                continue;
            }
            self.try_submit_outbound(driver, slot_idx);
            count += 1;
        }
        count
    }

    #[cold]
    fn handle_write_event_codec(&mut self, driver: &mut B::Driver, slot_idx: usize, result: i32) {
        match self.core.handle_codec_inflight(slot_idx, result) {
            super::super::core::CodecAck::NoOp => {}
            super::super::core::CodecAck::Err => self.disconnect_and_free(driver, slot_idx),
            super::super::core::CodecAck::Progressed => {
                self.core.try_submit_codec(driver, slot_idx);
                self.try_submit_outbound(driver, slot_idx);
                let close = self.core.slots[slot_idx]
                    .conn
                    .flags
                    .contains(ConnFlags::CLOSE_AFTER)
                    || self.core.codec_close_pending(slot_idx);
                if self.core.codec_idle(slot_idx) && close {
                    self.disconnect_and_free(driver, slot_idx);
                }
            }
        }
    }
}

impl<P, T, S, F, B> super::super::PoolWrite<B> for Pool<Client<P, T, S>, F, B>
where
    P: ClientProtocol,
    T: Transport,
    T::Addr: Clone,
    S: ConnectSource<T>,
    F: Profile,
    B: PoolDriver,
{
    fn on_write_event(
        &mut self,
        driver: &mut B::Driver,
        token: B::Token,
        result: i32,
        notif: bool,
    ) {
        let (slot_idx, result) = match self.core.classify_write(token, result, notif) {
            super::super::core::WriteDispatch::Skipped => return,
            super::super::core::WriteDispatch::Ready { slot_idx, result } => (slot_idx, result),
        };
        if <P::CodecLayer as CodecLayer>::IS_PASSTHROUGH {
            if !self.core.slots[slot_idx]
                .conn
                .flags
                .contains(ConnFlags::LIVE)
            {
                return;
            }
            if result < 0 {
                self.disconnect_and_free(driver, slot_idx);
                return;
            }
            let n = result as u32;
            if n > 0 {
                let user = self.core.slots[slot_idx].user.get_mut();
                user.session.ack_send(n);
            }
            if self.core.slots[slot_idx]
                .conn
                .flags
                .contains(ConnFlags::CLOSE_AFTER)
            {
                self.disconnect_and_free(driver, slot_idx);
                return;
            }
            self.try_submit_outbound(driver, slot_idx);
        } else {
            self.handle_write_event_codec(driver, slot_idx, result);
        }
    }
}
