content: {-# LANGUAGE CPP #-}
  {-# LANGUAGE PatternSynonyms #-}
  {-# LANGUAGE TypeFamilies #-}
  -- Must come after TypeFamilies, otherwise it is re-enabled.
  -- MonoLocalBinds enabled by TypeFamilies causes perf regressions in general.
  {-# LANGUAGE NoMonoLocalBinds #-}
  {-# LANGUAGE UndecidableInstances #-}
  {-# LANGUAGE ViewPatterns #-}
  
  -- |
  -- Module      : Streamly.Internal.Data.Stream.Type
  -- Copyright   : (c) 2018 Composewell Technologies
  --               (c) Roman Leshchinskiy 2008-2010
  -- License     : BSD-3-Clause
  -- Maintainer  : streamly@composewell.com
  -- Stability   : experimental
  -- Portability : GHC
  
  -- The stream type is inspired by the vector package.  A few functions in this
  -- module have been originally adapted from the vector package (c) Roman
  -- Leshchinskiy. See the notes in specific functions.
  
  module Streamly.Internal.Data.Stream.Type
      (
      -- * Type
        Step (..)
      -- XXX UnStream is exported to avoid a performance issue in some
      -- combinators if we use the pattern synonym "Stream".
      , Stream (Stream, UnStream)
  
      -- * Nested
      , Nested(..)
  
      -- * To StreamK
      , fromStreamK
      , toStreamK
  
      -- * From Unfold
      , unfold
  
      -- * Construction
      -- ** Primitives
      , nilM
      , consM
  
      -- ** From Values
      , fromPure
      , fromEffect
  
      -- ** From Containers
      , Streamly.Internal.Data.Stream.Type.fromList
  
      -- * Elimination
      -- ** Primitives
      , uncons
  
      -- ** Strict Left Folds
      , Streamly.Internal.Data.Stream.Type.fold
      , foldBreak
      , foldAddLazy
      , foldAdd
      , foldBreakEither
  
      , Streamly.Internal.Data.Stream.Type.foldl'
      , foldlM'
      , foldlx'
      , foldlMx'
  
      -- ** Lazy Right Folds
      , foldrM
      , foldrMx
      , Streamly.Internal.Data.Stream.Type.foldr
      , foldrS
  
      -- ** Specific Folds
      , drain
      , head
      , headElse
      , Streamly.Internal.Data.Stream.Type.toList
  
      -- * Mapping
      , map
      , mapM
  
      -- * Stateful Filters
      , take
      , takeWhile
      , takeWhileM
      , takeEndBy_
      , takeEndBy
      , takeEndByM
  
      -- * Combining Two Streams
      -- ** Appending
      -- | Append a stream after another. A special case of concatMap or
      -- unfoldEach Note, appending more than two streams is called @concat@
      -- which could be called appendMany or appendAll in append terminology and
      -- is equivalent to @concatMap id@. Append is equivalent to @mergeBy fst@.
      , AppendState(..)
      , append
  
      -- ** Zipping
      -- | Zip corresponding elements of two streams.
      , zipWithM
      , zipWith
  
      -- ** Cross Product
      , crossApply
      , crossApplyFst
      , crossApplySnd
      , crossWith
      , cross
      , FairUnfoldState (..)
      , fairCrossWithM
      , fairCrossWith
      , fairCross
      , loop -- forEach
      , loopBy
  
      -- * Unfold Many
      , UnfoldEachState (..)
      , unfoldEach
  
      -- * UnfoldCross
      , unfoldCross
  
      -- * ConcatMap
      -- | Generate streams by mapping a stream generator on each element of an
      -- input stream, append the resulting streams and flatten.
      , concatEffect
      , concatMap
      , concatMapM
      , concat
  
      -- * ConcatFor
      , concatFor
      , concatForM
  
      -- * Unfold Iterate
      , unfoldIterate
      , bfsUnfoldIterate
      , altBfsUnfoldIterate
  
      -- * Concat Iterate
      , concatIterateScan
      , concatIterate
      , bfsConcatIterate
      , altBfsConcatIterate
  
      -- * Fold Many
      , FoldMany (..) -- for inspection testing
      , FoldManyPost (..)
      , foldMany
      , foldManyPost
      , foldManySepBy
      , groupsOf
      , refoldMany
      , refoldIterateM
  
      -- * Fold Iterate
      , bfsReduceIterate
      , bfsFoldIterate
  
      -- * Splitting
      , indexEndBy
      , indexEndBy_
  
      -- * Multi-stream folds
      -- | These should probably be expressed using zipping operations.
      , eqBy
      , cmpBy
  
      -- * Utilities
      , splitAt
  
      -- * Deprecated
      , sliceOnSuffix
      , unfoldMany
      , indexOnSuffix
      , CrossStream
      , mkCross
      , unCross
      , reduceIterateBfs
      , unfoldIterateDfs
      , unfoldIterateBfs
      , unfoldIterateBfsRev
      , concatIterateDfs
      , concatIterateBfs
      , concatIterateBfsRev
      , foldEither -- XXX to be marked as deprecated
      )
  where
  
  #include "deprecation.h"
  #include "inline.hs"
  
  #if !MIN_VERSION_base(4,18,0)
  import Control.Applicative (liftA2)
  #endif
  import Control.Monad.Catch (MonadThrow, throwM)
  import Control.Monad.Trans.Class (MonadTrans(lift))
  import Control.Monad.IO.Class (MonadIO(..))
  import Data.Bifunctor (first)
  import Data.Foldable (Foldable(foldl'), fold, foldr)
  import Data.Functor (($>))
  import Data.Functor.Identity (Identity(..))
  #if __GLASGOW_HASKELL__ >= 810
  import Data.Kind (Type)
  #endif
  import Data.Maybe (fromMaybe)
  import Data.Semigroup (Endo(..))
  import Fusion.Plugin.Types (Fuse(..))
  import GHC.Base (build)
  import GHC.Exts (IsList(..), IsString(..), oneShot)
  import GHC.Types (SPEC(..))
  import Prelude hiding
      (head, map, mapM, take, concatMap, takeWhile, zipWith, concat, splitAt)
  import Text.Read
         ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec
         , readListPrecDefault)
  
  import Streamly.Internal.BaseCompat ((#.))
  import Streamly.Internal.Data.Fold.Type (Fold(..))
  import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
  import Streamly.Internal.Data.Refold.Type (Refold(..))
  import Streamly.Internal.Data.Stream.Step (Step (..))
  import Streamly.Internal.Data.SVar.Type (State, adaptState, defState)
  import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
  import Streamly.Internal.Data.Unfold.Type (Unfold(..))
  
  import qualified Streamly.Internal.Data.Fold.Type as FL hiding (foldr)
  import qualified Streamly.Internal.Data.StreamK.Type as K
  import qualified Streamly.Internal.Data.Unfold.Type as Unfold
  
  #include "DocTestDataStream.hs"
  
  ------------------------------------------------------------------------------
  -- The direct style stream type
  ------------------------------------------------------------------------------
  
  -- gst = global state
  
  -- | A stream consists of a step function that generates the next step given a
  -- current state, and the current state.
  data Stream m a =
      forall s. UnStream (State K.StreamK m a -> s -> m (Step s a)) s
  
  -- XXX This causes perf trouble when pattern matching with "Stream"  in a
  -- recursive way, e.g. in uncons, foldBreak, concatMap. We need to get rid of
  -- this.
  unShare :: Stream m a -> Stream m a
  unShare (UnStream step state) = UnStream step' state
      where step' gst = step (adaptState gst)
  
  pattern Stream :: (State K.StreamK m a -> s -> m (Step s a)) -> s -> Stream m a
  pattern Stream step state <- (unShare -> UnStream step state)
      where Stream = UnStream
  
  {-# COMPLETE Stream #-}
  
  ------------------------------------------------------------------------------
  -- Primitives
  ------------------------------------------------------------------------------
  
  -- | A stream that terminates without producing any output, but produces a side
  -- effect.
  --
  -- >>> nilM action = Stream.before action Stream.nil
  -- >>> Stream.fold Fold.toList (Stream.nilM (print "nil"))
  -- "nil"
  -- []
  --
  -- /Pre-release/
  {-# INLINE_NORMAL nilM #-}
  nilM :: Applicative m => m b -> Stream m a
  nilM m = Stream (\_ _ -> m $> Stop) ()
  
  infixr 5 `consM`
  
  -- XXX see https://github.com/composewell/streamly/issues/3126 - for using a
  -- list or maybe an effect "m (Stream m a)" along with "step" in the stream
  -- structure for better cons and append performance.
  
  -- | Like 'cons' but fuses an effect instead of a pure value.
  {-# INLINE_NORMAL consM #-}
  consM :: Applicative m => m a -> Stream m a -> Stream m a
  consM m (Stream step state) = Stream step1 Nothing
  
      where
  
      {-# INLINE_LATE step1 #-}
      step1 _ Nothing = (`Yield` Just state) <$> m
      step1 gst (Just st) = do
            (\case
              Yield a s -> Yield a (Just s)
              Skip  s   -> Skip (Just s)
              Stop      -> Stop) <$> step gst st
  
  -- | Decompose a stream into its head and tail. If the stream is empty, returns
  -- 'Nothing'. If the stream is non-empty, returns @Just (a, ma)@, where @a@ is
  -- the head of the stream and @ma@ its tail.
  --
  -- Properties:
  --
  -- >>> Nothing <- Stream.uncons Stream.nil
  -- >>> Just ("a", t) <- Stream.uncons (Stream.cons "a" Stream.nil)
  --
  -- This can be used to consume the stream in an imperative manner one element
  -- at a time, as it just breaks down the stream into individual elements and we
  -- can loop over them as we deem fit. For example, this can be used to convert
  -- a streamly stream into other stream types.
  --
  -- All the folds in this module can be expressed in terms of 'uncons', however,
  -- this is generally less efficient than specific folds because it takes apart
  -- the stream one element at a time, therefore, does not take adavantage of
  -- stream fusion.
  --
  -- 'foldBreak' is a more general way of consuming a stream piecemeal.
  --
  -- >>> :{
  -- uncons xs = do
  --     r <- Stream.foldBreak Fold.one xs
  --     return $ case r of
  --         (Nothing, _) -> Nothing
  --         (Just h, t) -> Just (h, t)
  -- :}
  --
  {-# INLINE_NORMAL uncons #-}
  uncons :: Monad m => Stream m a -> m (Maybe (a, Stream m a))
  uncons (UnStream step state) = go SPEC state
    where
      go !_ st = do
          r <- step defState st
          case r of
              Yield x s -> return $ Just (x, Stream step s)
              Skip  s   -> go SPEC s
              Stop      -> return Nothing
  
  ------------------------------------------------------------------------------
  -- From 'Unfold'
  ------------------------------------------------------------------------------
  
  data UnfoldState s = UnfoldNothing | UnfoldJust s
  
  -- | Convert an 'Unfold' into a stream by supplying it an input seed.
  --
  -- >>> s = Stream.unfold Unfold.replicateM (3, putStrLn "hello")
  -- >>> Stream.fold Fold.drain s
  -- hello
  -- hello
  -- hello
  --
  {-# INLINE_NORMAL unfold #-}
  unfold :: Applicative m => Unfold m a b -> a -> Stream m b
  unfold (Unfold ustep inject) seed = Stream step UnfoldNothing
  
      where
  
      {-# INLINE_LATE step #-}
      step _ UnfoldNothing = Skip . UnfoldJust <$> inject seed
      step _ (UnfoldJust st) = do
          (\case
              Yield x s -> Yield x (UnfoldJust s)
              Skip s    -> Skip (UnfoldJust s)
              Stop      -> Stop) <$> ustep st
  
  ------------------------------------------------------------------------------
  -- From Values
  ------------------------------------------------------------------------------
  
  -- | Create a singleton stream from a pure value.
  --
  -- >>> fromPure a = a `Stream.cons` Stream.nil
  -- >>> fromPure = pure
  -- >>> fromPure = Stream.fromEffect . pure
  --
  {-# INLINE_NORMAL fromPure #-}
  fromPure :: Applicative m => a -> Stream m a
  fromPure x = Stream (\_ s -> pure $ step undefined s) True
    where
      {-# INLINE_LATE step #-}
      step _ True  = Yield x False
      step _ False = Stop
  
  -- | Create a singleton stream from a monadic action.
  --
  -- >>> fromEffect m = m `Stream.consM` Stream.nil
  -- >>> fromEffect = Stream.sequence . Stream.fromPure
  --
  -- >>> Stream.fold Fold.drain $ Stream.fromEffect (putStrLn "hello")
  -- hello
  --
  {-# INLINE_NORMAL fromEffect #-}
  fromEffect :: Applicative m => m a -> Stream m a
  fromEffect m = Stream step True
  
      where
  
      {-# INLINE_LATE step #-}
      step _ True  = (`Yield` False) <$> m
      step _ False = pure Stop
  
  ------------------------------------------------------------------------------
  -- From Containers
  ------------------------------------------------------------------------------
  
  -- Adapted from the vector package.
  
  -- | Construct a stream from a list of pure values.
  {-# INLINE_LATE fromList #-}
  fromList :: Applicative m => [a] -> Stream m a
  #ifdef USE_UNFOLDS_EVERYWHERE
  fromList = unfold Unfold.fromList
  #else
  fromList = Stream step
    where
      {-# INLINE_LATE step #-}
      step _ (x:xs) = pure $ Yield x xs
      step _ []     = pure Stop
  #endif
  
  ------------------------------------------------------------------------------
  -- Conversions From/To
  ------------------------------------------------------------------------------
  
  -- | Convert a CPS encoded StreamK to direct style step encoded StreamD
  {-# INLINE_LATE fromStreamK #-}
  fromStreamK :: Applicative m => K.StreamK m a -> Stream m a
  fromStreamK = Stream step
      where
      step gst m1 =
          let stop       = pure Stop
              single a   = pure $ Yield a K.nil
              yieldk a r = pure $ Yield a r
           in K.foldStreamShared gst yieldk single stop m1
  
  -- | Convert a direct style step encoded StreamD to a CPS encoded StreamK
  {-# INLINE_LATE toStreamK #-}
  toStreamK :: Monad m => Stream m a -> K.StreamK m a
  toStreamK (Stream step state) = go state
      where
      go st = K.MkStream $ \gst yld _ stp ->
        let go' ss = do
             r <- step gst ss
             case r of
                 Yield x s -> yld x (go s)
                 Skip  s   -> go' s
                 Stop      -> stp
        in go' st
  
  {-# RULES "fromStreamK/toStreamK fusion"
      forall s. toStreamK (fromStreamK s) = s #-}
  {-# RULES "toStreamK/fromStreamK fusion"
      forall s. fromStreamK (toStreamK s) = s #-}
  
  ------------------------------------------------------------------------------
  -- Running a 'Fold'
  ------------------------------------------------------------------------------
  
  -- | Fold resulting in either breaking the stream or continuation of the fold.
  -- Instead of supplying the input stream in one go we can run the fold multiple
  -- times, each time supplying the next segment of the input stream. If the fold
  -- has not yet finished it returns a fold that can be run again otherwise it
  -- returns the fold result and the residual stream.
  --
  -- /Internal/
  {-# INLINE_NORMAL foldBreakEither #-}
  foldBreakEither, foldEither :: Monad m =>
      Fold m a b -> Stream m a -> m (Either (Fold m a b) (b, Stream m a))
  foldBreakEither (Fold fstep begin done final) (UnStream step state) = do
      res <- begin
      case res of
          FL.Partial fs -> go SPEC fs state
          FL.Done fb -> return $! Right (fb, Stream step state)
  
      where
  
      {-# INLINE go #-}
      go !_ !fs st = do
          r <- step defState st
          case r of
              Yield x s -> do
                  res <- fstep fs x
                  case res of
                      FL.Done b -> return $! Right (b, Stream step s)
                      FL.Partial fs1 -> go SPEC fs1 s
              Skip s -> go SPEC fs s
              Stop ->
                  let f = Fold fstep (return $ FL.Partial fs) done final
                   in return $! Left f
  
  -- XXX deprecate in next major release
  -- Renamed because we need a foldEithers to fold a stream of Either
  foldEither = foldBreakEither
  
  -- | Like 'fold' but also returns the remaining stream. The resulting stream
  -- would be 'Stream.nil' if the stream finished before the fold.
  --
  {-# INLINE_NORMAL foldBreak #-}
  foldBreak :: Monad m => Fold m a b -> Stream m a -> m (b, Stream m a)
  foldBreak fld strm = do
      r <- foldEither fld strm
      case r of
          Right res -> return res
          Left (Fold _ initial _ final) -> do
              res <- initial
              case res of
                  FL.Done _ -> error "foldBreak: unreachable state"
                  FL.Partial s -> do
                      b <- final s
                      return (b, nil)
  
      where
  
      nil = Stream (\_ _ -> return Stop) ()
  
  -- >>> fold f = Fold.extractM . Stream.foldAddLazy f
  -- >>> fold f = Stream.fold Fold.one . Stream.foldMany0 f
  -- >>> fold f = Fold.extractM <=< Stream.foldAdd f
  
  -- | Fold a stream using the supplied left 'Fold' and reducing the resulting
  -- expression strictly at each step. The behavior is similar to 'foldl''. A
  -- 'Fold' can terminate early without consuming the full stream. See the
  -- documentation of individual 'Fold's for termination behavior.
  --
  -- Definitions:
  --
  -- >>> fold f = fmap fst . Stream.foldBreak f
  -- >>> fold f = Stream.parse (Parser.fromFold f)
  --
  -- Example:
  --
  -- >>> Stream.fold Fold.sum (Stream.enumerateFromTo 1 100)
  -- 5050
  --
  {-# INLINE_NORMAL fold #-}
  fold :: Monad m => Fold m a b -> Stream m a -> m b
  fold fld strm = do
      (b, _) <- foldBreak fld strm
      return b
  
  -- | Append a stream to a fold lazily to build an accumulator incrementally.
  --
  -- Example, to continue folding a list of streams on the same sum fold:
  --
  -- >>> streams = [Stream.fromList [1..5], Stream.fromList [6..10]]
  -- >>> f = Prelude.foldl Stream.foldAddLazy Fold.sum streams
  -- >>> Stream.fold f Stream.nil
  -- 55
  --
  {-# INLINE_NORMAL foldAddLazy #-}
  foldAddLazy :: Monad m => Fold m a b -> Stream m a -> Fold m a b
  foldAddLazy (Fold fstep finitial fextract ffinal) (Stream sstep state) =
      Fold fstep initial fextract ffinal
  
      where
  
      initial = do
          res <- finitial
          case res of
              FL.Partial fs -> go SPEC fs state
              FL.Done fb -> return $ FL.Done fb
  
      {-# INLINE go #-}
      go !_ !fs st = do
          r <- sstep defState st
          case r of
              Yield x s -> do
                  res <- fstep fs x
                  case res of
                      FL.Done b -> return $ FL.Done b
                      FL.Partial fs1 -> go SPEC fs1 s
              Skip s -> go SPEC fs s
              Stop -> return $ FL.Partial fs
  
  -- >>> foldAdd f = Stream.foldAddLazy f >=> Fold.reduce
  
  -- |
  -- >>> foldAdd = flip Fold.addStream
  --
  foldAdd :: Monad m => Fold m a b -> Stream m a -> m (Fold m a b)
  foldAdd f =
      Streamly.Internal.Data.Stream.Type.fold (FL.duplicate f)
  
  ------------------------------------------------------------------------------
  -- Right Folds
  ------------------------------------------------------------------------------
  
  -- Adapted from the vector package.
  --
  -- XXX Use of SPEC constructor in folds causes 2x performance degradation in
  -- one shot operations, but helps immensely in operations composed of multiple
  -- combinators or the same combinator many times. There seems to be an
  -- opportunity to optimize here, can we get both, better perf for single ops
  -- as well as composed ops? Without SPEC, all single operation benchmarks
  -- become 2x faster.
  
  -- The way we want a left fold to be strict, dually we want the right fold to
  -- be lazy.  The correct signature of the fold function to keep it lazy must be
  -- (a -> m b -> m b) instead of (a -> b -> m b). We were using the latter
  -- earlier, which is incorrect. In the latter signature we have to feed the
  -- value to the fold function after evaluating the monadic action, depending on
  -- the bind behavior of the monad, the action may get evaluated immediately
  -- introducing unnecessary strictness to the fold. If the implementation is
  -- lazy the following example, must work:
  --
  -- S.foldrM (\x t -> if x then return t else return False) (return True)
  --  (S.fromList [False,undefined] :: Stream IO Bool)
  
  -- | Right associative/lazy pull fold. @foldrM build final stream@ constructs
  -- an output structure using the step function @build@. @build@ is invoked with
  -- the next input element and the remaining (lazy) tail of the output
  -- structure. It builds a lazy output expression using the two. When the "tail
  -- structure" in the output expression is evaluated it calls @build@ again thus
  -- lazily consuming the input @stream@ until either the output expression built
  -- by @build@ is free of the "tail" or the input is exhausted in which case
  -- @final@ is used as the terminating case for the output structure. For more
  -- details see the description in the previous section.
  --
  -- Example, determine if any element is 'odd' in a stream:
  --
  -- >>> s = Stream.fromList (2:4:5:undefined)
  -- >>> step x xs = if odd x then return True else xs
  -- >>> Stream.foldrM step (return False) s
  -- True
  --
  -- >>> import Control.Monad (join)
  -- >>> foldrM f z = join . Stream.foldr f z
  --
  {-# INLINE_NORMAL foldrM #-}
  foldrM :: Monad m => (a -> m b -> m b) -> m b -> Stream m a -> m b
  -- foldrM f z = join . Streamly.Internal.Data.Stream.StreamD.Type.foldr f z
  foldrM f z (Stream step state) = go SPEC state
    where
      {-# INLINE_LATE go #-}
      go !_ st = do
            r <- step defState st
            case r of
              Yield x s -> f x (go SPEC s)
              Skip s    -> go SPEC s
              Stop      -> z
  
  {-# INLINE_NORMAL foldrMx #-}
  foldrMx :: Monad m
      => (a -> m x -> m x) -> m x -> (m x -> m b) -> Stream m a -> m b
  foldrMx fstep final convert (Stream step state) = convert $ go SPEC state
    where
      {-# INLINE_LATE go #-}
      go !_ st = do
            r <- step defState st
            case r of
              Yield x s -> fstep x (go SPEC s)
              Skip s    -> go SPEC s
              Stop      -> final
  
  -- XXX Should we make all argument strict wherever we use SPEC?
  
  -- Note that foldr works on pure values, therefore it becomes necessarily
  -- strict when the monad m is strict. In that case it cannot terminate early,
  -- it would evaluate all of its input.  Though, this should work fine with lazy
  -- monads. For example, if "any" is implemented using "foldr" instead of
  -- "foldrM" it performs the same with Identity monad but performs 1000x slower
  -- with IO monad.
  
  -- | Right fold, lazy for lazy monads and pure streams, and strict for strict
  -- monads.
  --
  -- Please avoid using this routine in strict monads like IO unless you need a
  -- strict right fold. This is provided only for use in lazy monads (e.g.
  -- Identity) or pure streams. Note that with this signature it is not possible
  -- to implement a lazy foldr when the monad @m@ is strict. In that case it
  -- would be strict in its accumulator and therefore would necessarily consume
  -- all its input.
  --
  -- >>> foldr f z = Stream.foldrM (\a b -> f a <$> b) (return z)
  --
  -- Note: This is similar to Fold.foldr' (the right fold via left fold), but
  -- could be more efficient.
  --
  {-# INLINE_NORMAL foldr #-}
  foldr :: Monad m => (a -> b -> b) -> b -> Stream m a -> m b
  foldr f z = foldrM (liftA2 f . return) (return z)
  
  -- this performs horribly, should not be used
  {-# INLINE_NORMAL foldrS #-}
  foldrS
      :: Monad m
      => (a -> Stream m b -> Stream m b)
      -> Stream m b
      -> Stream m a
      -> Stream m b
  foldrS f final (Stream step state) = go SPEC state
    where
      {-# INLINE_LATE go #-}
      go !_ st = concatEffect $ fmap g $ step defState st
  
      g r =
          case r of
            Yield x s -> f x (go SPEC s)
            Skip s    -> go SPEC s
            Stop      -> final
  
  ------------------------------------------------------------------------------
  -- Left Folds
  ------------------------------------------------------------------------------
  
  -- XXX run begin action only if the stream is not empty.
  {-# INLINE_NORMAL foldlMx' #-}
  foldlMx' :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> Stream m a -> m b
  foldlMx' fstep begin done (Stream step state) =
      begin >>= \x -> go SPEC x state
    where
      -- XXX !acc?
      {-# INLINE_LATE go #-}
      go !_ acc st = acc `seq` do
          r <- step defState st
          case r of
              Yield x s -> do
                  acc' <- fstep acc x
                  go SPEC acc' s
              Skip s -> go SPEC acc s
              Stop   -> done acc
  
  {-# INLINE foldlx' #-}
  foldlx' :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream m a -> m b
  foldlx' fstep begin done =
      foldlMx' (\b a -> return (fstep b a)) (return begin) (return . done)
  
  -- Adapted from the vector package.
  -- XXX implement in terms of foldlMx'?
  {-# INLINE_NORMAL foldlM' #-}
  foldlM' :: Monad m => (b -> a -> m b) -> m b -> Stream m a -> m b
  foldlM' fstep mbegin (Stream step state) = do
      begin <- mbegin
      go SPEC begin state
    where
      {-# INLINE_LATE go #-}
      go !_ acc st = acc `seq` do
          r <- step defState st
          case r of
              Yield x s -> do
                  acc' <- fstep acc x
                  go SPEC acc' s
              Skip s -> go SPEC acc s
              Stop   -> return acc
  
  {-# INLINE foldl' #-}
  foldl' :: Monad m => (b -> a -> b) -> b -> Stream m a -> m b
  foldl' fstep begin = foldlM' (\b a -> return (fstep b a)) (return begin)
  
  ------------------------------------------------------------------------------
  -- Special folds
  ------------------------------------------------------------------------------
  
  -- >>> drain = mapM_ (\_ -> return ())
  
  -- |
  -- Definitions:
  --
  -- >>> drain = Stream.fold Fold.drain
  -- >>> drain = Stream.foldrM (\_ xs -> xs) (return ())
  --
  -- Run a stream, discarding the results.
  --
  {-# INLINE_LATE drain #-}
  drain :: Monad m => Stream m a -> m ()
  -- drain = foldrM (\_ xs -> xs) (return ())
  drain (Stream step state) = go SPEC state
    where
      go !_ st = do
          r <- step defState st
          case r of
              Yield _ s -> go SPEC s
              Skip s    -> go SPEC s
              Stop      -> return ()
  
  {-# INLINE_NORMAL head #-}
  head :: Monad m => Stream m a -> m (Maybe a)
  #ifdef USE_FOLDS_EVERYWHERE
  head = fold Fold.one
  #else
  head = foldrM (\x _ -> return (Just x)) (return Nothing)
  #endif
  
  {-# INLINE_NORMAL headElse #-}
  headElse :: Monad m => a -> Stream m a -> m a
  headElse a = foldrM (\x _ -> return x) (return a)
  
  ------------------------------------------------------------------------------
  -- To Containers
  ------------------------------------------------------------------------------
  
  -- This toList impl is faster (30% on streaming-benchmarks) than the
  -- corresponding left fold. The left fold retains an additional argument in the
  -- recursive loop.
  --
  -- Core for the right fold loop:
  --
  -- main_$s$wgo1
  --   = \ sc_s3e6 sc1_s3e5 ->
  --       case ># sc1_s3e5 100000# of {
  --         __DEFAULT ->
  --           case main_$s$wgo1 sc_s3e6 (+# sc1_s3e5 1#) of
  --
  -- Core for the left fold loop:
  --
  --  main_$s$wgo1
  --   = \ sc_s3oT sc1_s3oS sc2_s3oR ->
  --       case sc2_s3oR of fs2_a2lw { __DEFAULT ->
  --       case ># sc1_s3oS 100000# of {
  --         __DEFAULT ->
  --           let { wild_a2og = I# sc1_s3oS } in
  --           main_$s$wgo1
  --             sc_s3oT (+# sc1_s3oS 1#) (\ x_X9 -> fs2_a2lw (: wild_a2og x_X9));
  
  -- |
  -- Definitions:
  --
  -- >>> toList = Stream.foldr (:) []
  -- >>> toList = Stream.fold Fold.toList
  --
  -- Convert a stream into a list in the underlying monad. The list can be
  -- consumed lazily in a lazy monad (e.g. 'Identity'). In a strict monad (e.g.
  -- IO) the whole list is generated and buffered before it can be consumed.
  --
  -- /Warning!/ working on large lists accumulated as buffers in memory could be
  -- very inefficient, consider using "Streamly.Data.Array" instead.
  --
  -- Note that this could a bit more efficient compared to @Stream.fold
  -- Fold.toList@, and it can fuse with pure list consumers.
  --
  {-# INLINE_NORMAL toList #-}
  toList :: Monad m => Stream m a -> m [a]
  toList = Streamly.Internal.Data.Stream.Type.foldr (:) []
  
  -- Use foldr/build fusion to fuse with list consumers
  -- This can be useful when using the IsList instance
  {-# INLINE_LATE toListFB #-}
  toListFB :: (a -> b -> b) -> b -> Stream Identity a -> b
  toListFB c n (Stream step state) = go state
    where
      go st = case runIdentity (step defState st) of
               Yield x s -> x `c` go s
               Skip s    -> go s
               Stop      -> n
  
  {-# RULES "toList Identity" Streamly.Internal.Data.Stream.Type.toList = toListId #-}
  {-# INLINE_EARLY toListId #-}
  toListId :: Stream Identity a -> Identity [a]
  toListId s = Identity $ build (\c n -> toListFB c n s)
  
  ------------------------------------------------------------------------------
  -- Multi-stream folds
  ------------------------------------------------------------------------------
  
  -- Adapted from the vector package.
  
  -- | Compare two streams for equality
  {-# INLINE_NORMAL eqBy #-}
  eqBy :: Monad m => (a -> b -> Bool) -> Stream m a -> Stream m b -> m Bool
  eqBy eq (Stream step1 t1) (Stream step2 t2) = eq_loop0 SPEC t1 t2
    where
      eq_loop0 !_ s1 s2 = do
        r <- step1 defState s1
        case r of
          Yield x s1' -> eq_loop1 SPEC x s1' s2
          Skip    s1' -> eq_loop0 SPEC   s1' s2
          Stop        -> eq_null s2
  
      eq_loop1 !_ x s1 s2 = do
        r <- step2 defState s2
        case r of
          Yield y s2'
            | eq x y    -> eq_loop0 SPEC   s1 s2'
            | otherwise -> return False
          Skip    s2'   -> eq_loop1 SPEC x s1 s2'
          Stop          -> return False
  
      eq_null s2 = do
        r <- step2 defState s2
        case r of
          Yield _ _ -> return False
          Skip s2'  -> eq_null s2'
          Stop      -> return True
  
  -- Adapted from the vector package.
  
  -- | Compare two streams lexicographically.
  {-# INLINE_NORMAL cmpBy #-}
  cmpBy
      :: Monad m
      => (a -> b -> Ordering) -> Stream m a -> Stream m b -> m Ordering
  cmpBy cmp (Stream step1 t1) (Stream step2 t2) = cmp_loop0 SPEC t1 t2
    where
      cmp_loop0 !_ s1 s2 = do
        r <- step1 defState s1
        case r of
          Yield x s1' -> cmp_loop1 SPEC x s1' s2
          Skip    s1' -> cmp_loop0 SPEC   s1' s2
          Stop        -> cmp_null s2
  
      cmp_loop1 !_ x s1 s2 = do
        r <- step2 defState s2
        case r of
          Yield y s2' -> case x `cmp` y of
                           EQ -> cmp_loop0 SPEC s1 s2'
                           c  -> return c
          Skip    s2' -> cmp_loop1 SPEC x s1 s2'
          Stop        -> return GT
  
      cmp_null s2 = do
        r <- step2 defState s2
        case r of
          Yield _ _ -> return LT
          Skip s2'  -> cmp_null s2'
          Stop      -> return EQ
  
  ------------------------------------------------------------------------------
  -- Transformations
  ------------------------------------------------------------------------------
  
  -- Adapted from the vector package.
  
  -- |
  -- >>> mapM f = Stream.sequence . fmap f
  --
  -- Apply a monadic function to each element of the stream and replace it with
  -- the output of the resulting action.
  --
  -- >>> s = Stream.fromList ["a", "b", "c"]
  -- >>> Stream.fold Fold.drain $ Stream.mapM putStr s
  -- abc
  --
  -- This is functional equivalent of an imperative loop.
  --
  {-# INLINE_NORMAL mapM #-}
  mapM :: Monad m => (a -> m b) -> Stream m a -> Stream m b
  mapM f (Stream step state) = Stream step1 state
  
      where
  
      {-# INLINE_LATE step1 #-}
      step1 gst st = do
          r <- step (adaptState gst) st
          case r of
              Yield x s -> f x >>= \a -> return $ Yield a s
              Skip s    -> return $ Skip s
              Stop      -> return Stop
  
  {-# INLINE map #-}
  map :: Monad m => (a -> b) -> Stream m a -> Stream m b
  map f = mapM (return . f)
  
  -- (Functor m) based implementation of fmap does not fuse well in
  -- streaming-benchmarks. XXX need to investigate why.
  instance Monad m => Functor (Stream m) where
      {-# INLINE fmap #-}
      fmap = map
  
      {-# INLINE (<$) #-}
      (<$) = fmap . const
  
  ------------------------------------------------------------------------------
  -- Lists
  ------------------------------------------------------------------------------
  
  -- XXX Show instance is 10x slower compared to read, we can do much better.
  -- The list show instance itself is really slow.
  
  -- XXX The default definitions of "<" in the Ord instance etc. do not perform
  -- well, because they do not get inlined. Need to add INLINE in Ord class in
  -- base?
  
  instance IsList (Stream Identity a) where
      type (Item (Stream Identity a)) = a
  
      {-# INLINE fromList #-}
      fromList = Streamly.Internal.Data.Stream.Type.fromList
  
      {-# INLINE toList #-}
      toList = runIdentity . Streamly.Internal.Data.Stream.Type.toList
  
  instance Eq a => Eq (Stream Identity a) where
      {-# INLINE (==) #-}
      (==) xs ys = runIdentity $ eqBy (==) xs ys
  
  instance Ord a => Ord (Stream Identity a) where
      {-# INLINE compare #-}
      compare xs ys = runIdentity $ cmpBy compare xs ys
  
      {-# INLINE (<) #-}
      x < y =
          case compare x y of
              LT -> True
              _ -> False
  
      {-# INLINE (<=) #-}
      x <= y =
          case compare x y of
              GT -> False
              _ -> True
  
      {-# INLINE (>) #-}
      x > y =
          case compare x y of
              GT -> True
              _ -> False
  
      {-# INLINE (>=) #-}
      x >= y =
          case compare x y of
              LT -> False
              _ -> True
  
      {-# INLINE max #-}
      max x y = if x <= y then y else x
  
      {-# INLINE min #-}
      min x y = if x <= y then x else y
  
  instance Show a => Show (Stream Identity a) where
      showsPrec p dl = showParen (p > 10) $
          showString "fromList " . shows (GHC.Exts.toList dl)
  
  instance Read a => Read (Stream Identity a) where
      readPrec = parens $ prec 10 $ do
          Ident "fromList" <- lexP
          Streamly.Internal.Data.Stream.Type.fromList <$> readPrec
  
      readListPrec = readListPrecDefault
  
  instance (a ~ Char) => IsString (Stream Identity a) where
      {-# INLINE fromString #-}
      fromString = Streamly.Internal.Data.Stream.Type.fromList
  
  -------------------------------------------------------------------------------
  -- Foldable
  -------------------------------------------------------------------------------
  
  -- The default Foldable instance has several issues:
  -- 1) several definitions do not have INLINE on them, so we provide
  --    re-implementations with INLINE pragmas.
  -- 2) the definitions of sum/product/maximum/minimum are inefficient as they
  --    use right folds, they cannot run in constant memory. We provide
  --    implementations using strict left folds here.
  
  -- There is no Traversable instance because, there is no scalable cons for
  -- StreamD, use toList and fromList instead.
  
  instance (Foldable m, Monad m) => Foldable (Stream m) where
  
      {-# INLINE foldMap #-}
      foldMap f =
          Data.Foldable.fold
              . Streamly.Internal.Data.Stream.Type.foldr (mappend . f) mempty
  
      {-# INLINE foldr #-}
      foldr f z t = appEndo (foldMap (Endo #. f) t) z
  
      {-# INLINE foldl' #-}
      foldl' f z0 xs = Data.Foldable.foldr f' id xs z0
          where f' x k = oneShot $ \z -> k $! f z x
  
      {-# INLINE length #-}
      length = Data.Foldable.foldl' (\n _ -> n + 1) 0
  
      {-# INLINE elem #-}
      elem = any . (==)
  
      {-# INLINE maximum #-}
      maximum =
            fromMaybe (errorWithoutStackTrace "maximum: empty stream")
          . toMaybe
          . Data.Foldable.foldl' getMax Nothing'
  
          where
  
          getMax Nothing' x = Just' x
          getMax (Just' mx) x = Just' $! max mx x
  
      {-# INLINE minimum #-}
      minimum =
            fromMaybe (errorWithoutStackTrace "minimum: empty stream")
          . toMaybe
          . Data.Foldable.foldl' getMin Nothing'
  
          where
  
          getMin Nothing' x = Just' x
          getMin (Just' mn) x = Just' $! min mn x
  
      {-# INLINE sum #-}
      sum = Data.Foldable.foldl' (+) 0
  
      {-# INLINE product #-}
      product = Data.Foldable.foldl' (*) 1
  
  -------------------------------------------------------------------------------
  -- Filtering
  -------------------------------------------------------------------------------
  
  -- Adapted from the vector package.
  
  -- | Take first 'n' elements from the stream and discard the rest.
  --
  {-# INLINE_NORMAL take #-}
  take :: Applicative m => Int -> Stream m a -> Stream m a
  take n (Stream step state) = n `seq` Stream step' (state, 0)
  
      where
  
      {-# INLINE_LATE step' #-}
      step' gst (st, i) | i < n = do
          (\case
              Yield x s -> Yield x (s, i + 1)
              Skip s    -> Skip (s, i)
              Stop      -> Stop) <$> step gst st
      step' _ (_, _) = pure Stop
  
  -- Adapted from the vector package.
  
  -- | Same as 'takeWhile' but with a monadic predicate.
  --
  {-# INLINE_NORMAL takeWhileM #-}
  takeWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
  -- takeWhileM p = scanMaybe (FL.takingEndByM_ (\x -> not <$> p x))
  takeWhileM f (Stream step state) = Stream step' state
    where
      {-# INLINE_LATE step' #-}
      step' gst st = do
          r <- step gst st
          case r of
              Yield x s -> do
                  b <- f x
                  return $ if b then Yield x s else Stop
              Skip s -> return $ Skip s
              Stop   -> return Stop
  
  -- | End the stream as soon as the predicate fails on an element.
  --
  {-# INLINE takeWhile #-}
  takeWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
  takeWhile f = takeWhileM (return . f)
  
  -- Like takeWhile but with an inverted condition and also taking
  -- the matching element.
  
  {-# INLINE_NORMAL takeEndByM #-}
  takeEndByM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
  takeEndByM f (Stream step state) = Stream step' (Just state)
    where
      {-# INLINE_LATE step' #-}
      step' gst (Just st) = do
          r <- step gst st
          case r of
              Yield x s -> do
                  b <- f x
                  return $
                      if not b
                      then Yield x (Just s)
                      else Yield x Nothing
              Skip s -> return $ Skip (Just s)
              Stop   -> return Stop
  
      step' _ Nothing = return Stop
  
  {-# INLINE takeEndBy #-}
  takeEndBy :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
  takeEndBy f = takeEndByM (return . f)
  
  -- |
  -- >>> takeEndBy_ f = Stream.takeWhile (not . f)
  --
  {-# INLINE takeEndBy_ #-}
  takeEndBy_ :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
  takeEndBy_ f = takeWhile (not . f)
  
  ------------------------------------------------------------------------------
  -- Appending
  ------------------------------------------------------------------------------
  
  data AppendState s1 s2 = AppendFirst s1 | AppendSecond s2
  
  -- Performance Note: From an implementation perspective,
  -- StreamK.'Streamly.Data.StreamK.append' translates into a function call
  -- whereas Stream.'append' translates into a conditional branch (jump).
  -- However, the overhead of the function call in StreamK.append is incurred
  -- only once, while the overhead of the conditional branch in fused append is
  -- incurred for each element in the stream. As a result, StreamK.append has a
  -- linear time complexity of O(n), while fused append has a quadratic time
  -- complexity of O(n^2), where @n@ represents the number of 'append's used.
  
  -- | WARNING! O(n^2) time complexity wrt number of streams. Suitable for
  -- statically fusing a small number of streams. Use the O(n) complexity
  -- StreamK.'Streamly.Data.StreamK.append' otherwise.
  --
  -- Fuses two streams sequentially, yielding all elements from the first
  -- stream, and then all elements from the second stream.
  --
  -- >>> s1 = Stream.fromList [1,2]
  -- >>> s2 = Stream.fromList [3,4]
  -- >>> Stream.fold Fold.toList $ s1 `Stream.append` s2
  -- [1,2,3,4]
  --
  {-# INLINE_NORMAL append #-}
  append :: Monad m => Stream m a -> Stream m a -> Stream m a
  append (Stream step1 state1) (Stream step2 state2) =
      Stream step (AppendFirst state1)
  
      where
  
      {-# INLINE_LATE step #-}
      step gst (AppendFirst st) = do
          r <- step1 gst st
          return $ case r of
              Yield a s -> Yield a (AppendFirst s)
              Skip s -> Skip (AppendFirst s)
              Stop -> Skip (AppendSecond state2)
  
      step gst (AppendSecond st) = do
          r <- step2 gst st
          return $ case r of
              Yield a s -> Yield a (AppendSecond s)
              Skip s -> Skip (AppendSecond s)
              Stop -> Stop
  
  ------------------------------------------------------------------------------
  -- Zipping
  ------------------------------------------------------------------------------
  
  -- | Like 'zipWith' but using a monadic zipping function.
  --
  {-# INLINE_NORMAL zipWithM #-}
  zipWithM :: Monad m
      => (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
  zipWithM f (Stream stepa ta) (Stream stepb tb) = Stream step (ta, tb, Nothing)
    where
      {-# INLINE_LATE step #-}
      step gst (sa, sb, Nothing) = do
          r <- stepa (adaptState gst) sa
          return $
            case r of
              Yield x sa' -> Skip (sa', sb, Just x)
              Skip sa'    -> Skip (sa', sb, Nothing)
              Stop        -> Stop
  
      step gst (sa, sb, Just x) = do
          r <- stepb (adaptState gst) sb
          case r of
              Yield y sb' -> do
                  z <- f x y
                  return $ Yield z (sa, sb', Nothing)
              Skip sb' -> return $ Skip (sa, sb', Just x)
              Stop     -> return Stop
  
  {-# RULES "zipWithM xs xs"
      forall f xs. zipWithM @Identity f xs xs = mapM (\x -> f x x) xs #-}
  
  -- | WARNING! O(n^2) time complexity wrt number of streams. Suitable for
  -- statically fusing a small number of streams. Use the O(n) complexity
  -- StreamK.'Streamly.Data.StreamK.zipWith' otherwise.
  --
  -- Stream @a@ is evaluated first, followed by stream @b@, the resulting
  -- elements @a@ and @b@ are then zipped using the supplied zip function and the
  -- result @c@ is yielded to the consumer.
  --
  -- If stream @a@ or stream @b@ ends, the zipped stream ends. If stream @b@ ends
  -- first, the element @a@ from previous evaluation of stream @a@ is discarded.
  --
  -- >>> s1 = Stream.fromList [1,2,3]
  -- >>> s2 = Stream.fromList [4,5,6]
  -- >>> Stream.fold Fold.toList $ Stream.zipWith (+) s1 s2
  -- [5,7,9]
  --
  {-# INLINE zipWith #-}
  zipWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
  zipWith f = zipWithM (\a b -> return (f a b))
  
  ------------------------------------------------------------------------------
  -- Combine N Streams - concatAp
  ------------------------------------------------------------------------------
  
  -- XXX unfoldApplyEach
  
  -- | Apply a stream of functions to a stream of values and flatten the results.
  --
  -- Note that the second stream is evaluated multiple times.
  --
  -- >>> crossApply = Stream.crossWith id
  --
  {-# INLINE_NORMAL crossApply #-}
  crossApply :: Functor f => Stream f (a -> b) -> Stream f a -> Stream f b
  crossApply (Stream stepa statea) (Stream stepb stateb) =
      Stream step' (Left statea)
  
      where
  
      {-# INLINE_LATE step' #-}
      step' gst (Left st) = fmap
          (\case
              Yield f s -> Skip (Right (f, s, stateb))
              Skip    s -> Skip (Left s)
              Stop      -> Stop)
          (stepa (adaptState gst) st)
      step' gst (Right (f, os, st)) = fmap
          (\case
              Yield a s -> Yield (f a) (Right (f, os, s))
              Skip s    -> Skip (Right (f,os, s))
              Stop      -> Skip (Left os))
          (stepb (adaptState gst) st)
  
  -- This is shared by all fairUnfold, fairConcat combinators.
  data FairUnfoldState o i =
        FairUnfoldInit o ([i] -> [i])
      | FairUnfoldNext o ([i] -> [i]) [i]
      | FairUnfoldDrain ([i] -> [i]) [i]
  
  -- XXX will it perform better if we write it in the same way as crossApply?
  -- crossApply is faster than unfoldCross in equation solving benchmarks.
  
  -- | Like 'fairCrossWith' but with monadic function argument.
  --
  {-# INLINE_NORMAL fairCrossWithM #-}
  fairCrossWithM :: Monad m =>
      (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
  fairCrossWithM f (Stream step1 state1) (Stream step2 state2) =
      Stream step (FairUnfoldInit state1 id)
  
      where
  
      {-# INLINE_LATE step #-}
      step gst (FairUnfoldInit o ls) = do
          r <- step1 (adaptState gst) o
          return $ case r of
              Yield b o' -> Skip (FairUnfoldNext o' id (ls [(b,state2)]))
              Skip o' -> Skip (FairUnfoldInit o' ls)
              Stop -> Skip (FairUnfoldDrain id (ls []))
  
      step _ (FairUnfoldNext o ys []) =
              return $ Skip (FairUnfoldInit o ys)
  
      step gst (FairUnfoldNext o ys ((b,st):ls)) = do
          r <- step2 (adaptState gst) st
          case r of
              Yield c s ->
                  f b c >>= \x ->
                      return $ Yield x (FairUnfoldNext o (ys . ((b, s) :)) ls)
              Skip s    -> return $ Skip (FairUnfoldNext o ys ((b,s) : ls))
              Stop      -> return $ Skip (FairUnfoldNext o ys ls)
  
      step _ (FairUnfoldDrain ys []) =
          case ys [] of
              [] -> return Stop
              xs -> return $ Skip (FairUnfoldDrain id xs)
  
      step gst (FairUnfoldDrain ys ((b,st):ls)) = do
          r <- step2 (adaptState gst) st
          case r of
              Yield c s ->
                  f b c >>= \x ->
                      return $ Yield x (FairUnfoldDrain (ys . ((b,s) :)) ls)
              Skip s    -> return $ Skip (FairUnfoldDrain ys ((b,s) : ls))
              Stop      -> return $ Skip (FairUnfoldDrain ys ls)
  
  {-# INLINE_NORMAL crossApplySnd #-}
  crossApplySnd :: Functor f => Stream f a -> Stream f b -> Stream f b
  crossApplySnd (Stream stepa statea) (Stream stepb stateb) =
      Stream step (Left statea)
  
      where
  
      {-# INLINE_LATE step #-}
      step gst (Left st) =
          fmap
              (\case
                   Yield _ s -> Skip (Right (s, stateb))
                   Skip s -> Skip (Left s)
                   Stop -> Stop)
              (stepa (adaptState gst) st)
      step gst (Right (ostate, st)) =
          fmap
              (\case
                   Yield b s -> Yield b (Right (ostate, s))
                   Skip s -> Skip (Right (ostate, s))
                   Stop -> Skip (Left ostate))
              (stepb gst st)
  
  {-# INLINE_NORMAL crossApplyFst #-}
  crossApplyFst :: Functor f => Stream f a -> Stream f b -> Stream f a
  crossApplyFst (Stream stepa statea) (Stream stepb stateb) =
      Stream step (Left statea)
  
      where
  
      {-# INLINE_LATE step #-}
      step gst (Left st) =
          fmap
              (\case
                   Yield b s -> Skip (Right (s, stateb, b))
                   Skip s -> Skip (Left s)
                   Stop -> Stop)
              (stepa gst st)
      step gst (Right (ostate, st, b)) =
          fmap
              (\case
                   Yield _ s -> Yield b (Right (ostate, s, b))
                   Skip s -> Skip (Right (ostate, s, b))
                   Stop -> Skip (Left ostate))
              (stepb (adaptState gst) st)
  
  {-
  instance Applicative f => Applicative (Stream f) where
      {-# INLINE pure #-}
      pure = fromPure
  
      {-# INLINE (<*>) #-}
      (<*>) = crossApply
  
      {-# INLINE liftA2 #-}
      liftA2 f x = (<*>) (fmap f x)
  
      {-# INLINE (*>) #-}
      (*>) = crossApplySnd
  
      {-# INLINE (<*) #-}
      (<*) = crossApplyFst
  -}
  
  -- XXX We can use @Stream Identity b@ as the second stream to avoid running
  -- effects multiple times. Or it could be an array or an unfold i.e.
  -- unfoldCross.
  
  -- |
  -- Definition:
  --
  -- >>> crossWith f m1 m2 = fmap f m1 `Stream.crossApply` m2
  --
  -- Note that the second stream is evaluated multiple times.
  --
  -- Also see "Streamly.Data.Unfold.crossWith" for fast fusible static cross
  -- product option.
  --
  {-# INLINE crossWith #-}
  crossWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
  crossWith f m1 m2 = fmap f m1 `crossApply` m2
  
  -- | Like 'crossWith' but interleaves the outer and inner loops fairly. See
  -- 'fairConcatFor' for more details.
  --
  {-# INLINE fairCrossWith #-}
  fairCrossWith :: Monad m =>
      (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
  fairCrossWith f = fairCrossWithM (\a b -> return $ f a b)
  
  -- | Given a @Stream m a@ and @Stream m b@ generate a stream with all possible
  -- combinations of the tuple @(a, b)@.
  --
  -- Definition:
  --
  -- >>> cross = Stream.crossWith (,)
  --
  -- The second stream is evaluated multiple times. If that is not desired it can
  -- be cached in an 'Data.Array.Array' and then generated from the array before
  -- calling this function. Caching may also improve performance if the stream is
  -- expensive to evaluate.
  --
  -- Time: O(m x n)
  --
  -- /Pre-release/
  {-# INLINE cross #-}
  cross :: Monad m => Stream m a -> Stream m b -> Stream m (a, b)
  cross = crossWith (,)
  
  -- | Like 'cross' but interleaves the outer and inner loops fairly. See
  -- 'fairConcatFor' for more details.
  {-# INLINE fairCross #-}
  fairCross :: Monad m => Stream m a -> Stream m b -> Stream m (a, b)
  fairCross = fairCrossWith (,)
  
  -- crossWith/cross should ideally use Stream m b as the first stream, because
  -- we are transforming Stream m a using that. We provide loop with arguments
  -- flipped.
  
  -- crossMap or crossInner?
  
  -- | Loop the supplied stream (first argument) around each element of the input
  -- stream (second argument) generating tuples.  This is an argument flipped
  -- version of 'cross'.
  {-# INLINE loop #-}
  loop :: Monad m => Stream m b -> Stream m a -> Stream m (a, b)
  loop = crossWith (\b a -> (a,b))
  
  -- | Loop by unfold. Unfold a value into a stream and nest it with the input
  -- stream. This is much faster than 'loop' due to stream fusion.
  {-# INLINE loopBy #-}
  loopBy :: Monad m => Unfold m x b -> x -> Stream m a -> Stream m (a, b)
  loopBy u x s =
      let u1 = Unfold.lmap snd u
          u2 = Unfold.map (first fst) (Unfold.carry u1)
       in unfoldEach u2 $ fmap (, x) s
  
  ------------------------------------------------------------------------------
  -- Combine N Streams - unfoldEach
  ------------------------------------------------------------------------------
  
  {-# ANN type UnfoldEachState Fuse #-}
  data UnfoldEachState o i =
        UnfoldEachOuter o
      | UnfoldEachInner o i
  
  -- | @unfoldEach unfold stream@ uses @unfold@ to map the input stream elements
  -- to streams and then flattens the generated streams into a single output
  -- stream.
  
  -- This is like 'concatMap' but uses an unfold with an explicit state to
  -- generate the stream instead of a 'Stream' type generator. This allows better
  -- optimization via fusion.  This can be many times more efficient than
  -- 'concatMap'.
  --
  -- 'unfoldEach' is equivalent in expressive power to 'concatMap'. However,
  -- using it as concatMap — by lifting a function 'f :: a -> Stream m b' into
  -- an 'Unfold' — results in the same degraded performance as 'concatMap':
  
  -- | Like 'concatMap' but uses an 'Unfold' for stream generation. Unlike
  -- 'concatMap' this can fuse the 'Unfold' code with the inner loop and
  -- therefore provide many times better performance.
  --
  -- >>> concatMap f = Stream.unfoldEach (Unfold.lmap f Unfold.fromStream)
  --
  -- Here is an example of a two level nested loop much faster than
  -- 'concatMap' based nesting.
  --
  -- >>> :{
  -- outerLoop =
  --   flip Stream.mapM (Stream.fromList [1,2,3]) $ \x -> do
  --       liftIO $ putStrLn (show x)
  --       return x
  -- innerUnfold = Unfold.carry $ Unfold.lmap (const [4,5,6]) Unfold.fromList
  -- innerLoop =
  --      flip Unfold.mapM innerUnfold $ \(x, y) -> do
  --          when (x == 1) $ liftIO $ putStrLn (show y)
  --          pure $ (x, y)
  -- :}
  --
  -- >>> Stream.toList $ Stream.unfoldEach innerLoop outerLoop
  -- 1
  -- 4
  -- 5
  -- 6
  -- 2
  -- 3
  -- [(1,4),(1,5),(1,6),(2,4),(2,5),(2,6),(3,4),(3,5),(3,6)]
  --
  {-# INLINE_NORMAL unfoldEach #-}
  unfoldEach, unfoldMany :: Monad m => Unfold m a b -> Stream m a -> Stream m b
  unfoldEach (Unfold istep inject) (Stream ostep ost) =
      Stream step (UnfoldEachOuter ost)
  
      where
  
      {-# INLINE_LATE step #-}
      step gst (UnfoldEachOuter o) = do
          r <- ostep (adaptState gst) o
          case r of
              Yield a o' -> do
                  i <- inject a
                  i `seq` return (Skip (UnfoldEachInner o' i))
              Skip o' -> return $ Skip (UnfoldEachOuter o')
              Stop -> return Stop
  
      step _ (UnfoldEachInner o i) = do
          r <- istep i
          return $ case r of
              Yield x i' -> Yield x (UnfoldEachInner o i')
              Skip i'    -> Skip (UnfoldEachInner o i')
              Stop       -> Skip (UnfoldEachOuter o)
  
  RENAME(unfoldMany,unfoldEach)
  
  -- XXX unfoldEach generates faster code than unfoldCross with
  -- everything unboxed i.e. 0 allocations.
  
  -- | Generates a cross product of two streams and then unfolds each tuple.
  --
  -- A two level nested loop much faster than 'concatMap' based nesting.
  --
  -- >>> :{
  -- outerLoop =
  --   flip Stream.mapM (Stream.fromList [1,2,3]) $ \x -> do
  --       liftIO $ putStrLn (show x)
  --       return x
  -- innerLoop =
  --   flip Stream.mapM (Stream.fromList [4,5,6]) $ \y -> do
  --       -- liftIO $ putStrLn (show y)
  --       return y
  -- innerUnfold =
  --   flip Unfold.mapM Unfold.identity $ \(x,y) -> do
  --      when (x == 1) $ liftIO $ putStrLn (show y)
  --      pure $ (x, y)
  -- :}
  --
  -- >>> Stream.toList $ Stream.unfoldCross innerUnfold outerLoop innerLoop
  -- 1
  -- 4
  -- 5
  -- 6
  -- 2
  -- 3
  -- [(1,4),(1,5),(1,6),(2,4),(2,5),(2,6),(3,4),(3,5),(3,6)]
  --
  -- Note: 'unfoldEach' may generate faster code, so use that when possible.
  -- Also see "Streamly.Data.Unfold.cross" for fast fusible static cross product
  -- option.
  --
  {-# INLINE unfoldCross #-}
  unfoldCross :: Monad m =>
      Unfold m (a,b) c -> Stream m a -> Stream m b -> Stream m c
  unfoldCross unf m1 m2 = unfoldEach unf $ crossWith (,) m1 m2
  
  ------------------------------------------------------------------------------
  -- Combine N Streams - concatMap
  ------------------------------------------------------------------------------
  
  -- Adapted from the vector package.
  
  -- If we are iterating over n elements flat vs m nesting of n^{1/m} elements.
  -- The total iterations in the nesting case will be
  -- let x = n^{1/m} in {x + x^2 + x^3 + ... + x^m} = x * {(x^m - 1)/(x-1)}
  -- i.e. (n-1)/(1-1/x) which is not very high. However, the decision tree in the
  -- state of concatMap is traversed from root to leaf for each element, and the
  -- state updates also updates the tree from leaf to root upon yielding an
  -- element which makes the allocations as well as CPU performance quadratic.
  
  -- | Map a stream producing monadic function on each element of the stream
  -- and then flatten the results into a single stream. Since the stream
  -- generation function is monadic, unlike 'concatMap', it can produce an
  -- effect at the beginning of each iteration of the inner loop.
  --
  -- See 'unfoldEach' for a faster alternative.
  --
  {-# INLINE_NORMAL concatMapM #-}
  concatMapM :: Monad m => (a -> m (Stream m b)) -> Stream m a -> Stream m b
  concatMapM f (Stream step state) = Stream step' (Left state)
    where
      {-# INLINE_LATE step' #-}
      step' gst (Left st) = do
          r <- step (adaptState gst) st
          case r of
              Yield a s -> do
                  b_stream <- f a
                  return $ Skip (Right (b_stream, s))
              Skip s -> return $ Skip (Left s)
              Stop -> return Stop
  
      -- XXX using the pattern synonym "Stream" causes a major performance issue
      -- here even if the synonym does not include an adaptState call. Need to
      -- find out why. Is that something to be fixed in GHC?
      step' gst (Right (UnStream inner_step inner_st, st)) = do
          r <- inner_step (adaptState gst) inner_st
          case r of
              Yield b inner_s ->
                  return $ Yield b (Right (Stream inner_step inner_s, st))
              Skip inner_s ->
                  return $ Skip (Right (Stream inner_step inner_s, st))
              Stop -> return $ Skip (Left st)
  
  -- | Map a stream producing function on each element of the stream and then
  -- flatten the results into a single stream.
  --
  -- >>> concatMap f = Stream.concat . fmap f
  -- >>> concatMap f = Stream.concatMapM (return . f)
  -- >>> concatMap f = Stream.unfoldEach (Unfold.lmap f Unfold.fromStream)
  --
  -- See argument flipped version 'concatFor' for more detailed documentation.
  --
  -- NOTE: We recommend using 'unfoldEach' or 'unfoldCross' instead of
  -- 'concatMap' especially in performance critical code. 'unfoldEach' is much
  -- faster than 'concatMap' and matches its expressive power in terms of
  -- generating dependent inner streams, there is one important distinction
  -- though: the nesting structure when using 'unfoldEach' is fixed statically in
  -- the code. In contrast, 'concatMap' allows dynamic and arbitrary nesting
  -- through monadic composition. This means that deeply nested or
  -- programmatically determined levels of nesting are easier to express and
  -- compose with 'concatMap', though often at the cost of performance and
  -- fusion.
  --
  {-# INLINE concatMap #-}
  concatMap :: Monad m => (a -> Stream m b) -> Stream m a -> Stream m b
  concatMap f = concatMapM (return . f)
  
  -- XXX Add smap/sfor (mapAccum) as a more ergonomic substitute for scan, not
  -- exposing the state in the output stream.
  -- XXX add sconcatMap/sconcatFor as stateful alternatives.
  
  -- | Map a stream generating function on each element of a stream and
  -- concatenate the results. This is the same as the bind function of the monad
  -- instance. It is just a flipped 'concatMap' but more convenient to use for
  -- nested use case, feels like an imperative @for@ loop. It is in fact
  -- equivalent to @concat . for@.
  --
  -- >>> concatFor = flip Stream.concatMap
  --
  -- A concatenating @for@ loop:
  --
  -- >>> :{
  -- Stream.toList $
  --     Stream.concatFor (Stream.fromList [1,2,3]) $ \x ->
  --       Stream.fromPure x
  -- :}
  -- [1,2,3]
  --
  -- Use 'unfoldEach' instead of 'concatFor' where possible, unfoldEach is much
  -- faster due to fusion.
  --
  -- Nested concatenating @for@ loops:
  --
  -- >>> :{
  -- Stream.toList $
  --     Stream.concatFor (Stream.fromList [1,2,3]) $ \x ->
  --      Stream.concatFor (Stream.fromList [4,5,6]) $ \y ->
  --       Stream.fromPure (x, y)
  -- :}
  -- [(1,4),(1,5),(1,6),(2,4),(2,5),(2,6),(3,4),(3,5),(3,6)]
  --
  -- If total iterations are kept the same, each increase in the nesting level
  -- increases the cost by roughly 2 times.
  --
  -- For significantly faster multi-level nesting, prefer using the better
  -- fusible, applicative-like 'crossWith' over 'concatFor' where possible.
  --
  -- 'concatFor' is monad-like: it allows expressing dependencies between the
  -- outer and the inner loops of the nesting, it means that the stream generated
  -- by the inner loop is dynamically governed by the outer loop. This expressive
  -- power comes at a significant performance cost.
  --
  -- NOTE: We recommend using 'unfoldEach' or 'unfoldCross' instead of
  -- 'concatFor' especially in performance critical code. 'unfoldEach' is much
  -- faster than 'concatFor' and matches its expressive power in terms of
  -- generating dependent inner streams, there is one important distinction
  -- though: the nesting structure when using 'unfoldEach' is fixed statically in
  -- the code. In contrast, 'concatFor' allows dynamic and arbitrary nesting
  -- through monadic composition. This means that deeply nested or
  -- programmatically determined levels of nesting are easier to express and
  -- compose with 'concatFor', though often at the cost of performance and
  -- fusion.
  --
  --
  {-# INLINE concatFor #-}
  concatFor :: Monad m => Stream m a -> (a -> Stream m b) -> Stream m b
  concatFor = flip concatMap
  
  -- NOTE: A monad instance can do with just concatMap because we lift an effect
  -- explicitly using liftIO and then bind/applicative concats it. Not sure if it
  -- gets fused and the overhead of additional concat removed. However, for
  -- explicit use concatForM is more ergonomic as we do not need to lift and
  -- concat, we can just use the do notation in the underlying monad..
  
  -- | Like 'concatFor' but maps an effectful function. It allows conveniently
  -- mixing monadic effects with streams.
  --
  -- >>> import Control.Monad.IO.Class (liftIO)
  -- >>> :{
  -- Stream.toList $
  --     Stream.concatForM (Stream.fromList [1,2,3]) $ \x -> do
  --       liftIO $ putStrLn (show x)
  --       pure $ Stream.fromPure x
  -- :}
  -- 1
  -- 2
  -- 3
  -- [1,2,3]
  --
  -- Nested concatentating @for@ loops:
  --
  -- >>> :{
  -- Stream.toList $
  --     Stream.concatForM (Stream.fromList [1,2,3]) $ \x -> do
  --       liftIO $ putStrLn (show x)
  --       pure $ Stream.concatForM (Stream.fromList [4,5,6]) $ \y -> do
  --         when (x == 1) $ liftIO $ putStrLn (show y)
  --         pure $ Stream.fromPure (x, y)
  -- :}
  -- 1
  -- 4
  -- 5
  -- 6
  -- 2
  -- 3
  -- [(1,4),(1,5),(1,6),(2,4),(2,5),(2,6),(3,4),(3,5),(3,6)]
  --
  {-# INLINE concatForM #-}
  concatForM :: Monad m => Stream m a -> (a -> m (Stream m b)) -> Stream m b
  concatForM = flip concatMapM
  
  -- | Flatten a stream of streams to a single stream.
  --
  -- >>> concat = Stream.concatMap id
  --
  -- /Pre-release/
  {-# INLINE concat #-}
  concat :: Monad m => Stream m (Stream m a) -> Stream m a
  concat = concatMap id
  
  -- >>> concatEffect = Stream.concat . lift    -- requires (MonadTrans t)
  -- >>> concatEffect = join . lift             -- requires (MonadTrans t, Monad (Stream m))
  
  -- | Flatten a stream generated by an effect i.e. concat the effect monad with
  -- the stream monad.
  --
  -- >>> concatEffect = Stream.concat . Stream.fromEffect
  -- >>> concatEffect eff = Stream.concatMapM (\() -> eff) (Stream.fromPure ())
  --
  -- See also: 'concat', 'sequence'
  --
  {-# INLINE concatEffect #-}
  concatEffect :: Monad m => m (Stream m a) -> Stream m a
  concatEffect generator = concatMapM (\() -> generator) (fromPure ())
  
  {-
  -- NOTE: even though concatMap for StreamD is 4x faster compared to StreamK,
  -- the monad instance does not seem to be significantly faster.
  instance Monad m => Monad (Stream m) where
      {-# INLINE return #-}
      return = pure
  
      {-# INLINE (>>=) #-}
      (>>=) = flip concatMap
  
      {-# INLINE (>>) #-}
      (>>) = (*>)
  -}
  
  ------------------------------------------------------------------------------
  -- Traversing a tree top down
  ------------------------------------------------------------------------------
  
  -- Next stream is to be generated by the return value of the previous stream. A
  -- general intuitive way of doing that could be to use an appending monad
  -- instance for streams where the result of the previous stream is used to
  -- generate the next one. In the first pass we can just emit the values in the
  -- stream and keep building a buffered list/stream, once done we can then
  -- process the buffered stream.
  
  -- | Generate a stream from an initial state, scan and concat the stream,
  -- generate a stream again from the final state of the previous scan and repeat
  -- the process.
  {-# INLINE_NORMAL concatIterateScan #-}
  concatIterateScan :: Monad m =>
         (b -> a -> m b)
      -> (b -> m (Maybe (b, Stream m a)))
      -> b
      -> Stream m a
  concatIterateScan scanner generate initial = Stream step (Left initial)
  
      where
  
      {-# INLINE_LATE step #-}
      step _ (Left acc) = do
          r <- generate acc
          case r of
              Nothing -> return Stop
              Just v -> return $ Skip (Right v)
  
      step gst (Right (st, UnStream inner_step inner_st)) = do
          r <- inner_step (adaptState gst) inner_st
          case r of
              Yield b inner_s -> do
                  acc <- scanner st b
                  return $ Yield b (Right (acc, Stream inner_step inner_s))
              Skip inner_s ->
                  return $ Skip (Right (st, Stream inner_step inner_s))
              Stop -> return $ Skip (Left st)
  
  -- Note: The iterate function returns a Maybe Stream instead of returning a nil
  -- stream for indicating a leaf node. This is to optimize so that we do not
  -- have to store any state. This makes the stored state proportional to the
  -- number of non-leaf nodes rather than total number of nodes.
  
  -- | Same as 'concatIterateBfs' except that the traversal of the last
  -- element on a level is emitted first and then going backwards up to the first
  -- element (reversed ordering). This may be slightly faster than
  -- 'concatIterateBfs'.
  --
  {-# INLINE_NORMAL altBfsConcatIterate #-}
  altBfsConcatIterate, concatIterateBfsRev :: Monad m =>
         (a -> Maybe (Stream m a))
      -> Stream m a
      -> Stream m a
  altBfsConcatIterate f stream = Stream step (stream, [])
  
      where
  
      {-# INLINE_LATE step #-}
      step gst (UnStream step1 st, xs) = do
          r <- step1 (adaptState gst) st
          case r of
              Yield a s -> do
                  let xs1 =
                          case f a of
                              Nothing -> xs
                              Just x -> x:xs
                  return $ Yield a (Stream step1 s, xs1)
              Skip s -> return $ Skip (Stream step1 s, xs)
              Stop ->
                  case xs of
                      (y:ys) -> return $ Skip (y, ys)
                      [] -> return Stop
  
  RENAME(concatIterateBfsRev,altBfsConcatIterate)
  
  -- | Similar to 'concatIterate' except that it traverses the stream in
  -- breadth first style (BFS). First, all the elements in the input stream are
  -- emitted, and then their traversals are emitted.
  --
  -- Example, list a directory tree using BFS:
  --
  -- >>> f = either (Just . Dir.readEitherPaths id) (const Nothing)
  -- >>> input = Stream.fromEffect (Left <$> Path.fromString ".")
  -- >>> ls = Stream.bfsConcatIterate f input
  --
  -- /Pre-release/
  {-# INLINE_NORMAL bfsConcatIterate #-}
  bfsConcatIterate, concatIterateBfs :: Monad m =>
         (a -> Maybe (Stream m a))
      -> Stream m a
      -> Stream m a
  bfsConcatIterate f stream = Stream step (stream, [], [])
  
      where
  
      {-# INLINE_LATE step #-}
      step gst (UnStream step1 st, xs, ys) = do
          r <- step1 (adaptState gst) st
          case r of
              Yield a s -> do
                  let ys1 =
                          case f a of
                              Nothing -> ys
                              Just y -> y:ys
                  return $ Yield a (Stream step1 s, xs, ys1)
              Skip s -> return $ Skip (Stream step1 s, xs, ys)
              Stop ->
                  case xs of
                      (x:xs1) -> return $ Skip (x, xs1, ys)
                      [] ->
                          case reverse ys of
                              (x:xs1) -> return $ Skip (x, xs1, [])
                              [] -> return Stop
  
  RENAME(concatIterateBfs,bfsConcatIterate)
  
  -- | Traverse the stream in depth first style (DFS). Map each element in the
  -- input stream to a stream and flatten, recursively map the resulting elements
  -- as well to a stream and flatten until no more streams are generated.
  --
  -- Example, list a directory tree using DFS:
  --
  -- >>> f = either (Just . Dir.readEitherPaths id) (const Nothing)
  -- >>> input = Stream.fromEffect (Left <$> Path.fromString ".")
  -- >>> ls = Stream.concatIterate f input
  --
  -- This is equivalent to using @concatIterateWith StreamK.append@.
  --
  -- /Pre-release/
  {-# INLINE_NORMAL concatIterate #-}
  concatIterate, concatIterateDfs :: Monad m =>
         (a -> Maybe (Stream m a))
      -> Stream m a
      -> Stream m a
  concatIterate f stream = Stream step (stream, [])
  
      where
  
      {-# INLINE_LATE step #-}
      step gst (UnStream step1 st, xs) = do
          r <- step1 (adaptState gst) st
          case r of
              Yield a s -> do
                  let st1 =
                          case f a of
                              Nothing -> (Stream step1 s, xs)
                              Just x -> (x, Stream step1 s:xs)
                  return $ Yield a st1
              Skip s -> return $ Skip (Stream step1 s, xs)
              Stop ->
                  case xs of
                      (y:ys) -> return $ Skip (y, ys)
                      [] -> return Stop
  
  RENAME(concatIterateDfs,concatIterate)
  
  {-# ANN type IterateUnfoldState Fuse #-}
  data IterateUnfoldState o i =
        IterateUnfoldOuter o
      | IterateUnfoldInner o i [i]
  
  -- | Same as 'concatIterate' but more efficient due to stream fusion.
  --
  -- Example, list a directory tree using DFS:
  --
  -- >>> f = Unfold.either (Dir.eitherReaderPaths id) Unfold.nil
  -- >>> input = Stream.fromEffect (Left <$> Path.fromString ".")
  -- >>> ls = Stream.unfoldIterate f input
  --
  -- /Pre-release/
  {-# INLINE_NORMAL unfoldIterate #-}
  unfoldIterate, unfoldIterateDfs :: Monad m =>
         Unfold m a a
      -> Stream m a
      -> Stream m a
  unfoldIterate (Unfold istep inject) (Stream ostep ost) =
      Stream step (IterateUnfoldOuter ost)
  
      where
  
      {-# INLINE_LATE step #-}
      step gst (IterateUnfoldOuter o) = do
          r <- ostep (adaptState gst) o
          case r of
              Yield a s -> do
                  i <- inject a
                  i `seq` return (Yield a (IterateUnfoldInner s i []))
              Skip s -> return $ Skip (IterateUnfoldOuter s)
              Stop -> return Stop
  
      step _ (IterateUnfoldInner o i ii) = do
          r <- istep i
          case r of
              Yield x s -> do
                  i1 <- inject x
                  i1 `seq` return $ Yield x (IterateUnfoldInner o i1 (s:ii))
              Skip s -> return $ Skip (IterateUnfoldInner o s ii)
              Stop ->
                  case ii of
                      (y:ys) -> return $ Skip (IterateUnfoldInner o y ys)
                      [] -> return $ Skip (IterateUnfoldOuter o)
  
  RENAME(unfoldIterateDfs,unfoldIterate)
  
  {-# ANN type IterateUnfoldBFSRevState Fuse #-}
  data IterateUnfoldBFSRevState o i =
        IterateUnfoldBFSRevOuter o [i]
      | IterateUnfoldBFSRevInner i [i]
  
  -- | Like 'bfsUnfoldIterate' but processes the children in reverse order,
  -- therefore, may be slightly faster.
  --
  -- /Pre-release/
  {-# INLINE_NORMAL altBfsUnfoldIterate #-}
  altBfsUnfoldIterate, unfoldIterateBfsRev :: Monad m =>
         Unfold m a a
      -> Stream m a
      -> Stream m a
  altBfsUnfoldIterate (Unfold istep inject) (Stream ostep ost) =
      Stream step (IterateUnfoldBFSRevOuter ost [])
  
      where
  
      {-# INLINE_LATE step #-}
      step gst (IterateUnfoldBFSRevOuter o ii) = do
          r <- ostep (adaptState gst) o
          case r of
              Yield a s -> do
                  i <- inject a
                  i `seq` return (Yield a (IterateUnfoldBFSRevOuter s (i:ii)))
              Skip s -> return $ Skip (IterateUnfoldBFSRevOuter s ii)
              Stop ->
                  case ii of
                      (y:ys) -> return $ Skip (IterateUnfoldBFSRevInner y ys)
                      [] -> return Stop
  
      step _ (IterateUnfoldBFSRevInner i ii) = do
          r <- istep i
          case r of
              Yield x s -> do
                  i1 <- inject x
                  i1 `seq` return $ Yield x (IterateUnfoldBFSRevInner s (i1:ii))
              Skip s -> return $ Skip (IterateUnfoldBFSRevInner s ii)
              Stop ->
                  case ii of
                      (y:ys) -> return $ Skip (IterateUnfoldBFSRevInner y ys)
                      [] -> return Stop
  
  RENAME(unfoldIterateBfsRev,altBfsUnfoldIterate)
  
  {-# ANN type IterateUnfoldBFSState Fuse #-}
  data IterateUnfoldBFSState o i =
        IterateUnfoldBFSOuter o [i]
      | IterateUnfoldBFSInner i [i] [i]
  
  -- | Like 'unfoldIterate' but uses breadth first style traversal.
  --
  -- /Pre-release/
  {-# INLINE_NORMAL bfsUnfoldIterate #-}
  bfsUnfoldIterate, unfoldIterateBfs :: Monad m =>
         Unfold m a a
      -> Stream m a
      -> Stream m a
  bfsUnfoldIterate (Unfold istep inject) (Stream ostep ost) =
      Stream step (IterateUnfoldBFSOuter ost [])
  
      where
  
      {-# INLINE_LATE step #-}
      step gst (IterateUnfoldBFSOuter o rii) = do
          r <- ostep (adaptState gst) o
          case r of
              Yield a s -> do
                  i <- inject a
                  i `seq` return (Yield a (IterateUnfoldBFSOuter s (i:rii)))
              Skip s -> return $ Skip (IterateUnfoldBFSOuter s rii)
              Stop ->
                  case reverse rii of
                      (y:ys) -> return $ Skip (IterateUnfoldBFSInner y ys [])
                      [] -> return Stop
  
      step _ (IterateUnfoldBFSInner i ii rii) = do
          r <- istep i
          case r of
              Yield x s -> do
                  i1 <- inject x
                  i1 `seq` return $ Yield x (IterateUnfoldBFSInner s ii (i1:rii))
              Skip s -> return $ Skip (IterateUnfoldBFSInner s ii rii)
              Stop ->
                  case ii of
                      (y:ys) -> return $ Skip (IterateUnfoldBFSInner y ys rii)
                      [] ->
                          case reverse rii of
                              (y:ys) -> return $ Skip (IterateUnfoldBFSInner y ys [])
                              [] -> return Stop
  
  RENAME(unfoldIterateBfs,bfsUnfoldIterate)
  
  ------------------------------------------------------------------------------
  -- Folding a tree bottom up
  ------------------------------------------------------------------------------
  
  -- | Binary BFS style reduce, folds a level entirely using the supplied fold
  -- function, collecting the outputs as next level of the tree, then repeats the
  -- same process on the next level. The last elements of a previously folded
  -- level are folded first.
  {-# INLINE_NORMAL bfsReduceIterate #-}
  bfsReduceIterate, reduceIterateBfs :: Monad m =>
      (a -> a -> m a) -> Stream m a -> m (Maybe a)
  bfsReduceIterate f (Stream step state) = go SPEC state [] Nothing
  
      where
  
      go _ st xs Nothing = do
          r <- step defState st
          case r of
              Yield x1 s -> go SPEC s xs (Just x1)
              Skip s -> go SPEC s xs Nothing
              Stop ->
                  case xs of
                      [] -> return Nothing
                      _ -> goBuf SPEC xs []
      go _ st xs (Just x1) = do
          r2 <- step defState st
          case r2 of
              Yield x2 s -> do
                  x <- f x1 x2
                  go SPEC s (x:xs) Nothing
              Skip s -> go SPEC s xs (Just x1)
              Stop ->
                  case xs of
                      [] -> return (Just x1)
                      _ -> goBuf SPEC (x1:xs) []
  
      goBuf _ [] ys = goBuf SPEC ys []
      goBuf _ [x1] ys = do
          case ys of
              [] -> return (Just x1)
              (x2:xs) -> do
                  y <- f x1 x2
                  goBuf SPEC xs [y]
      goBuf _ (x1:x2:xs) ys = do
          y <- f x1 x2
          goBuf SPEC xs (y:ys)
  
  RENAME(reduceIterateBfs,bfsReduceIterate)
  
  -- | N-Ary BFS style iterative fold, if the input stream finished before the
  -- fold then it returns Left otherwise Right. If the fold returns Left we
  -- terminate.
  --
  -- /Unimplemented/
  bfsFoldIterate ::
      Fold m a (Either a a) -> Stream m a -> m (Maybe a)
  bfsFoldIterate = undefined
  
  ------------------------------------------------------------------------------
  -- Grouping/Splitting
  ------------------------------------------------------------------------------
  
  -- s = stream state, fs = fold state
  {-# ANN type FoldManyPost Fuse #-}
  #if __GLASGOW_HASKELL__ >= 810
  type FoldManyPost :: Type -> Type -> Type -> Type -> Type
  #endif
  data FoldManyPost s fs b a
      = FoldManyPostStart s
      | FoldManyPostLoop s fs
      | FoldManyPostYield b (FoldManyPost s fs b a)
      | FoldManyPostDone
  
  -- Note that using a closed fold e.g. @Fold.take 0@, would result in an
  -- infinite stream without consuming the input.
  --
  -- We can call foldManyPost as foldMany0, but we should probably remove it.
  -- Like foldMany0, "scan" should ideally be "scan0" always resulting in a
  -- non-empty stream, and "postscan" should be called just "scan" because it is
  -- much more common. But those names cannot be changed now.
  
  -- | Like 'foldMany' but evaluates the fold even if the fold did not receive
  -- any input, therefore, always results in a non-empty output even on an empty
  -- stream (default result of the fold).
  --
  -- Example, empty stream, compare with 'foldMany':
  --
  -- >>> f = Fold.take 2 Fold.toList
  -- >>> fmany = Stream.fold Fold.toList . Stream.foldManyPost f
  -- >>> fmany $ Stream.fromList []
  -- [[]]
  --
  -- Example, last empty fold is included, compare with 'foldMany':
  --
  -- >>> fmany $ Stream.fromList [1..4]
  -- [[1,2],[3,4],[]]
  --
  -- Example, last fold non-empty, same as 'foldMany':
  --
  -- >>> fmany $ Stream.fromList [1..5]
  -- [[1,2],[3,4],[5]]
  --
  -- /Pre-release/
  --
  {-# INLINE_NORMAL foldManyPost #-}
  foldManyPost :: Monad m => Fold m a b -> Stream m a -> Stream m b
  foldManyPost (Fold fstep initial _ final) (Stream step state) =
      Stream step' (FoldManyPostStart state)
  
      where
  
      {-# INLINE consume #-}
      consume x s fs = do
          res <- fstep fs x
          return
              $ Skip
              $ case res of
                    FL.Done b -> FoldManyPostYield b (FoldManyPostStart s)
                    FL.Partial ps -> FoldManyPostLoop s ps
  
      {-# INLINE_LATE step' #-}
      step' _ (FoldManyPostStart st) = do
          r <- initial
          return
              $ Skip
              $ case r of
                    FL.Done b -> FoldManyPostYield b (FoldManyPostStart st)
                    FL.Partial fs -> FoldManyPostLoop st fs
      step' gst (FoldManyPostLoop st fs) = do
          r <- step (adaptState gst) st
          case r of
              Yield x s -> consume x s fs
              Skip s -> return $ Skip (FoldManyPostLoop s fs)
              Stop -> do
                  b <- final fs
                  return $ Skip (FoldManyPostYield b FoldManyPostDone)
      step' _ (FoldManyPostYield b next) = return $ Yield b next
      step' _ FoldManyPostDone = return Stop
  
  -- | Apply fold f1 infix separated by fold f2.
  --
  -- /Unimplemented/
  {-# INLINE_NORMAL foldManySepBy #-}
  foldManySepBy :: -- Monad m =>
      Fold m a b -> Fold m a b -> Stream m a -> Stream m b
  foldManySepBy _f1 _f2 = undefined
  
  {-# ANN type FoldMany Fuse #-}
  #if __GLASGOW_HASKELL__ >= 810
  type FoldMany :: Type -> Type -> Type -> Type -> Type
  #endif
  data FoldMany s fs b a
      = FoldManyStart s
      | FoldManyFirst fs s
      | FoldManyLoop s fs
      | FoldManyYield b (FoldMany s fs b a)
      | FoldManyDone
  
  -- XXX Nested foldMany does not fuse.
  
  -- | Apply a terminating 'Fold' repeatedly on a stream and emit the results in
  -- the output stream. If the last fold is empty, it's result is not emitted.
  -- This means if the input stream is empty the result is also an empty stream.
  -- See 'foldManyPost' for an alternate behavior which always results in a
  -- non-empty stream even if the input stream is empty.
  --
  -- Definition:
  --
  -- >>> foldMany f = Stream.parseMany (Parser.fromFold f)
  --
  -- Example, empty stream, omits the empty fold value:
  --
  -- >>> f = Fold.take 2 Fold.toList
  -- >>> fmany = Stream.fold Fold.toList . Stream.foldMany f
  -- >>> fmany $ Stream.fromList []
  -- []
  --
  -- Example, omits the last empty fold value:
  --
  -- >>> fmany $ Stream.fromList [1..4]
  -- [[1,2],[3,4]]
  --
  -- Example, last fold non-empty:
  --
  -- >>> fmany $ Stream.fromList [1..5]
  -- [[1,2],[3,4],[5]]
  --
  -- Note that using a closed fold e.g. @Fold.take 0@, would result in an
  -- infinite stream on a non-empty input stream.
  --
  {-# INLINE_NORMAL foldMany #-}
  foldMany :: Monad m => Fold m a b -> Stream m a -> Stream m b
  foldMany (Fold fstep initial _ final) (Stream step state) =
      Stream step' (FoldManyStart state)
  
      where
  
      {-# INLINE consume #-}
      consume x s fs = do
          res <- fstep fs x
          return
              $ Skip
              $ case res of
                    FL.Done b -> FoldManyYield b (FoldManyStart s)
                    FL.Partial ps -> FoldManyLoop s ps
  
      {-# INLINE_LATE step' #-}
      step' _ (FoldManyStart st) = do
          r <- initial
          return
              $ Skip
              $ case r of
                    FL.Done b -> FoldManyYield b (FoldManyStart st)
                    FL.Partial fs -> FoldManyFirst fs st
      step' gst (FoldManyFirst fs st) = do
          r <- step (adaptState gst) st
          case r of
              Yield x s -> consume x s fs
              Skip s -> return $ Skip (FoldManyFirst fs s)
              Stop -> final fs >> return Stop
      step' gst (FoldManyLoop st fs) = do
          r <- step (adaptState gst) st
          case r of
              Yield x s -> consume x s fs
              Skip s -> return $ Skip (FoldManyLoop s fs)
              Stop -> do
                  b <- final fs
                  return $ Skip (FoldManyYield b FoldManyDone)
      step' _ (FoldManyYield b next) = return $ Yield b next
      step' _ FoldManyDone = return Stop
  
  -- | Group the input stream into groups of @n@ elements each and then fold each
  -- group using the provided fold function.
  --
  -- Definition:
  --
  -- >>> groupsOf n f = Stream.foldMany (Fold.take n f)
  --
  -- Usage:
  --
  -- >>> Stream.toList $ Stream.groupsOf 2 Fold.toList (Stream.enumerateFromTo 1 10)
  -- [[1,2],[3,4],[5,6],[7,8],[9,10]]
  --
  -- This can be considered as an n-fold version of 'take' where we apply
  -- 'take' repeatedly on the leftover stream until the stream exhausts.
  --
  {-# INLINE groupsOf #-}
  groupsOf :: Monad m => Int -> Fold m a b -> Stream m a -> Stream m b
  groupsOf n f = foldMany (FL.take n f)
  
  -- Keep the argument order consistent with refoldIterateM.
  
  -- | Like 'foldMany' but for the 'Refold' type.  The supplied action is used as
  -- the initial value for each refold.
  --
  -- /Internal/
  {-# INLINE_NORMAL refoldMany #-}
  refoldMany :: Monad m => Refold m x a b -> m x -> Stream m a -> Stream m b
  refoldMany (Refold fstep inject extract) action (Stream step state) =
      Stream step' (FoldManyStart state)
  
      where
  
      {-# INLINE consume #-}
      consume x s fs = do
          res <- fstep fs x
          return
              $ Skip
              $ case res of
                    FL.Done b -> FoldManyYield b (FoldManyStart s)
                    FL.Partial ps -> FoldManyLoop s ps
  
      {-# INLINE_LATE step' #-}
      step' _ (FoldManyStart st) = do
          r <- action >>= inject
          return
              $ Skip
              $ case r of
                    FL.Done b -> FoldManyYield b (FoldManyStart st)
                    FL.Partial fs -> FoldManyFirst fs st
      step' gst (FoldManyFirst fs st) = do
          r <- step (adaptState gst) st
          case r of
              Yield x s -> consume x s fs
              Skip s -> return $ Skip (FoldManyFirst fs s)
              Stop -> return Stop
      step' gst (FoldManyLoop st fs) = do
          r <- step (adaptState gst) st
          case r of
              Yield x s -> consume x s fs
              Skip s -> return $ Skip (FoldManyLoop s fs)
              Stop -> do
                  b <- extract fs
                  return $ Skip (FoldManyYield b FoldManyDone)
      step' _ (FoldManyYield b next) = return $ Yield b next
      step' _ FoldManyDone = return Stop
  
  {-# ANN type CIterState Fuse #-}
  data CIterState s f fs b
      = CIterInit s f
      | CIterConsume s fs
      | CIterYield b (CIterState s f fs b)
      | CIterStop
  
  -- | Like 'foldIterateM' but using the 'Refold' type instead. This could be
  -- much more efficient due to stream fusion.
  --
  -- /Internal/
  {-# INLINE_NORMAL refoldIterateM #-}
  refoldIterateM ::
         Monad m => Refold m b a b -> m b -> Stream m a -> Stream m b
  refoldIterateM (Refold fstep finject fextract) initial (Stream step state) =
      Stream stepOuter (CIterInit state initial)
  
      where
  
      {-# INLINE iterStep #-}
      iterStep st action = do
          res <- action
          return
              $ Skip
              $ case res of
                    FL.Partial fs -> CIterConsume st fs
                    FL.Done fb -> CIterYield fb $ CIterInit st (return fb)
  
      {-# INLINE_LATE stepOuter #-}
      stepOuter _ (CIterInit st action) = do
          iterStep st (action >>= finject)
      stepOuter gst (CIterConsume st fs) = do
          r <- step (adaptState gst) st
          case r of
              Yield x s -> iterStep s (fstep fs x)
              Skip s -> return $ Skip $ CIterConsume s fs
              Stop -> do
                  b <- fextract fs
                  return $ Skip $ CIterYield b CIterStop
      stepOuter _ (CIterYield a next) = return $ Yield a next
      stepOuter _ CIterStop = return Stop
  
  -- | The refold @indexerBy f n@ takes an (index, len) tuple as initial input,
  -- and returns @(index + len + n, b)@ as output where @b@ is the output of the
  -- fold.
  {-# INLINE indexerBy #-}
  indexerBy :: Monad m =>
      Fold m a Int -> Int -> Refold m (Int, Int) a (Int, Int)
  indexerBy (Fold step1 initial1 extract1 _final) n =
      Refold step inject extract
  
      where
  
      inject (i, len) = do
          r <- initial1
          return $ case r of
              FL.Partial s -> FL.Partial $ Tuple' (i + len + n) s
              FL.Done l -> FL.Done (i, l)
  
      step (Tuple' i s) x = do
          r <- step1 s x
          return $ case r of
              FL.Partial s1 -> FL.Partial $ Tuple' i s1
              FL.Done len -> FL.Done (i, len)
  
      extract (Tuple' i s) = (i,) <$> extract1 s
  
  -- | Like 'splitEndBy_' but generates a stream of (index, len) tuples marking
  -- the places where the predicate matches in the stream.
  --
  -- >>> Stream.toList $ Stream.indexEndBy_ (== '/') $ Stream.fromList "/home/harendra"
  -- [(0,0),(1,4),(6,8)]
  --
  -- /Pre-release/
  {-# INLINE indexEndBy_ #-}
  indexEndBy_, indexOnSuffix :: Monad m =>
      (a -> Bool) -> Stream m a -> Stream m (Int, Int)
  indexEndBy_ predicate =
      refoldIterateM
          (indexerBy (FL.takeEndBy_ predicate FL.length) 1)
          (return (-1, 0))
  
  RENAME(indexOnSuffix,indexEndBy_)
  
  -- Alternate implementation
  {-# INLINE_NORMAL _indexEndBy_ #-}
  _indexEndBy_ :: Monad m => (a -> Bool) -> Stream m a -> Stream m (Int, Int)
  _indexEndBy_ p (Stream step1 state1) = Stream step (Just (state1, 0, 0))
  
      where
  
      {-# INLINE_LATE step #-}
      step gst (Just (st, i, len)) = i `seq` len `seq` do
        r <- step1 (adaptState gst) st
        return
          $ case r of
                Yield x s ->
                  if p x
                  then Yield (i, len + 1) (Just (s, i + len + 1, 0))
                  else Skip (Just (s, i, len + 1))
                Skip s -> Skip (Just (s, i, len))
                Stop -> if len == 0 then Stop else Yield (i, len) Nothing
      step _ Nothing = return Stop
  
  {-# DEPRECATED sliceOnSuffix "Please use indexEndBy_ instead." #-}
  sliceOnSuffix :: Monad m => (a -> Bool) -> Stream m a -> Stream m (Int, Int)
  sliceOnSuffix = indexEndBy_
  
  -- | Like 'splitEndBy' but generates a stream of (index, len) tuples marking
  -- the places where the predicate matches in the stream.
  --
  -- >>> Stream.toList $ Stream.indexEndBy (== '/') $ Stream.fromList "/home/harendra"
  -- [(0,1),(1,5),(6,8)]
  --
  -- /Pre-release/
  {-# INLINE indexEndBy #-}
  indexEndBy :: Monad m =>
      (a -> Bool) -> Stream m a -> Stream m (Int, Int)
  indexEndBy predicate =
      refoldIterateM
          (indexerBy (FL.takeEndBy predicate FL.length) 0)
          (return (0, 0))
  
  ------------------------------------------------------------------------------
  -- Stream with a cross product style monad instance
  ------------------------------------------------------------------------------
  
  -- XXX Nested performs better than the StreamK.Nested when nesting two
  -- loops, however, StreamK.Nested seems to be better for more than two nestings,
  -- need to do more perf investigation.
  
  -- | A newtype wrapper for the 'Stream' type with a cross product style monad
  -- instance.
  --
  -- A 'Monad' bind behaves like a @for@ loop:
  --
  -- >>> :{
  -- Stream.fold Fold.toList $ Stream.unNested $ do
  --     x <- Stream.Nested $ Stream.fromList [1,2]
  --     -- Perform the following actions for each x in the stream
  --     return x
  -- :}
  -- [1,2]
  --
  -- Nested monad binds behave like nested @for@ loops:
  --
  -- >>> :{
  -- Stream.fold Fold.toList $ Stream.unNested $ do
  --     x <- Stream.Nested $ Stream.fromList [1,2]
  --     y <- Stream.Nested $ Stream.fromList [3,4]
  --     -- Perform the following actions for each x, for each y
  --     return (x, y)
  -- :}
  -- [(1,3),(1,4),(2,3),(2,4)]
  --
  newtype Nested m a = Nested {unNested :: Stream m a}
          deriving (Functor, Foldable)
  
  {-# DEPRECATED CrossStream "Use Nested instead." #-}
  type CrossStream = Nested
  
  {-# DEPRECATED mkCross "Use Nested instead." #-}
  {-# INLINE mkCross #-}
  mkCross :: Stream m a -> Nested m a
  mkCross = Nested
  
  {-# INLINE unCross #-}
  unCross :: Nested m a -> Stream m a
  unCross = unNested
  
  -- Pure (Identity monad) stream instances
  deriving instance IsList (Nested Identity a)
  deriving instance (a ~ Char) => IsString (Nested Identity a)
  deriving instance Eq a => Eq (Nested Identity a)
  deriving instance Ord a => Ord (Nested Identity a)
  
  -- Do not use automatic derivation for this to show as "fromList" rather than
  -- "fromList Identity".
  instance Show a => Show (Nested Identity a) where
      {-# INLINE show #-}
      show (Nested xs) = show xs
  
  instance Read a => Read (Nested Identity a) where
      {-# INLINE readPrec #-}
      readPrec = fmap Nested readPrec
  
  ------------------------------------------------------------------------------
  -- Applicative
  ------------------------------------------------------------------------------
  
  -- Note: we need to define all the typeclass operations because we want to
  -- INLINE them.
  instance Monad m => Applicative (Nested m) where
      {-# INLINE pure #-}
      pure x = Nested (fromPure x)
  
      {-# INLINE (<*>) #-}
      (Nested s1) <*> (Nested s2) =
          Nested (crossApply s1 s2)
  
      {-# INLINE liftA2 #-}
      liftA2 f x = (<*>) (fmap f x)
  
      {-# INLINE (*>) #-}
      (Nested s1) *> (Nested s2) =
          Nested (crossApplySnd s1 s2)
  
      {-# INLINE (<*) #-}
      (Nested s1) <* (Nested s2) =
          Nested (crossApplyFst s1 s2)
  
  ------------------------------------------------------------------------------
  -- Monad
  ------------------------------------------------------------------------------
  
  instance Monad m => Monad (Nested m) where
      return = pure
  
      -- Benchmarks better with StreamD bind and pure:
      -- toList, filterAllout, *>, *<, >> (~2x)
      --
  
      -- Benchmarks better with CPS bind and pure:
      -- Prime sieve (25x)
      -- n binds, breakAfterSome, filterAllIn, state transformer (~2x)
      --
      {-# INLINE (>>=) #-}
      (>>=) (Nested m) f = Nested (concatMap (unNested . f) m)
  
      {-# INLINE (>>) #-}
      (>>) = (*>)
  
  ------------------------------------------------------------------------------
  -- Transformers
  ------------------------------------------------------------------------------
  
  instance (MonadIO m) => MonadIO (Nested m) where
      liftIO x = Nested (fromEffect $ liftIO x)
  
  instance MonadTrans Nested where
      {-# INLINE lift #-}
      lift x = Nested (fromEffect x)
  
  instance (MonadThrow m) => MonadThrow (Nested m) where
      throwM = lift . throwM
  
  ------------------------------------------------------------------------------
  -- Utilities
  ------------------------------------------------------------------------------
  
  -- | Inlined definition. Without the inline "serially/parser/take" benchmark
  -- degrades and parseMany does not fuse. Even using "inline" at the callsite
  -- does not help.
  {-# INLINE splitAt #-}
  splitAt :: String -> Int -> [a] -> ([a],[a])
  splitAt desc n ls
    | n < 0 = seekOver n
    | n == 0 = ([], ls)
    | otherwise = splitAt' n ls
  
      where
  
      splitAt' :: Int -> [a] -> ([a], [a])
      splitAt' 0  []     = ([], [])
      splitAt' m  []     = seekUnder n m
      splitAt' 1  (x:xs) = ([x], xs)
      splitAt' m  (x:xs) = (x:xs', xs'')
  
          where
  
          (xs', xs'') = splitAt' (m - 1) xs
  
      seekOver x =
          error $ desc ++ ": bug in parser, seeking ["
              ++ show (negate x)
              ++ "] elements in future"
  
      seekUnder x y =
          error $ desc ++ ": bug in parser, backtracking ["
              ++ show x
              ++ "] elements. Goes ["
              ++ show y
              ++ "] elements beyond backtrack buffer"
  