//! Traits and implementations of arbitrary data streams.
//!
//! Streams are similar to the `Iterator` trait in that they represent some sequential set of items
//! which can be retrieved one by one. Where `Stream`s differ is that they are allowed to return
//! errors instead of just `None` and if they implement the `RangeStreamOnce` trait they are also
//! capable of returning multiple items at the same time, usually in the form of a slice.
//!
//! In addition to he functionality above, a proper `Stream` usable by a `Parser` must also have a
//! position (marked by the `Positioned` trait) and must also be resetable (marked by the
//! `ResetStream` trait). The former is used to ensure that errors at different points in the stream
//! aren't combined and the latter is used in parsers such as `or` to try multiple alternative
//! parses.

use crate::lib::{cmp::Ordering, fmt, marker::PhantomData, mem, str::Chars};

use crate::{
    error::{
        ParseError,
        ParseResult::{self, *},
        StreamError, StringStreamError, Tracked, UnexpectedParse,
    },
    Parser,
};

#[doc(hidden)]
#[macro_export]
macro_rules! clone_resetable {
    (( $($params: tt)* ) $ty: ty) => {
        impl<$($params)*> ResetStream for $ty
            where Self: StreamOnce
        {
            type Checkpoint = Self;

            fn checkpoint(&self) -> Self {
                self.clone()
            }
            #[inline]
            fn reset(&mut self, checkpoint: Self) -> Result<(), Self::Error> {
                *self = checkpoint;
                Ok(())
            }
        }
    }
}

#[cfg(feature = "std")]
/// Stream wrapper which provides a `ResetStream` impl for `StreamOnce` impls which do not have
/// one.
pub mod buffered;
#[cfg(feature = "std")]
pub mod easy;
/// Stream wrapper which provides more detailed position information.
pub mod position;
/// Stream wrapper allowing `std::io::Read` to be used
#[cfg(feature = "std")]
pub mod read;
/// Stream wrapper allowing custom state to be used.
pub mod state;

/// A type which has a position.
pub trait Positioned: StreamOnce {
    /// Returns the current position of the stream.
    fn position(&self) -> Self::Position;
}

/// Convenience alias over the `StreamError` for the input stream `Input`
///
/// ```
/// #[macro_use]
/// extern crate combine;
/// use combine::{easy, Parser, Stream, many1};
/// use combine::parser::char::letter;
/// use combine::stream::StreamErrorFor;
/// use combine::error::{ParseError, StreamError};
///
/// parser!{
///    fn parser[Input]()(Input) -> String
///     where [ Input: Stream<Token = char>, ]
///     {
///         many1(letter()).and_then(|word: String| {
///             if word == "combine" {
///                 Ok(word)
///             } else {
///                 // The alias makes it easy to refer to the `StreamError` type of `Input`
///                 Err(StreamErrorFor::<Input>::expected_static_message("combine"))
///             }
///         })
///     }
/// }
///
/// fn main() {
/// }
/// ```
pub type StreamErrorFor<Input> = <<Input as StreamOnce>::Error as ParseError<
    <Input as StreamOnce>::Token,
    <Input as StreamOnce>::Range,
    <Input as StreamOnce>::Position,
>>::StreamError;

/// `StreamOnce` represents a sequence of items that can be extracted one by one.
pub trait StreamOnce {
    /// The type of items which is yielded from this stream.
    type Token: Clone;

    /// The type of a range of items yielded from this stream.
    /// Types which do not a have a way of yielding ranges of items should just use the
    /// `Self::Token` for this type.
    type Range: Clone;

    /// Type which represents the position in a stream.
    /// `Ord` is required to allow parsers to determine which of two positions are further ahead.
    type Position: Clone + Ord;

    type Error: ParseError<Self::Token, Self::Range, Self::Position>;
    /// Takes a stream and removes its first token, yielding the token and the rest of the elements.
    /// Returns `Err` if no element could be retrieved.
    fn uncons(&mut self) -> Result<Self::Token, StreamErrorFor<Self>>;

    /// Returns `true` if this stream only contains partial input.
    ///
    /// See `PartialStream`.
    fn is_partial(&self) -> bool {
        false
    }
}

/// A `StreamOnce` which can create checkpoints which the stream can be reset to
pub trait ResetStream: StreamOnce {
    type Checkpoint: Clone;

    /// Creates a `Checkpoint` at the current position which can be used to reset the stream
    /// later to the current position
    fn checkpoint(&self) -> Self::Checkpoint;
    /// Attempts to reset the stream to an earlier position.
    fn reset(&mut self, checkpoint: Self::Checkpoint) -> Result<(), Self::Error>;
}

