use super::super::Pool;
use super::super::slot::ConnFlags;
use super::Server;
use super::state::{SlotEgress, StreamState};
use super::wake::WakeQueue;
use crate::SlotId;
use crate::WriteBufStorage;
use crate::protocol::server::ServerProtocol;
use crate::protocol::server::{PendingOutcome, PendingResponse};
use dope_core::driver::PoolDriver;
use dope_core::profile::Profile;
use dope_transport::Transport;
use std::task::{Context, Poll};

impl<H: ServerProtocol, T: Transport, F: Profile, B: PoolDriver> Pool<Server<H, T>, F, B> {
    #[inline(always)]
    pub(crate) fn drain_ready_queue(&mut self, driver: &mut B::Driver) {
        if !WakeQueue::is_pending() {
            return;
        }
        self.drain_ready_queue_slow(driver);
    }

    #[cold]
    #[inline(never)]
    fn drain_ready_queue_slow(&mut self, driver: &mut B::Driver) {
        loop {
            let id = match WakeQueue::next() {
                Some(i) => i,
                None => return,
            };
            self.poll_pending(driver, id);
            self.poll_long(driver, id);
        }
    }

    #[cold]
    #[inline(never)]
    pub(super) fn poll_pending(&mut self, driver: &mut B::Driver, id: SlotId) {
        let slot_idx = id.slot() as usize;
        if slot_idx >= self.core.slots.len() {
            return;
        }
        let user = self.core.slots[slot_idx].user.get();
        let fresh = match user.pending_future.as_ref() {
            Some(pf) => pf.generation == id.generation(),
            None => false,
        };
        if !fresh {
            return;
        }
        {
            let conn = &self.core.slots[slot_idx].conn;
            if !conn.flags.contains(ConnFlags::LIVE) || conn.generation != id.generation() {
                self.core.slots[slot_idx].user.get_mut().pending_future = None;
                return;
            }
        }
        let waker = WakeQueue::make_waker(id);
        let mut cx = Context::from_waker(&waker);
        let poll_result = {
            let user = self.core.slots[slot_idx].user.get_mut();
            let pf = user
                .pending_future
                .as_mut()
                .expect("just validated by fresh check");

            unsafe { pf.future.poll_pinned(&mut cx) }
        };
        match poll_result {
            Poll::Pending => {}
            Poll::Ready(reply) => self.finish_async(driver, slot_idx, reply),
        }
    }

    #[cold]
    #[inline(never)]
    pub(super) fn poll_long(&mut self, driver: &mut B::Driver, id: SlotId) {
        let slot_idx = id.slot() as usize;
        if slot_idx >= self.core.slots.len() {
            return;
        }
        let user = self.core.slots[slot_idx].user.get();
        let fresh = match user.long_future.as_ref() {
            Some(lf) => lf.generation == id.generation(),
            None => false,
        };
        if !fresh {
            return;
        }
        {
            let conn = &self.core.slots[slot_idx].conn;
            if !conn.flags.contains(ConnFlags::LIVE) || conn.generation != id.generation() {
                self.core.slots[slot_idx].user.get_mut().long_future = None;
                return;
            }
        }
        let waker = WakeQueue::make_waker(id);
        let mut cx = Context::from_waker(&waker);
        let poll_result = {
            let user = self.core.slots[slot_idx].user.get_mut();
            let lf = user
                .long_future
                .as_mut()
                .expect("just validated by fresh check");
            lf.future.as_mut().poll(&mut cx)
        };

        let wh = self.core.slots[slot_idx].conn.write_head as usize;
        let written = {
            let entry = &mut self.core.slots[slot_idx];
            let cap = entry.buf.write_buf.as_slice().len();
            let room = cap.saturating_sub(wh);
            if room > 0 {
                let slice = &mut entry.buf.write_buf.as_mut_slice()[wh..wh + room];
                let user = entry.user.get_mut();
                self.direction
                    .protocol
                    .drain_outbound(&mut user.protocol_state, slice)
            } else {
                0
            }
        };
        if written > 0 {
            let new_head = (wh as u32 + written).min(u16::MAX as u32) as u16;
            self.core.slots[slot_idx].conn.write_head = new_head;
        }

        let close = poll_result.is_ready() || {
            let user = self.core.slots[slot_idx].user.get();
            self.direction.protocol.close_pending(&user.protocol_state)
        };
        if close {
            self.core.slots[slot_idx]
                .conn
                .flags
                .set(ConnFlags::CLOSE_AFTER, true);
            self.core.slots[slot_idx].user.get_mut().long_future = None;
        }

        self.submit_slot_write(driver, slot_idx);
    }

    #[cold]
    #[inline(never)]
    pub(super) fn finish_async(
        &mut self,
        driver: &mut B::Driver,
        slot_idx: usize,
        reply: PendingResponse,
    ) {
        let entry = &mut self.core.slots[slot_idx];
        entry.user.get_mut().pending_future = None;
        let outcome = {
            let wh = entry.conn.write_head as usize;
            let write_slice = &mut entry.buf.write_buf.as_mut_slice()[wh..];
            (reply.serializer)(write_slice)
        };
        match outcome {
            PendingOutcome::Inline {
                written,
                close_after,
            } => {
                entry.conn.flags.set(ConnFlags::PARKED, false);
                let new_head = (entry.conn.write_head as u32).saturating_add(written as u32);
                entry.conn.write_head = new_head.min(u16::MAX as u32) as u16;
                if close_after {
                    entry.conn.flags.set(ConnFlags::CLOSE_AFTER, true);
                }
            }
            PendingOutcome::Static {
                hdr_written,
                body,
                close_after,
            } => {
                entry.conn.flags.set(ConnFlags::PARKED, false);
                let new_head =
                    (entry.conn.write_head as u32 + hdr_written).min(u16::MAX as u32) as u16;
                entry.conn.write_head = new_head;
                entry.conn.pending_body_ptr = body.as_ptr();
                entry.conn.pending_body_len = body.len() as u32;
                if close_after {
                    entry.conn.flags.set(ConnFlags::CLOSE_AFTER, true);
                }
            }
            PendingOutcome::Stream {
                hdr_written,
                stream,
                close_after,
            } => {
                entry.conn.flags.set(ConnFlags::PARKED, false);
                let new_head =
                    (entry.conn.write_head as u32 + hdr_written).min(u16::MAX as u32) as u16;
                entry.conn.write_head = new_head;
                entry.user.get_mut().egress =
                    SlotEgress::Stream(Box::new(StreamState::new(stream)));
                if close_after {
                    entry.conn.flags.set(ConnFlags::CLOSE_AFTER, true);
                }
            }
        }
        self.parse_and_dispatch(slot_idx);
        self.submit_slot_write(driver, slot_idx);
    }
}
