use std::future::Future;
use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};

use tokio::io::AsyncBufRead;

#[cfg(feature = "stream")]
use futures_core::stream::Stream;

use crate::parser::{Options, Parse, Parser, WithOptions};
use crate::{Error, Result};

use pin_project_lite::pin_project;

pub struct ReadEvent<'x, T, P: Parse> {
	inner: Pin<&'x mut AsyncReader<T, P>>,
}

impl<'x, T: AsyncBufRead + Unpin, P: Parse> Future for ReadEvent<'x, T, P> {
	type Output = Result<Option<P::Output>>;

	fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
		self.inner.as_mut().poll_read(cx)
	}
}

pub struct ReadAll<'x, T, P: Parse, F> {
	cb: F,
	inner: Pin<&'x mut AsyncReader<T, P>>,
}

impl<'x, P: Parse, T: AsyncBufRead + Unpin, F: FnMut(P::Output) -> () + Send + Unpin> Future
	for ReadAll<'x, T, P, F>
{
	type Output = Result<()>;

	fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
		loop {
			match self.inner.as_mut().poll_read(cx) {
				Poll::Ready(Ok(Some(ev))) => {
					(self.cb)(ev);
				}
				Poll::Ready(Ok(None)) => return Poll::Ready(Ok(())),
				Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
				Poll::Pending => return Poll::Pending,
			}
		}
	}
}

#[cfg(feature = "stream")]
#[cfg_attr(docsrs, doc(cfg(all(feature = "stream", feature = "tokio"))))]
impl<T: AsyncBufRead, P: Parse> Stream for AsyncReader<T, P> {
	type Item = Result<P::Output>;

	fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
		match self.poll_read(cx) {
			Poll::Pending => Poll::Pending,
			Poll::Ready(Ok(Some(v))) => Poll::Ready(Some(Ok(v))),
			Poll::Ready(Ok(None)) => Poll::Ready(None),
			Poll::Ready(Err(e)) => Poll::Ready(Some(Err(e))),
		}
	}
}

pin_project! {
	/**
	# Tokio-compatible asynchronous restricted XML 1.0 parser

	The [`AsyncReader`] allows parsing XML documents from a
	[`tokio::io::AsyncBufRead`], asynchronously. It operates similarly as the
	[`Reader`][`crate::Reader`] does, but it works with asynchronous data
	sources instead of synchronous (blocking) ones.

	Events can be obtained through [`read`][`Self::read`] and
	[`read_all`][`Self::read_all`]. If the `stream` feature is enabled,
	[`AsyncReader`] also implements the
	[`Stream`][`futures_core::stream::Stream`] trait.

	## Example

	The example is a bit pointless because it does not really demonstrate the
	asynchronicity.

	```
	use rxml::{AsyncReader, Error, ResolvedEvent, XmlVersion};
	use tokio::io::AsyncRead;
	# tokio_test::block_on(async {
	let mut doc = &b"<?xml version='1.0'?><hello>World!</hello>"[..];
	// this converts the doc into an tokio::io::AsyncRead
	let mut pp = AsyncReader::<_>::new(&mut doc);
	// we expect the first event to be the XML declaration
	let ev = pp.read().await;
	assert!(matches!(ev.unwrap().unwrap(), ResolvedEvent::XmlDeclaration(_, XmlVersion::V1_0)));
	# })
	```

	## Parsing without namespace expansion

	To parse an XML document without namespace expansion using tokio,
	pass [`RawParser`] as second type argument:

	```
	use rxml::{AsyncReader, Error, RawEvent, RawParser, XmlVersion};
	use tokio::io::AsyncRead;
	# tokio_test::block_on(async {
	let mut doc = &b"<?xml version='1.0'?><hello>World!</hello>"[..];
	// this converts the doc into an tokio::io::AsyncRead
	let mut pp = AsyncReader::<_, RawParser>::new(&mut doc);
	// we expect the first event to be the XML declaration
	let ev = pp.read().await;
	assert!(matches!(ev.unwrap().unwrap(), RawEvent::XmlDeclaration(_, XmlVersion::V1_0)));
	# })
	```

	Note the caveats in the [`RawParser`] documentation before using it!

	[`RawParser`]: crate::parser::RawParser
	*/
	#[project = AsyncReaderProj]
	pub struct AsyncReader<T, P: Parse = Parser>{
		#[pin]
		inner: T,
		parser: P,
	}
}

impl<T: AsyncBufRead, P: Parse + Default> AsyncReader<T, P> {
	/// Create a reader using a parser with default options, wrapping the
	/// given reader.
	pub fn new(inner: T) -> Self {
		Self::wrap(inner, P::default())
	}
}

impl<T: AsyncBufRead, P: Parse + WithOptions> AsyncReader<T, P> {
	/// Create a reader while configuring the parser with the given options.
	pub fn with_options(inner: T, options: Options) -> Self {
		Self::wrap(inner, P::with_options(options))
	}
}

impl<T: AsyncBufRead, P: Parse> AsyncReader<T, P> {
	/// Create a reader from its inner parts.
	pub fn wrap(inner: T, parser: P) -> Self {
		Self { inner, parser }
	}

	/// Decompose the AsyncReader into its parts
	pub fn into_inner(self) -> (T, P) {
		(self.inner, self.parser)
	}