clone_resetable! {('a) &'a str}
clone_resetable! {('a, T) &'a [T]}
clone_resetable! {('a, T) SliceStream<'a, T> }
clone_resetable! {(T: Clone) IteratorStream<T>}

/// A stream of tokens which can be duplicated
///
/// This is a trait over types which implement the `StreamOnce`, `ResetStream` and `Positioned`
/// traits. If you need a custom `Stream` object then implement those traits and `Stream` is
/// implemented automatically.
pub trait Stream: StreamOnce + ResetStream + Positioned {}

impl<Input> Stream for Input
where
    Input: StreamOnce + Positioned + ResetStream,
    Input::Error: ParseError<Input::Token, Input::Range, Input::Position>,
{
}

#[inline]
pub fn uncons<Input>(input: &mut Input) -> ParseResult<Input::Token, Input::Error>
where
    Input: ?Sized + Stream,
{
    match input.uncons() {
        Ok(x) => CommitOk(x),
        Err(err) => wrap_stream_error(input, err),
    }
}

/// A `RangeStream` is an extension of `StreamOnce` which allows for zero copy parsing.
pub trait RangeStreamOnce: StreamOnce + ResetStream {
    /// Takes `size` elements from the stream.
    /// Fails if the length of the stream is less than `size`.
    fn uncons_range(&mut self, size: usize) -> Result<Self::Range, StreamErrorFor<Self>>;

    /// Takes items from stream, testing each one with `predicate`.
    /// returns the range of items which passed `predicate`.
    fn uncons_while<F>(&mut self, f: F) -> Result<Self::Range, StreamErrorFor<Self>>
    where
        F: FnMut(Self::Token) -> bool;

    #[inline]
    /// Takes items from stream, testing each one with `predicate`
    /// returns a range of at least one items which passed `predicate`.
    ///
    /// # Note
    ///
    /// This may not return `PeekOk` as it should uncons at least one token.
    fn uncons_while1<F>(&mut self, mut f: F) -> ParseResult<Self::Range, StreamErrorFor<Self>>
    where
        F: FnMut(Self::Token) -> bool,
    {
        let mut consumed = false;
        let result = self.uncons_while(|c| {
            let ok = f(c);
            consumed |= ok;
            ok
        });
        if consumed {
            match result {
                Ok(x) => CommitOk(x),
                Err(x) => CommitErr(x),
            }
        } else {
            PeekErr(Tracked::from(
                StreamErrorFor::<Self>::unexpected_static_message(""),
            ))
        }
    }

    /// Returns the distance between `self` and `end`. The returned `usize` must be so that
    ///
    /// ```ignore
    /// let start = stream.checkpoint();
    /// stream.uncons_range(distance);
    /// stream.distance(&start) == distance
    /// ```
    fn distance(&self, end: &Self::Checkpoint) -> usize;

    /// Returns the entire range of `self`
    fn range(&self) -> Self::Range;
}

/// A `RangeStream` is an extension of `Stream` which allows for zero copy parsing.
pub trait RangeStream: Stream + RangeStreamOnce {}

impl<Input> RangeStream for Input where Input: RangeStreamOnce + Stream {}

#[doc(hidden)]
pub fn wrap_stream_error<T, Input>(
    input: &Input,
    err: <Input::Error as ParseError<Input::Token, Input::Range, Input::Position>>::StreamError,
) -> ParseResult<T, <Input as StreamOnce>::Error>
where
    Input: ?Sized + StreamOnce + Positioned,
{
    let err = Input::Error::from_error(input.position(), err);
    if input.is_partial() {
        CommitErr(err)
    } else {
        PeekErr(err.into())
    }
}

#[inline]
pub fn uncons_range<Input>(
    input: &mut Input,
    size: usize,
) -> ParseResult<Input::Range, <Input as StreamOnce>::Error>
where
    Input: ?Sized + RangeStream,
{
    match input.uncons_range(size) {
        Err(err) => wrap_stream_error(input, err),
        Ok(x) => {
            if size == 0 {
                PeekOk(x)
            } else {
                CommitOk(x)
            }
        }
    }
}

#[doc(hidden)]
pub fn input_at_eof<Input>(input: &mut Input) -> bool
where
    Input: ?Sized + Stream,
{
    let before = input.checkpoint();
    let x = input
        .uncons()
        .err()
        .map_or(false, |err| err.is_unexpected_end_of_input());
    input.reset(before).is_ok() && x
}

/// Removes items from the input while `predicate` returns `true`.
#[inline]
pub fn uncons_while<Input, F>(
    input: &mut Input,
    predicate: F,
) -> ParseResult<Input::Range, Input::Error>
where
    F: FnMut(Input::Token) -> bool,
    Input: ?Sized + RangeStream,
    Input::Range: Range,
{
    match input.uncons_while(predicate) {
        Err(err) => wrap_stream_error(input, err),
        Ok(x) => {
            if input.is_partial() && input_at_eof(input) {
                // Partial inputs which encounter end of file must fail to let more input be
                // retrieved
                CommitErr(Input::Error::from_error(
                    input.position(),
                    StreamError::end_of_input(),
                ))
            } else if x.len() == 0 {
                PeekOk(x)
            } else {
                CommitOk(x)
            }
        }
    }
}

#[inline]
/// Takes items from stream, testing each one with `predicate`
/// returns a range of at least one items which passed `predicate`.
///
/// # Note
///
/// This may not return `PeekOk` as it should uncons at least one token.
pub fn uncons_while1<Input, F>(
    input: &mut Input,
    predicate: F,
) -> ParseResult<Input::Range, Input::Error>
where
    F: FnMut(Input::Token) -> bool,
    Input: ?Sized + RangeStream,
{
    match input.uncons_while1(predicate) {
        CommitOk(x) => {
            if input.is_partial() && input_at_eof(input) {
                // Partial inputs which encounter end of file must fail to let more input be
                // retrieved
                CommitErr(Input::Error::from_error(
                    input.position(),
                    StreamError::end_of_input(),
                ))
            } else {
                CommitOk(x)
            }
        }
        PeekErr(_) => {
            if input.is_partial() && input_at_eof(input) {
                // Partial inputs which encounter end of file must fail to let more input be
                // retrieved
                CommitErr(Input::Error::from_error(
                    input.position(),
                    StreamError::end_of_input(),
                ))
            } else {
                PeekErr(Input::Error::empty(input.position()).into())
            }
        }
        CommitErr(err) => {
            if input.is_partial() && input_at_eof(input) {
                // Partial inputs which encounter end of file must fail to let more input be
                // retrieved
                CommitErr(Input::Error::from_error(
                    input.position(),
                    StreamError::end_of_input(),
                ))
            } else {
                wrap_stream_error(input, err)
            }
        }
        PeekOk(_) => unreachable!(),
    }
}

/// Trait representing a range of elements.
pub trait Range {
    /// Returns the remaining length of `self`.
    /// The returned length need not be the same as the number of items left in the stream.
    fn len(&self) -> usize;

    /// Returns `true` if the range does not contain any elements (`Range::len() == 0`)
    fn is_empty(&self) -> bool {
        self.len() == 0
    }
}

impl<'a, I> StreamOnce for &'a mut I
where
    I: StreamOnce + ?Sized,
{
    type Token = I::Token;

    type Range = I::Range;

    type Position = I::Position;

    type Error = I::Error;
    fn uncons(&mut self) -> Result<Self::Token, StreamErrorFor<Self>> {
        (**self).uncons()
    }

    fn is_partial(&self) -> bool {
        (**self).is_partial()
    }
}

impl<'a, I> Positioned for &'a mut I
where
    I: Positioned + ?Sized,
{
    #[inline]
    fn position(&self) -> Self::Position {
        (**self).position()
    }
}

impl<'a, I> ResetStream for &'a mut I
where
    I: ResetStream + ?Sized,
{
    type Checkpoint = I::Checkpoint;

    fn checkpoint(&self) -> Self::Checkpoint {
        (**self).checkpoint()
    }

    fn reset(&mut self, checkpoint: Self::Checkpoint) -> Result<(), Self::Error> {
        (**self).reset(checkpoint)
    }
}

impl<'a, I> RangeStreamOnce for &'a mut I
where
    I: RangeStreamOnce + ?Sized,
{
    #[inline]
    fn uncons_while<F>(&mut self, f: F) -> Result<Self::Range, StreamErrorFor<Self>>
    where
        F: FnMut(Self::Token) -> bool,
    {
        (**self).uncons_while(f)
    }

    #[inline]
    fn uncons_while1<F>(&mut self, f: F) -> ParseResult<Self::Range, StreamErrorFor<Self>>
    where
        F: FnMut(Self::Token) -> bool,
    {
        (**self).uncons_while1(f)
    }

    #[inline]
    fn uncons_range(&mut self, size: usize) -> Result<Self::Range, StreamErrorFor<Self>> {
        (**self).uncons_range(size)
    }

    #[inline]
    fn distance(&self, end: &Self::Checkpoint) -> usize {
        (**self).distance(end)
    }

    fn range(&self) -> Self::Range {
        (**self).range()
    }
}

impl<'a, I> Range for &'a mut I
where
    I: Range + ?Sized,
{
    fn len(&self) -> usize {
        (**self).len()
    }
}

impl<'a> StreamOnce for &'a str {
    type Token = char;
    type Range = &'a str;
    type Position = PointerOffset<str>;
    type Error = StringStreamError;

    #[inline]
    fn uncons(&mut self) -> Result<char, StreamErrorFor<Self>> {
        let mut chars = self.chars();
        match chars.next() {
            Some(c) => {
                *self = chars.as_str();
                Ok(c)
            }
            None => Err(StringStreamError::Eoi),
        }
    }
}

impl<'a> Positioned for &'a str {
    #[inline]
    fn position(&self) -> Self::Position {
        PointerOffset::new(self.as_bytes().position().0)
    }
}

fn str_uncons_while<'a, F>(slice: &mut &'a str, mut chars: Chars<'a>, mut f: F) -> &'a str
where
    F: FnMut(char) -> bool,
{
    let mut last_char_size = 0;

    macro_rules! test_next {
        () => {
            match chars.next() {
                Some(c) => {
                    if !f(c) {
                        last_char_size = c.len_utf8();
                        break;
                    }
                }
                None => break,
            }
        };
    }
    loop {
        test_next!();
        test_next!();
        test_next!();
        test_next!();
        test_next!();
        test_next!();
        test_next!();
        test_next!();
    }

    let len = slice.len() - chars.as_str().len() - last_char_size;
    let (result, rest) = slice.split_at(len);
    *slice = rest;
    result
}

impl<'a> RangeStreamOnce for &'a str {
    fn uncons_while<F>(&mut self, f: F) -> Result<&'a str, StreamErrorFor<Self>>
    where
        F: FnMut(Self::Token) -> bool,
    {
        Ok(str_uncons_while(self, self.chars(), f))
    }

    #[inline]
    fn uncons_while1<F>(&mut self, mut f: F) -> ParseResult<Self::Range, StreamErrorFor<Self>>
    where
        F: FnMut(Self::Token) -> bool,
    {
        let mut chars = self.chars();
        match chars.next() {
            Some(c) => {
                if !f(c) {
                    return PeekErr(Tracked::from(StringStreamError::UnexpectedParse));
                }
            }
            None => return PeekErr(Tracked::from(StringStreamError::UnexpectedParse)),
        }

        CommitOk(str_uncons_while(self, chars, f))
    }

    #[inline]
    fn uncons_range(&mut self, size: usize) -> Result<&'a str, StreamErrorFor<Self>> {
        fn is_char_boundary(s: &str, index: usize) -> bool {
            if index == s.len() {
                return true;
            }
            match s.as_bytes().get(index) {
                None => false,
                Some(&b) => b < 128 || b >= 192,
            }
        }
        if size <= self.len() {
            if is_char_boundary(self, size) {
                let (result, remaining) = self.split_at(size);
                *self = remaining;
                Ok(result)
            } else {
                Err(StringStreamError::CharacterBoundary)
            }
        } else {
            Err(StringStreamError::Eoi)
        }
    }

    #[inline]
    fn distance(&self, end: &Self) -> usize {
        self.position().0 - end.position().0
    }

    fn range(&self) -> Self::Range {
        self
    }
}

impl<'a> Range for &'a str {
    #[inline]
    fn len(&self) -> usize {
        str::len(self)
    }
}

impl<'a, T> Range for &'a [T] {
    #[inline]
    fn len(&self) -> usize {
        <[T]>::len(self)
    }
}

fn slice_uncons_while<'a, T, F>(slice: &mut &'a [T], mut i: usize, mut f: F) -> &'a [T]
where
    F: FnMut(T) -> bool,
    T: Clone,
{
    let len = slice.len();
    let mut found = false;

    macro_rules! check {
        () => {
            if !f(unsafe { slice.get_unchecked(i).clone() }) {
                found = true;
                break;
            }
            i += 1;
        };
    }

    while len - i >= 8 {
        check!();
        check!();
        check!();
        check!();
        check!();
        check!();
        check!();
        check!();
    }

    if !found {
        while i < len {
            if !f(unsafe { slice.get_unchecked(i).clone() }) {
                break;
            }
            i += 1;
        }
    }

    let (result, remaining) = slice.split_at(i);
    *slice = remaining;
    result
}

impl<'a, T> RangeStreamOnce for &'a [T]
where
    T: Clone + PartialEq,
{
    #[inline]
    fn uncons_range(&mut self, size: usize) -> Result<&'a [T], StreamErrorFor<Self>> {
        if size <= self.len() {
            let (result, remaining) = self.split_at(size);
            *self = remaining;
            Ok(result)
        } else {
            Err(UnexpectedParse::Eoi)
        }
    }

    #[inline]
    fn uncons_while<F>(&mut self, f: F) -> Result<&'a [T], StreamErrorFor<Self>>
    where
        F: FnMut(Self::Token) -> bool,
    {
        Ok(slice_uncons_while(self, 0, f))
    }

    #[inline]
    fn uncons_while1<F>(&mut self, mut f: F) -> ParseResult<Self::Range, StreamErrorFor<Self>>
    where
        F: FnMut(Self::Token) -> bool,
    {
        if self.is_empty() || !f(unsafe { (*self.get_unchecked(0)).clone() }) {
            return PeekErr(Tracked::from(UnexpectedParse::Unexpected));
        }

        CommitOk(slice_uncons_while(self, 1, f))
    }

    #[inline]
    fn distance(&self, end: &Self) -> usize {
        end.len() - self.len()
    }

    fn range(&self) -> Self::Range {
        self
    }
}

