diff --git a/Streaming.hs b/Streaming.hs --- a/Streaming.hs +++ b/Streaming.hs @@ -17,9 +17,11 @@ wrap, -- * Transforming streams + decompose, maps, mapsM, distribute, + eithers, -- * Inspecting a stream inspect, @@ -30,20 +32,19 @@ interleaves, -- * Eliminating a 'Stream' - intercalates, - concats, iterTM, iterT, destroy, mapsM_, - runEffect, + run, -- * Splitting and joining 'Stream's splitsAt, takes, chunksOf, concats, - + intercalates, + -- * Base functor for streams of individual items Of (..), lazily, diff --git a/Streaming/Internal.hs b/Streaming/Internal.hs --- a/Streaming/Internal.hs +++ b/Streaming/Internal.hs @@ -29,8 +29,10 @@ -- * Transforming streams , maps , mapsM + , decompose , mapsM_ - , runEffect + , eithers + , run , distribute -- * Splitting streams @@ -59,12 +61,13 @@ import Data.Foldable ( Foldable(..) ) import Data.Traversable import Control.Monad.Morph -import Data.Monoid +import Data.Monoid (Monoid (..), (<>)) import Data.Functor.Identity import GHC.Exts ( build ) import Data.Data ( Data, Typeable ) import Prelude hiding (splitAt) import Data.Functor.Compose +import Data.Functor.Sum {- $stream The 'Stream' data type is equivalent to @FreeT@ and can represent any effectful @@ -123,14 +126,40 @@ Delay m -> Delay (liftM loop m) Step f -> Step (fmap loop f) {-# INLINABLE (>>) #-} - stream >>= f = loop stream where - loop stream0 = case stream0 of - Step f -> Step (fmap loop f) - Delay m -> Delay (liftM loop m) - Return r -> f r - {-# INLINABLE (>>=) #-} + (>>=) = _bind + -- stream >>= f = + -- loop stream where + -- loop stream0 = case stream0 of + -- Step fstr -> Step (fmap loop fstr) + -- Delay m -> Delay (liftM loop m) + -- Return r -> f r + -- {-# INLINABLE (>>=) #-} + fail = lift . fail + + +_bind + :: (Functor f, Monad m) + => Stream f m r + -> (r -> Stream f m s) + -> Stream f m s +_bind p0 f = go p0 where + go p = case p of + Step fstr -> Step (fmap go fstr) + Delay m -> Delay (m >>= \s -> return (go s)) + Return r -> f r + +{-# RULES + "_bind (Step fstr) f" forall fstr f . + _bind (Step fstr) f = Step (fmap (\p -> _bind p f) fstr); + "_bind (Delay m) f" forall m f . + _bind (Delay m) f = Delay (m >>= \p -> return (_bind p f)); + "_bind (Return r) f" forall r f . + _bind (Return r) f = f r; + #-} + + instance (Functor f, Monad m) => Applicative (Stream f m) where pure = Return {-# INLINE pure #-} @@ -283,7 +312,12 @@ Step f -> Step (phi (fmap loop f)) {-# INLINABLE maps #-} --- | Map layers of one functor to another with a transformation involving the base monad +{- | Map layers of one functor to another with a transformation involving the base monad + @maps@ is more fundamental than @mapsM@, which is best understood as a convenience + for effecting this frequent composition: + +> mapsM phi = decompose . maps (Compose . phi) +-} mapsM :: (Monad m, Functor f) => (forall x . f x -> m (g x)) -> Stream f m r -> Stream g m r mapsM phi = loop where loop stream = case stream of @@ -292,22 +326,41 @@ Step f -> Delay (liftM Step (phi (fmap loop f))) {-# INLINABLE mapsM #-} +{-| Resort a succession of layers of the form @m (f x)@. Though @mapsM@ + is best understood as: +> mapsM phi = decompose . maps (Compose . phi) + + we could as well define @decompose@ by @mapsM@: + +> decompose = mapsM getCompose + +-} +decompose :: (Monad m, Functor f) => Stream (Compose m f) m r -> Stream f m r +decompose = loop where + loop stream = case stream of + Return r -> Return r + Delay m -> Delay (liftM loop m) + Step (Compose mstr) -> Delay $ do + str <- mstr + return (Step (fmap loop str)) + + {-| Run the effects in a stream that merely layers effects. -} -runEffect :: Monad m => Stream m m r -> m r -runEffect = loop where +run :: Monad m => Stream m m r -> m r +run = loop where loop stream = case stream of Return r -> return r Delay m -> m >>= loop Step mrest -> mrest >>= loop -{-# INLINABLE runEffect #-} +{-# INLINABLE run #-} {-| Map each layer to an effect in the base monad, and run them all. -} mapsM_ :: (Functor f, Monad m) => (forall x . f x -> m x) -> Stream f m r -> m r -mapsM_ f str = runEffect (maps f str) +mapsM_ f str = run (maps f str) {-# INLINABLE mapsM_ #-} @@ -424,30 +477,7 @@ Step fs -> Step $ Step $ fmap (fmap loop . splitsAt (n0-1)) fs {-# INLINABLE chunksOf #-} -{- | Make it possible to \'run\' the underlying transformed monad. A simple - minded example might be: - -> debugFibs = flip runStateT 1 $ distribute $ loop 1 where -> loop n = do -> S.yield n -> s <- lift get -> liftIO $ putStr "Current state is: " >> print s -> lift $ put (s + n :: Int) -> loop s - ->>> S.print $ S.take 4 $ S.drop 4 $ debugFibs -Current state is: 1 -Current state is: 2 -Current state is: 3 -Current state is: 5 -5 -Current state is: 8 -8 -Current state is: 13 -13 -Current state is: 21 -21 - +{- | Make it possible to \'run\' the underlying transformed monad. -} distribute :: (Monad m, Functor f, MonadTrans t, MFunctor t, Monad (t (Stream f m))) => Stream f (t m) r -> t (Stream f m) r @@ -591,7 +621,16 @@ {-# INLINE interleaves #-} - +eithers :: (Monad m, Applicative h) => + (forall x . f x -> h x) -> (forall x . g x -> h x) -> Stream (Sum f g) m r -> Stream h m r +eithers f g = loop where + loop str = case str of + Return r -> Return r + Delay m -> Delay (liftM loop m) + Step str' -> case str' of + InL s -> Step (fmap loop (f s)) + InR t -> Step (fmap loop (g t)) + diff --git a/Streaming/Prelude.hs b/Streaming/Prelude.hs --- a/Streaming/Prelude.hs +++ b/Streaming/Prelude.hs @@ -2,15 +2,7 @@ simplify and optimize the conception of Producer manipulation contained in Pipes.Group, Pipes.Parse and the like. This is very simple and unmysterious; it is independent of piping and conduiting, and can be used with any - rational \"streaming IO\" system. - - Some interoperation incantations would be e.g. - -> Pipes.unfoldr Streaming.next :: Stream (Of a) m r -> Producer a m r -> Streaming.unfoldr Pipes.next :: Producer a m r -> Stream (Of a) m r -> Streaming.reread IOStreams.read :: InputStream a -> Stream (Of a) IO () -> IOStreams.unfoldM Streaming.uncons :: Stream (Of a) IO () -> IO (InputStream a) -> Conduit.unfoldM Streaming.uncons :: Stream (Of a) m () -> Source m a + rational \"streaming IO\" system. Import qualified thus: @@ -150,6 +142,9 @@ import Data.Data ( Data, Typeable ) import Data.Functor.Identity import Control.Monad.Trans +import Control.Applicative (Applicative (..)) +import Data.Functor (Functor (..), (<$)) + import qualified Prelude as Prelude import Data.Foldable (Foldable) import Data.Traversable (Traversable) @@ -165,14 +160,47 @@ import qualified System.IO as IO import Foreign.C.Error (Errno(Errno), ePIPE) import Control.Exception (throwIO, try) - +import Data.Monoid (Monoid (..)) -- | A left-strict pair; the base functor for streams of individual elements. data Of a b = !a :> b - deriving (Data, Eq, Foldable, Functor, Ord, + deriving (Data, Eq, Foldable, Ord, Read, Show, Traversable, Typeable) -infixr 4 :> +infixr 5 :> +instance (Monoid a, Monoid b) => Monoid (Of a b) where + mempty = mempty :> mempty + {-#INLINE mempty #-} + mappend (m :> w) (m' :> w') = mappend m m' :> mappend w w' + {-#INLINE mappend #-} + mconcat = foldr mappend mempty + {-#INLINE mconcat #-} + +instance Functor (Of a) where + fmap f (a :> x) = a :> f x + {-#INLINE fmap #-} + a <$ (b :> x) = b :> a + {-#INLINE (<$) #-} + +instance Monoid a => Applicative (Of a) where + pure x = mempty :> x + {-#INLINE pure #-} + m :> f <*> m' :> x = mappend m m' :> f x + {-#INLINE (<*>) #-} + m :> x *> m' :> y = mappend m m' :> y + {-#INLINE (*>) #-} + m :> x <* m' :> y = mappend m m' :> x + {-#INLINE (<*) #-} + +instance Monoid a => Monad (Of a) where + return x = mempty :> x + {-#INLINE return #-} + m :> x >> m' :> y = mappend m m' :> y + {-#INLINE (>>) #-} + m :> x >>= f = let m' :> y = f x in mappend m m' :> y + {-#INLINE (>>=) #-} + + lazily :: Of a b -> (a,b) lazily = \(a:>b) -> (a,b) {-# INLINE lazily #-} @@ -186,6 +214,8 @@ snd' :: Of a b -> b snd' (a :> b) = b + + {-| Break a sequence when a element falls under a predicate, keeping the rest of the stream as the return value. @@ -327,7 +357,7 @@ -- | Ignore the first n elements of a stream, but carry out the actions drop :: (Monad m) => Int -> Stream (Of a) m r -> Stream (Of a) m r drop = loop where - loop n stream + loop !n stream | n <= 0 = stream | otherwise = case stream of Return r -> Return r @@ -363,12 +393,10 @@ {- | Stream the elements of a foldable container. ->>> S.print $ S.map (*100) $ each [1..3] >> yield 4 -0 +>>> S.print $ S.map (*100) $ each [1..3] 100 200 300 -400 >>> S.print $ S.map (*100) $ each [1..3] >> lift readLn >>= yield 100 @@ -783,7 +811,7 @@ -} product' :: (Monad m, Num a) => Stream (Of a) m r -> m (Of a r) product' = fold' (*) 1 id -{-# INLINAE product' #-} +{-# INLINE product' #-} -- --------------- -- read diff --git a/streaming.cabal b/streaming.cabal --- a/streaming.cabal +++ b/streaming.cabal @@ -1,29 +1,29 @@ name: streaming -version: 0.1.0.15 +version: 0.1.0.16 cabal-version: >=1.10 build-type: Simple synopsis: a free monad transformer optimized for streaming applications -description: @Stream@ can be used wherever @FreeT@ is used. The compiler's +description: @Stream@ can be used wherever + + is used. The compiler's standard range of optimizations work better for operations written in terms of `Stream`. @FreeT f m r@ / @Stream f m r@ is of course extremely general, and many functor-general combinators are exported by the general module @Streaming@. . - See the examples in @Streaming.Prelude@ for a sense of how - simple the library is to use and think about. - That module is focused on employment with such base functors - (readings of the @f@ in @Stream f m r@) that express - different forms of effectful sequences. Some of these appear - elsewhere under titles like + @Streaming.Prelude@ is focused on elementary /streaming/ applications. + Here the free iteration of the \'base\' functors + (readings of the @f@ in @Stream f m r@) express + forms of effectful sequence or succession. Some of types in question + appear in the streaming IO libraries under titles like . > pipes: Producer a m r, Producer a m (Producer a m r), FreeT (Producer a m) m r > io-streams: InputStream a, Generator a r > conduit: Source m a, ConduitM () o m r . - and the like. @Streaming.Prelude@ closely follows @Pipes.Prelude@, but cleverly /omits the pipes/: + @Streaming.Prelude@ closely follows @Pipes.Prelude@, but cleverly /omits the pipes/: . - > ghci> S.stdoutLn $ S.take 2 S.stdinLn > let's > let's @@ -40,7 +40,7 @@ > 49 . Somehow, we didn't even need a four-character operator for that, nor advice - about best practices; just ordinary Haskell common sense. + about best practices! - just ordinary Haskell common sense. . The simplest form of interoperation with @@ -49,7 +49,9 @@ > Pipes.unfoldr Streaming.next :: Stream (Of a) m r -> Producer a m r > Streaming.unfoldr Pipes.next :: Producer a m r -> Stream (Of a) m r . - Interoperation with + (@streaming@ can be mixed with @pipes@ wherever @pipes@ + itself employs @Control.Monad.Trans.Free@; speedups are frequently + appreciable.) Interoperation with is thus: . @@ -102,7 +104,7 @@ DeriveFunctor, DeriveTraversable, UndecidableInstances - build-depends: base >=4.6 && <4.9 + build-depends: base >=4.6 && <5 , mtl >=2.1 && <2.3 , mmorph >=1.0 && <1.2 , transformers >=0.3 && <0.5