	/// Access the inner AsyncBufRead
	#[deprecated(since = "0.10.0", note = "use inner() instead")]
	#[doc(hidden)]
	pub fn get_inner(&self) -> &T {
		&self.inner
	}

	/// Access the inner AsyncBufRead, mutably
	#[deprecated(since = "0.10.0", note = "use inner_mut() instead")]
	#[doc(hidden)]
	pub fn get_inner_mut(&mut self) -> &mut T {
		&mut self.inner
	}

	/// Access the parser
	#[deprecated(since = "0.10.0", note = "use parser() instead")]
	#[doc(hidden)]
	pub fn get_parser(&self) -> &P {
		&self.parser
	}

	/// Access the parser, mutably
	#[deprecated(since = "0.10.0", note = "use parser_mut() instead")]
	#[doc(hidden)]
	pub fn get_parser_mut(&mut self) -> &mut P {
		&mut self.parser
	}

	/// Access the inner AsyncBufRead
	pub fn inner(&self) -> &T {
		&self.inner
	}

	/// Access the inner AsyncBufRead, mutably
	pub fn inner_mut(&mut self) -> &mut T {
		&mut self.inner
	}

	/// Access the parser
	pub fn parser(&self) -> &P {
		&self.parser
	}

	/// Access the parser, mutably
	pub fn parser_mut(&mut self) -> &mut P {
		&mut self.parser
	}

	/// Release temporary buffers and other ephemeral allocations.
	///
	/// This is sensible to call when it is expected that no more data will be
	/// processed by the parser for a while and the memory is better used
	/// elsewhere.
	#[inline(always)]
	pub fn release_temporaries(&mut self) {
		self.parser.release_temporaries();
	}
}

impl<T, P: Parse> AsyncReader<T, P> {
	fn parse_step(
		parser: &mut P,
		buf: &mut &[u8],
		may_eof: bool,
	) -> (usize, Poll<Result<Option<P::Output>>>) {
		let old_len = buf.len();
		// need to guard eof with the buf len here, because we only know that we are actually at eof by the fact that we see an empty buffer.
		let at_eof = may_eof && buf.len() == 0;
		let result = parser.parse(buf, at_eof);
		let new_len = buf.len();
		assert!(new_len <= old_len);
		let read = old_len - new_len;
		match result {
			Ok(v) => (read, Poll::Ready(Ok(v))),
			Err(Error::IO(ioerr)) if ioerr.kind() == io::ErrorKind::WouldBlock => {
				(read, Poll::Pending)
			}
			Err(e) => (read, Poll::Ready(Err(e))),
		}
	}
}

impl<T: AsyncBufRead, P: Parse> AsyncReader<T, P> {
	/// Attempts to parse a single event from the source.
	///
	/// If the EOF has been reached with a valid document, `None` is returned.
	///
	/// I/O errors may be retried, all other errors are fatal (and will be
	/// returned again by the parser on the next invocation without reading
	/// further data from the source).
	///
	/// In most cases, it is advisable to use [`read`][`Self::read`] instead.
	pub fn poll_read(
		self: Pin<&mut Self>,
		cx: &mut Context<'_>,
	) -> Poll<Result<Option<P::Output>>> {
		let mut this = self.project();
		loop {
			let mut buf = match this.inner.as_mut().poll_fill_buf(cx) {
				Poll::Pending => {
					// a.k.a. WouldBlock
					// we always try an empty read here because the lexer needs that
					return Self::parse_step(this.parser, &mut &[][..], false).1;
				}
				Poll::Ready(Ok(buf)) => buf,
				Poll::Ready(Err(e)) => return Poll::Ready(Err(e.into())),
			};
			let (consumed, result) = Self::parse_step(this.parser, &mut buf, true);
			this.inner.as_mut().consume(consumed);
			match result {
				// if we get a pending here, we need to ask the source for more data!
				Poll::Pending => continue,
				Poll::Ready(v) => return Poll::Ready(v),
			}
		}
	}
}

impl<T: AsyncBufRead + Unpin, P: Parse> AsyncReader<T, P> {
	/// Read a single event from the parser.
	///
	/// If the EOF has been reached with a valid document, `None` is returned.
	///
	/// I/O errors may be retried, all other errors are fatal (and will be
	/// returned again by the parser on the next invocation without reading
	/// further data from the source).
	///
	/// Equivalent to:
	///
	/// ```ignore
	/// async fn read(&mut self) -> Result<Option<ResolvedEvent>>;
	/// ```
	pub fn read(&mut self) -> ReadEvent<'_, T, P> {
		ReadEvent {
			inner: Pin::new(self),
		}
	}

	/// Read all events which can be produced from the data source (at this
	/// point in time).
	///
	/// The given `cb` is invoked for each event.
	///
	/// I/O errors may be retried, all other errors are fatal (and will be
	/// returned again by the parser on the next invocation without reading
	/// further data from the source).
	///
	/// Equivalent to:
	///
	/// ```ignore
	///     async fn read_all<F>(&mut self, mut cb: F) -> Result<()>
	///            where F: FnMut(ResolvedEvent) -> () + Send
	/// ```
	pub fn read_all<F>(&mut self, cb: F) -> ReadAll<'_, T, P, F> {
		ReadAll {
			inner: Pin::new(self),
			cb,
		}
	}
}