impl<'a, T> Positioned for &'a [T]
where
    T: Clone + PartialEq,
{
    #[inline]
    fn position(&self) -> Self::Position {
        PointerOffset::new(self.as_ptr() as usize)
    }
}

impl<'a, T> StreamOnce for &'a [T]
where
    T: Clone + PartialEq,
{
    type Token = T;
    type Range = &'a [T];
    type Position = PointerOffset<[T]>;
    type Error = UnexpectedParse;

    #[inline]
    fn uncons(&mut self) -> Result<T, StreamErrorFor<Self>> {
        match self.split_first() {
            Some((first, rest)) => {
                *self = rest;
                Ok(first.clone())
            }
            None => Err(UnexpectedParse::Eoi),
        }
    }
}

/// Stream type which indicates that the stream is partial if end of input is reached
#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Debug)]
pub struct PartialStream<S>(pub S);

impl<S> From<S> for PartialStream<S> {
    fn from(t: S) -> Self {
        PartialStream(t)
    }
}

impl<S> Positioned for PartialStream<S>
where
    S: Positioned,
{
    #[inline]
    fn position(&self) -> Self::Position {
        self.0.position()
    }
}

impl<S> ResetStream for PartialStream<S>
where
    S: ResetStream,
{
    type Checkpoint = S::Checkpoint;

    #[inline]
    fn checkpoint(&self) -> Self::Checkpoint {
        self.0.checkpoint()
    }

    #[inline]
    fn reset(&mut self, checkpoint: Self::Checkpoint) -> Result<(), S::Error> {
        self.0.reset(checkpoint)
    }
}

