Perhaps use explicit "fold" style

[tomjaguarpaw]

Many streaming functions factor through a "fold" that guarantees that they consume their arguments sequentially and transform them with monad homomorphisms before re-emitting them. (cf. foldMap). Is this style something you're interested in pursuing, either internally or in the public API?

{-# LANGUAGE Rank2Types #-}
{-# LANGUAGE ExistentialQuantification #-}

import qualified Streaming as S
import qualified Streaming.Prelude as S
import qualified Control.Monad.Free as F
import qualified Control.Monad.Trans.Writer as W
import qualified Data.Monoid as M

data Fold f m n = forall o. Monad o => Fold (forall a. f a -> o a)
                                            (forall a. m a -> o a)
                                            (forall a. o a -> n a)

fold :: (Functor f,
         Monad n,
         Monad m)
     => Fold f m n
     -> S.Stream f m a
     -> n a
fold fold s = case fold of
  Fold ff fm fo -> fo (S.iterT f (S.hoist fm s))
    where f = S.join . ff

erase :: Monad m
      => Fold (S.Of t) m (S.Stream S.Identity m)
erase = Fold f S.lift id
  where f (_ S.:> rest) = S.yields (S.Identity rest)

filter :: Monad m
       => t
       -> Fold (S.Of Bool) m (S.Stream (S.Of Bool) m)
filter pred = Fold f S.lift id
  where f (a S.:> rest) = if a
                          then S.yield a >> return rest
                          else return rest

with :: (Monad m, Functor f)
     => (a -> f x)
     -> Fold (S.Of a) m (S.Stream f m)
with g = Fold f S.lift id
  where f (a S.:> rest) = S.yields (g a) >> return rest

-- This one needs a little post processing
elem :: (Eq a, Monad m)
     => a
     -> Fold (S.Of a) m (W.WriterT M.Any m)
elem a' = Fold f S.lift id
  where f (a S.:> rest) = if a == a'
                          then W.tell (M.Any True) >> return rest
                          else return rest

[chessai]

I'm confused as to what the actual benefit is here - mainly because I think I misunderstand you. What does "consume their arguments sequentially" mean? What are the arguments here? Arguments to the functions defined in streaming? What does sequentially mean? The way it's worded makes it seem to refer to the arguments, but I'm not sure what arguments is, so that doesn't exactly make sense. Do you mean 'linearly' instead of 'sequentially'?

[tomjaguarpaw]

The easiest way to understand what I mean is to look at the code. There exists a function fold such that many of the standard streaming functions are fold . f for some f.