impl<S> StreamOnce for PartialStream<S>
where
    S: StreamOnce,
{
    type Token = S::Token;
    type Range = S::Range;
    type Position = S::Position;
    type Error = S::Error;

    #[inline]
    fn uncons(&mut self) -> Result<S::Token, StreamErrorFor<Self>> {
        self.0.uncons()
    }

    fn is_partial(&self) -> bool {
        true
    }
}

impl<S> RangeStreamOnce for PartialStream<S>
where
    S: RangeStreamOnce,
{
    #[inline]
    fn uncons_range(&mut self, size: usize) -> Result<Self::Range, StreamErrorFor<Self>> {
        self.0.uncons_range(size)
    }

    #[inline]
    fn uncons_while<F>(&mut self, f: F) -> Result<Self::Range, StreamErrorFor<Self>>
    where
        F: FnMut(Self::Token) -> bool,
    {
        self.0.uncons_while(f)
    }

    fn uncons_while1<F>(&mut self, f: F) -> ParseResult<Self::Range, StreamErrorFor<Self>>
    where
        F: FnMut(Self::Token) -> bool,
    {
        self.0.uncons_while1(f)
    }

    #[inline]
    fn distance(&self, end: &Self::Checkpoint) -> usize {
        self.0.distance(end)
    }

    #[inline]
    fn range(&self) -> Self::Range {
        self.0.range()
    }
}

/// Stream type which indicates that the stream is complete if end of input is reached
///
/// For most streams this is already the default but this wrapper can be used to override a nested
/// `PartialStream`
#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Debug)]
#[repr(transparent)]
pub struct CompleteStream<S>(pub S);

impl<S> From<S> for CompleteStream<S> {
    fn from(t: S) -> Self {
        CompleteStream(t)
    }
}

impl<'s, S> From<&'s mut S> for &'s mut CompleteStream<S> {
    fn from(t: &'s mut S) -> Self {
        // SAFETY repr(transparent) is specified on CompleteStream
        unsafe { mem::transmute(t) }
    }
}

impl<S> Positioned for CompleteStream<S>
where
    S: Positioned,
{
    #[inline]
    fn position(&self) -> Self::Position {
        self.0.position()
    }
}

impl<S> ResetStream for CompleteStream<S>
where
    S: ResetStream,
{
    type Checkpoint = S::Checkpoint;

    #[inline]
    fn checkpoint(&self) -> Self::Checkpoint {
        self.0.checkpoint()
    }

    #[inline]
    fn reset(&mut self, checkpoint: Self::Checkpoint) -> Result<(), S::Error> {
        self.0.reset(checkpoint)
    }
}

impl<S> StreamOnce for CompleteStream<S>
where
    S: StreamOnce,
{
    type Token = S::Token;
    type Range = S::Range;
    type Position = S::Position;
    type Error = S::Error;

    #[inline]
    fn uncons(&mut self) -> Result<S::Token, StreamErrorFor<Self>> {
        self.0.uncons()
    }

    fn is_partial(&self) -> bool {
        false
    }
}

impl<S> RangeStreamOnce for CompleteStream<S>
where
    S: RangeStreamOnce,
{
    #[inline]
    fn uncons_range(&mut self, size: usize) -> Result<Self::Range, StreamErrorFor<Self>> {
        self.0.uncons_range(size)
    }

    #[inline]
    fn uncons_while<F>(&mut self, f: F) -> Result<Self::Range, StreamErrorFor<Self>>
    where
        F: FnMut(Self::Token) -> bool,
    {
        self.0.uncons_while(f)
    }

    fn uncons_while1<F>(&mut self, f: F) -> ParseResult<Self::Range, StreamErrorFor<Self>>
    where
        F: FnMut(Self::Token) -> bool,
    {
        self.0.uncons_while1(f)
    }

    #[inline]
    fn distance(&self, end: &Self::Checkpoint) -> usize {
        self.0.distance(end)
    }

    #[inline]
    fn range(&self) -> Self::Range {
        self.0.range()
    }
}

#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Debug)]
pub struct MaybePartialStream<S>(pub S, pub bool);

impl<S> Positioned for MaybePartialStream<S>
where
    S: Positioned,
{
    #[inline]
    fn position(&self) -> Self::Position {
        self.0.position()
    }
}

impl<S> ResetStream for MaybePartialStream<S>
where
    S: ResetStream,
{
    type Checkpoint = S::Checkpoint;

    #[inline]
    fn checkpoint(&self) -> Self::Checkpoint {
        self.0.checkpoint()
    }

    #[inline]
    fn reset(&mut self, checkpoint: Self::Checkpoint) -> Result<(), S::Error> {
        self.0.reset(checkpoint)
    }
}

impl<S> StreamOnce for MaybePartialStream<S>
where
    S: StreamOnce,
{
    type Token = S::Token;
    type Range = S::Range;
    type Position = S::Position;
    type Error = S::Error;

    #[inline]
    fn uncons(&mut self) -> Result<S::Token, StreamErrorFor<Self>> {
        self.0.uncons()
    }

    fn is_partial(&self) -> bool {
        self.1
    }
}

impl<S> RangeStreamOnce for MaybePartialStream<S>
where
    S: RangeStreamOnce,
{
    #[inline]
    fn uncons_range(&mut self, size: usize) -> Result<Self::Range, StreamErrorFor<Self>> {
        self.0.uncons_range(size)
    }

    #[inline]
    fn uncons_while<F>(&mut self, f: F) -> Result<Self::Range, StreamErrorFor<Self>>
    where
        F: FnMut(Self::Token) -> bool,
    {
        self.0.uncons_while(f)
    }

    fn uncons_while1<F>(&mut self, f: F) -> ParseResult<Self::Range, StreamErrorFor<Self>>
    where
        F: FnMut(Self::Token) -> bool,
    {
        self.0.uncons_while1(f)
    }

    #[inline]
    fn distance(&self, end: &Self::Checkpoint) -> usize {
        self.0.distance(end)
    }

    #[inline]
    fn range(&self) -> Self::Range {
        self.0.range()
    }
}

/// Newtype for constructing a stream from a slice where the items in the slice are not copyable.
#[derive(Copy, Eq, PartialEq, Ord, PartialOrd, Debug)]
pub struct SliceStream<'a, T: 'a>(pub &'a [T]);

impl<'a, T> Clone for SliceStream<'a, T> {
    fn clone(&self) -> SliceStream<'a, T> {
        SliceStream(self.0)
    }
}

impl<'a, T> Positioned for SliceStream<'a, T>
where
    T: PartialEq + 'a,
{
    #[inline]
    fn position(&self) -> Self::Position {
        PointerOffset::new(self.0.as_ptr() as usize)
    }
}

impl<'a, T> StreamOnce for SliceStream<'a, T>
where
    T: PartialEq + 'a,
{
    type Token = &'a T;
    type Range = &'a [T];
    type Position = PointerOffset<[T]>;
    type Error = UnexpectedParse;

    #[inline]
    fn uncons(&mut self) -> Result<&'a T, StreamErrorFor<Self>> {
        match self.0.split_first() {
            Some((first, rest)) => {
                self.0 = rest;
                Ok(first)
            }
            None => Err(UnexpectedParse::Eoi),
        }
    }
}

fn slice_uncons_while_ref<'a, T, F>(slice: &mut &'a [T], mut i: usize, mut f: F) -> &'a [T]
where
    F: FnMut(&'a T) -> bool,
{
    let len = slice.len();
    let mut found = false;

    macro_rules! check {
        () => {
            if !f(unsafe { slice.get_unchecked(i) }) {
                found = true;
                break;
            }
            i += 1;
        };
    }

    while len - i >= 8 {
        check!();
        check!();
        check!();
        check!();
        check!();
        check!();
        check!();
        check!();
    }

    if !found {
        while i < len {
            if !f(unsafe { slice.get_unchecked(i) }) {
                break;
            }
            i += 1;
        }
    }

    let (result, remaining) = slice.split_at(i);
    *slice = remaining;
    result
}

impl<'a, T> RangeStreamOnce for SliceStream<'a, T>
where
    T: PartialEq + 'a,
{
    #[inline]
    fn uncons_range(&mut self, size: usize) -> Result<&'a [T], StreamErrorFor<Self>> {
        if size <= self.0.len() {
            let (range, rest) = self.0.split_at(size);
            self.0 = rest;
            Ok(range)
        } else {
            Err(UnexpectedParse::Eoi)
        }
    }

    #[inline]
    fn uncons_while<F>(&mut self, f: F) -> Result<&'a [T], StreamErrorFor<Self>>
    where
        F: FnMut(Self::Token) -> bool,
    {
        Ok(slice_uncons_while_ref(&mut self.0, 0, f))
    }

    #[inline]
    fn uncons_while1<F>(&mut self, mut f: F) -> ParseResult<Self::Range, StreamErrorFor<Self>>
    where
        F: FnMut(Self::Token) -> bool,
    {
        if self.0.is_empty() || !f(unsafe { self.0.get_unchecked(0) }) {
            return PeekErr(Tracked::from(UnexpectedParse::Unexpected));
        }

        CommitOk(slice_uncons_while_ref(&mut self.0, 1, f))
    }

    #[inline]
    fn distance(&self, end: &Self) -> usize {
        end.0.len() - self.0.len()
    }

    fn range(&self) -> Self::Range {
        self.0
    }
}

/// Wrapper around iterators which allows them to be treated as a stream.
/// Returned by [`from_iter`].
///
/// [`from_iter`]: fn.from_iter.html
#[derive(Copy, Clone, Debug)]
pub struct IteratorStream<Input>(Input);

impl<Input> IteratorStream<Input>
where
    Input: Iterator,
{
    /// Converts an `Iterator` into a stream.
    ///
    /// NOTE: This type do not implement `Positioned` and `Clone` and must be wrapped with types
    ///     such as `BufferedStreamRef` and `State` to become a `Stream` which can be parsed
    pub fn new<T>(iter: T) -> IteratorStream<Input>
    where
        T: IntoIterator<IntoIter = Input, Item = Input::Item>,
    {
        IteratorStream(iter.into_iter())
    }
}

impl<Input> Iterator for IteratorStream<Input>
where
    Input: Iterator,
{
    type Item = Input::Item;
    fn next(&mut self) -> Option<Input::Item> {
        self.0.next()
    }
}

impl<Input: Iterator> StreamOnce for IteratorStream<Input>
where
    Input::Item: Clone + PartialEq,
{
    type Token = Input::Item;
    type Range = Input::Item;
    type Position = ();
    type Error = UnexpectedParse;

    #[inline]
    fn uncons(&mut self) -> Result<Self::Token, StreamErrorFor<Self>> {
        match self.next() {
            Some(x) => Ok(x),
            None => Err(UnexpectedParse::Eoi),
        }
    }
}

/// Newtype around a pointer offset into a slice stream (`&[T]`/`&str`).
pub struct PointerOffset<T: ?Sized>(pub usize, PhantomData<T>);

impl<T: ?Sized> Clone for PointerOffset<T> {
    fn clone(&self) -> Self {
        PointerOffset::new(self.0)
    }
}

impl<T: ?Sized> Copy for PointerOffset<T> {}

impl<T: ?Sized> Default for PointerOffset<T> {
    fn default() -> Self {
        PointerOffset::new(0)
    }
}

impl<T: ?Sized> PartialEq for PointerOffset<T> {
    fn eq(&self, other: &Self) -> bool {
        self.0 == other.0
    }
}

impl<T: ?Sized> Eq for PointerOffset<T> {}

impl<T: ?Sized> PartialOrd for PointerOffset<T> {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        self.0.partial_cmp(&other.0)
    }
}

impl<T: ?Sized> Ord for PointerOffset<T> {
    fn cmp(&self, other: &Self) -> Ordering {
        self.0.cmp(&other.0)
    }
}

impl<T> fmt::Debug for PointerOffset<T>
where
    T: ?Sized,
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self)
    }
}

impl<T> fmt::Display for PointerOffset<T>
where
    T: ?Sized,
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "PointerOffset({:?})", self.0 as *const ())
    }
}

impl<T> PointerOffset<T>
where
    T: ?Sized,
{
    pub fn new(offset: usize) -> Self {
        PointerOffset(offset, PhantomData)
    }

    /// Converts the pointer-based position into an indexed position.
    ///
    /// ```rust
    /// # extern crate combine;
    /// # use combine::*;
    /// # fn main() {
    /// let text = "b";
    /// let err = token('a').easy_parse(text).unwrap_err();
    /// assert_eq!(err.position.0, text.as_ptr() as usize);
    /// assert_eq!(err.map_position(|p| p.translate_position(text)).position, 0);
    /// # }
    /// ```
    pub fn translate_position(mut self, initial_slice: &T) -> usize {
        self.0 -= initial_slice as *const T as *const () as usize;
        self.0
    }
}

/// Decodes `input` using `parser`.
///
/// Return `Ok(Some(token), consumed_data)` if there was enough data to finish parsing using
/// `parser`.
/// Returns `Ok(None, consumed_data)` if `input` did not contain enough data to finish parsing
/// using `parser`.
///
/// See `examples/async.rs` for example usage in a `tokio_io::codec::Decoder`
pub fn decode<Input, P>(
    mut parser: P,
    mut input: &mut Input,
    partial_state: &mut P::PartialState,
) -> Result<(Option<P::Output>, usize), <Input as StreamOnce>::Error>
where
    P: Parser<Input>,
    Input: RangeStream,
{
    let start = input.checkpoint();
    match parser.parse_with_state(&mut input, partial_state) {
        Ok(message) => Ok((Some(message), input.distance(&start))),
        Err(err) => {
            if input.is_partial() && err.is_unexpected_end_of_input() {
                Ok((None, input.distance(&start)))
            } else {
                Err(err)
            }
        }
    }
}

#[cfg(feature = "std")]
use crate::stream::position::IndexPositioner;

#[cfg(feature = "std")]
use std::io::{self, BufRead};

/// Used together with the `decode!` macro
#[cfg(feature = "std")]
#[cfg_attr(docsrs, doc(cfg(feature = "std")))]
#[cfg_attr(feature = "pin-project", pin_project::pin_project)]
pub struct Decoder<R, S, P = IndexPositioner> {
    #[cfg_attr(feature = "pin-project", pin)]
    reader: R,
    pub positioner: P,
    state: S,
    remaining: Vec<u8>,
}

#[cfg(feature = "std")]
#[cfg_attr(docsrs, doc(cfg(feature = "std")))]
impl<R, S, P> Decoder<R, S, P>
where
    P: Default,
    S: Default,
{
    pub fn new(reader: R) -> Self {
        Decoder {
            reader,
            positioner: P::default(),
            state: S::default(),
            remaining: Vec::new(),
        }
    }
}

#[cfg(feature = "std")]
impl<R, S, P> Decoder<R, S, P> {
    #[doc(hidden)]
    pub fn remaining_len(&self) -> usize {
        self.remaining.len()
    }
}

#[cfg(feature = "std")]
impl<R, S, P> Decoder<R, S, P>
where
    R: BufRead,
{
    #[doc(hidden)]
    pub fn before_parse(&mut self) -> io::Result<(&mut S, &mut P, &[u8], bool)> {
        let mut end_of_input = false;
        let buffer = self.reader.fill_buf()?;
        if buffer.len() == 0 {
            end_of_input = true;
        }
        Ok((
            &mut self.state,
            &mut self.positioner,
            if !self.remaining.is_empty() {
                self.remaining.extend(buffer);
                &self.remaining[..]
            } else {
                buffer
            },
            end_of_input,
        ))
    }

    #[doc(hidden)]
    pub fn after_parse(
        &mut self,
        remaining_data: usize,
        mut removed: usize,
        done: bool,
    ) -> io::Result<()> {
        if !self.remaining.is_empty() {
            // Remove the data we have parsed and adjust `removed` to be the amount of data we
            // consumed from `self.reader`
            self.remaining.drain(..removed);
            if removed >= remaining_data {
                removed -= remaining_data;
            } else {
                removed = 0;
            }
        }

        let consume = if done {
            removed
        } else {
            // We have not enough data to produce a Value but we know that all the data of
            // the current buffer are necessary. Consume all the buffered data to ensure
            // that the next iteration actually reads more data.
            let buffer = self.reader.fill_buf()?;
            if remaining_data == 0 {
                self.remaining.extend(&buffer[removed..]);
            }
            buffer.len()
        };
        self.reader.consume(consume);
        Ok(())
    }
}

#[cfg(feature = "tokio-02")]
use std::pin::Pin;

// https://github.com/tokio-rs/tokio/pull/1687
#[cfg(feature = "tokio-02")]
async fn fill_buf<R>(reader: Pin<&mut R>) -> io::Result<&[u8]>
where
    R: tokio_02_dep::io::AsyncBufRead,
{
    let mut reader = Some(reader);
    use futures_03::{future, task::Poll};
    future::poll_fn(move |cx| match reader.take() {
        Some(mut r) => match r.as_mut().poll_fill_buf(cx) {
            // SAFETY We either drop `self.reader` and return a slice with the lifetime of the
            // reader or we return Pending/Err (neither which contains `'a`).
            // In either case `poll_fill_buf` can not be called while it's contents are exposed
            Poll::Ready(Ok(x)) => unsafe { return Ok(&*(x as *const _)).into() },
            Poll::Ready(Err(err)) => Err(err).into(),
            Poll::Pending => {
                reader = Some(r);
                Poll::Pending
            }
        },
        None => panic!("fill_buf polled after completion"),
    })
    .await
}

#[cfg(feature = "tokio-02")]
impl<R, S, P> Decoder<R, S, P>
where
    R: tokio_02_dep::io::AsyncBufRead,
{
    #[doc(hidden)]
    pub async fn before_parse_async(
        self: Pin<&mut Self>,
    ) -> io::Result<(&mut S, &mut P, &[u8], bool)> {
        let self_ = self.project();
        let mut end_of_input = false;
        let buffer = fill_buf(self_.reader).await?;
        if buffer.len() == 0 {
            end_of_input = true;
        }
        Ok((
            self_.state,
            self_.positioner,
            if !self_.remaining.is_empty() {
                self_.remaining.extend(buffer);
                &self_.remaining[..]
            } else {
                buffer
            },
            end_of_input,
        ))
    }

    #[doc(hidden)]
    pub async fn after_parse_async(
        self: Pin<&mut Self>,
        remaining_data: usize,
        mut removed: usize,
        done: bool,
    ) -> io::Result<()> {
        let mut self_ = self.project();
        if !self_.remaining.is_empty() {
            // Remove the data we have parsed and adjust `removed` to be the amount of data we
            // consumed from `self.reader`
            self_.remaining.drain(..removed);
            if removed >= remaining_data {
                removed -= remaining_data;
            } else {
                removed = 0;
            }
        }

        let consume = if done {
            removed
        } else {
            // We have not enough data to produce a Value but we know that all the data of
            // the current buffer are necessary. Consume all the buffered data to ensure
            // that the next iteration actually reads more data.
            let buffer = fill_buf(self_.reader.as_mut()).await?;
            if remaining_data == 0 {
                self_.remaining.extend(&buffer[removed..]);
            }
            buffer.len()
        };
        self_.reader.consume(consume);
        Ok(())
    }
}

/// Parses an instance of `std::io::BufRead` as a `&[u8]` without reading the entire file into
/// memory.
///
/// This is defined as a macro to work around the lack of Higher Ranked Types. See the
/// example for how to pass a parser to the macro (constructing parts of the parser outside of
/// the `decode_buf_read!` call is unlikely to work.
///
/// ```
/// use std::{
///     fs::File,
///     io::BufReader
/// };
/// use combine::{decode_buf_read, satisfy, skip_many1, many1, sep_end_by, Parser, stream::Decoder};
///
/// let mut decoder = Decoder::<_, _>::new(BufReader::with_capacity(100, File::open("README.md").unwrap()));
/// let is_whitespace = |b: u8| b == b' ' || b == b'\r' || b == b'\n';
/// assert_eq!(
///     decode_buf_read!(
///         decoder,
///         {
///             let word = many1(satisfy(|b| !is_whitespace(b)));
///             sep_end_by(word, skip_many1(satisfy(is_whitespace))).map(|words: Vec<Vec<u8>>| words.len())
///         },
///         combine::easy::Stream::from
///     ).map_err(|err: combine::easy::Errors<u8, &[u8], usize>| err),
///     Ok(819),
/// );
/// ```
#[cfg(feature = "std")]
#[cfg_attr(docsrs, doc(cfg(feature = "std")))]
#[macro_export]
macro_rules! decode_buf_read {
    ($decoder: expr, $parser: expr) => {
        $crate::decode_buf_read!($decoder, $parser, |x| x)
    };

    ($decoder: expr, $parser: expr, $input_stream: expr) => {
        match $decoder {
            ref mut decoder => 'outer: loop {
                use $crate::stream::position::Positioner;

                let remaining_data = decoder.remaining_len();

                let (opt, removed) = {
                    let (state, positioner, buffer, end_of_input) = match decoder.before_parse() {
                        Ok(x) => x,
                        Err(err) => {
                            break 'outer Err($crate::error::ParseError::from_error(
                                Positioner::<u8>::position(&decoder.positioner),
                                $crate::error::StreamError::other(err),
                            ))
                        }
                    };

                    let mut stream = $crate::stream::MaybePartialStream(
                        $input_stream($crate::stream::position::Stream {
                            input: buffer,
                            positioner,
                        }),
                        !end_of_input,
                    );
                    match $crate::stream::decode($parser, &mut stream, state) {
                        Ok(x) => x,
                        Err(err) => break 'outer Err(err.into()),
                    }
                };

                match decoder.after_parse(remaining_data, removed, opt.is_some()) {
                    Ok(x) => x,
                    Err(err) => {
                        break 'outer Err($crate::error::ParseError::from_error(
                            Positioner::<u8>::position(&decoder.positioner),
                            $crate::error::StreamError::other(err),
                        ))
                    }
                }

                if let Some(v) = opt {
                    break 'outer Ok(v);
                }
            },
        }
    };
}

/// Parses an instance of `tokio::io::BufRead` as a `&[u8]` without reading the entire file into
/// memory.
///
/// This is defined as a macro to work around the lack of Higher Ranked Types. See the
/// example for how to pass a parser to the macro (constructing parts of the parser outside of
/// the `decode_buf_read!` call is unlikely to work.
///
/// ```
/// # use tokio_02_dep as tokio;
/// use futures_03::pin_mut;
/// use tokio::{
///     fs::File,
///     io::BufReader
/// };
///
/// use combine::{decode_tokio_buf_read, satisfy, skip_many1, many1, sep_end_by, Parser, stream::Decoder};
///
/// #[tokio::main]
/// async fn main() {
///     let mut decoder = Decoder::<_, _>::new(BufReader::with_capacity(100, File::open("README.md").await.unwrap()));
///     pin_mut!(decoder);
///     let is_whitespace = |b: u8| b == b' ' || b == b'\r' || b == b'\n';
///     assert_eq!(
///         decode_tokio_buf_read!(
///             decoder,
///             {
///                 let word = many1(satisfy(|b| !is_whitespace(b)));
///                 sep_end_by(word, skip_many1(satisfy(is_whitespace))).map(|words: Vec<Vec<u8>>| words.len())
///             },
///             combine::easy::Stream::from
///         ).map_err(|err: combine::easy::Errors<u8, &[u8], usize>| err),
///         Ok(819),
///     );
/// }
/// ```
#[cfg(feature = "tokio-02")]
#[cfg_attr(docsrs, doc(cfg(feature = "tokio-02")))]
#[macro_export]
macro_rules! decode_tokio_buf_read {
    ($decoder: expr, $parser: expr) => {
        $crate::decode_tokio_buf_read!(async $decoder, $parser, |x| x)
    };

    ($decoder: expr, $parser: expr, $input_stream: expr) => {
        match $decoder {
            ref mut decoder => 'outer: loop {
                use $crate::stream::position::Positioner;

                let remaining_data = decoder.remaining_len();

                let (opt, removed) = {
                    let (state, positioner, buffer, end_of_input) =
                        match decoder.as_mut().before_parse_async().await {
                            Ok(x) => x,
                            Err(err) => {
                                break 'outer Err($crate::error::ParseError::from_error(
                                    Positioner::<u8>::position(&decoder.positioner),
                                    $crate::error::StreamError::other(err),
                                ))
                            }
                        };

                    let mut stream = $crate::stream::MaybePartialStream(
                        $input_stream($crate::stream::position::Stream {
                            input: buffer,
                            positioner,
                        }),
                        !end_of_input,
                    );
                    match $crate::stream::decode($parser, &mut stream, state) {
                        Ok(x) => x,
                        Err(err) => break 'outer Err(err.into()),
                    }
                };

                match decoder
                    .as_mut()
                    .after_parse_async(remaining_data, removed, opt.is_some())
                    .await
                {
                    Ok(x) => x,
                    Err(err) => {
                        break 'outer Err($crate::error::ParseError::from_error(
                            Positioner::<u8>::position(&decoder.positioner),
                            $crate::error::StreamError::other(err),
                        ))
                    }
                }

                if let Some(v) = opt {
                    break 'outer Ok(v);
                }
            },
        }
    };
}

#[cfg(test)]
mod tests {

    use super::*;

    #[test]
    #[inline]
    fn uncons_range_at_end() {
        assert_eq!("".uncons_range(0), Ok(""));
        assert_eq!("123".uncons_range(3), Ok("123"));
        assert_eq!((&[1][..]).uncons_range(1), Ok(&[1][..]));
        let s: &[u8] = &[];
        assert_eq!(SliceStream(s).uncons_range(0), Ok(&[][..]));
    }

    #[test]
    fn larger_than_1_byte_items_return_correct_distance() {
        let mut input = &[123i32, 0i32][..];

        let before = input.checkpoint();
        assert_eq!(input.distance(&before), 0);

        input.uncons().unwrap();
        assert_eq!(input.distance(&before), 1);

        input.uncons().unwrap();
        assert_eq!(input.distance(&before), 2);

        input.reset(before.clone()).unwrap();
        assert_eq!(input.distance(&before), 0);
    }
}