simple-conduit 0.1.0 → 0.2.0
raw patch · 2 files changed
+196/−25 lines, 2 filesdep +foldldep +stm
Dependencies added: foldl, stm
Files
- Conduit/Simple.hs +193/−24
- simple-conduit.cabal +3/−1
Conduit/Simple.hs view
@@ -2,6 +2,7 @@ {-# LANGUAGE TypeFamilies #-} {-# LANGUAGE TupleSections #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-} -- | Please see the project README for more details: --@@ -15,19 +16,23 @@ import Control.Applicative import Control.Concurrent.Async.Lifted+import Control.Concurrent.STM import Control.Exception.Lifted+import Control.Foldl import Control.Monad hiding (mapM) import Control.Monad.Base import Control.Monad.Catch hiding (bracket) import Control.Monad.IO.Class+import Control.Monad.Morph import Control.Monad.Primitive-import Control.Monad.Trans.Class import Control.Monad.Trans.Control import Control.Monad.Trans.Either import Control.Monad.Trans.State import Data.Bifunctor import Data.Builder import Data.ByteString hiding (hPut, putStrLn)+import Data.Foldable+import Data.Functor.Identity import Data.IOData import Data.MonoTraversable import Data.Monoid@@ -38,7 +43,6 @@ import Data.Text import Data.Textual.Encoding import Data.Traversable-import Data.Vector.Generic hiding (mapM, foldM, modify) import Data.Word import Prelude hiding (mapM) import System.FilePath ((</>))@@ -58,6 +62,57 @@ type Conduit a m b r = Source m a r -> Source m b r type Sink a m r = Source m a r -> m r +-- | When wrapped in a 'SourceWrapper' using 'wrap', Sources offer a number of+-- typeclass instances, one of which is Monad. As a Monad, it behaves very+-- much list the list monad: the value bound is each element of the+-- iteration in turn.+--+-- @+-- sinkList $ getSource $ do+-- x <- wrap $ yieldMany [1..3]+-- y <- wrap $ yieldMany [4..6]+-- wrap $ yieldOne (x, y)+--+-- ==> [(1,4),(1,5),(1,6),(2,4),(2,5),(2,6),(3,4),(3,5),(3,6)]+-- @+newtype SourceWrapper m a = SourceWrapper { getSource :: forall r. Source m a r }++wrap :: (forall r. Source m a r) -> SourceWrapper m a+wrap = SourceWrapper++instance Monad m => Monoid (SourceWrapper m a) where+ mempty = SourceWrapper $ \z _ -> return z+ SourceWrapper x `mappend` SourceWrapper y = SourceWrapper $ x <+> y++instance Foldable (SourceWrapper Identity) where+ foldMap f (SourceWrapper await) =+ runIdentity $ resolve await mempty $ \r x -> return $ r <> f x++instance Functor (SourceWrapper m) where+ fmap f (SourceWrapper await) =+ SourceWrapper $ \z yield -> await z $ \r x -> yield r (f x)++instance Applicative (SourceWrapper m) where+ pure x = SourceWrapper $ \z yield -> yield z x+ SourceWrapper f <*> SourceWrapper g =+ SourceWrapper $ \z yield ->+ f z $ \r x ->+ g r $ \r' y ->+ yield r' (x y)++instance Monad (SourceWrapper m) where+ return = pure+ SourceWrapper await >>= f =+ SourceWrapper $ \z yield ->+ await z $ \r x ->+ getSource (f x) r $ \r' y ->+ yield r' y++newtype SinkWrapper a m r = SinkWrapper { getSink :: SourceWrapper m a -> m r }++instance Monad m => Functor (SinkWrapper a m) where+ fmap f (SinkWrapper k) = SinkWrapper $ \await -> f `liftM` k await+ -- | Promote any sink to a source. This can be used as if it were a source -- transformer (aka, a conduit): --@@ -89,15 +144,6 @@ ($$) = flip ($) {-# INLINE ($$) #-} --- | Since Sources are not Monads in this library (as they are in the full--- conduit library), they can be sequentially "chained" using this append--- operator. If Source were a newtype, we could make it an instance of--- Monoid.-infixr 3 <+>-(<+>) :: Monad m => Source m a r -> Conduit a m a r-x <+> y = \r f -> flip y f =<< x r f-{-# INLINE (<+>) #-}- -- | This is just like 'Control.Monad.Trans.Either.bimapEitherT', but it only -- requires a 'Monad' constraint rather than 'Functor'. rewrap :: Monad m => (a -> b) -> EitherT a m a -> EitherT b m b@@ -758,8 +804,7 @@ pure x = ZipSource $ yieldOne x ZipSource l <*> ZipSource r = ZipSource (zipSourceApp l r) --- | Sequence a collection of sources, feeding them all the same input and--- yielding a collection of their results.+-- | Sequence a collection of sources. -- -- >>> sinkList $ sequenceSources [yieldOne 1, yieldOne 2, yieldOne 3] -- [[1,2,3]]@@ -767,19 +812,42 @@ => f (Source m a r) -> Source m (f a) r sequenceSources = getZipSource . sequenceA . fmap ZipSource +-- | Since Sources are not Monads in this library (as they are in the full+-- conduit library), they can be sequentially "chained" using this append+-- operator. If Source were a newtype, we could make it an instance of+-- Monoid.+infixr 3 <+>+(<+>) :: Monad m => Source m a r -> Conduit a m a r+x <+> y = \r f -> flip y f =<< x r f+{-# INLINE (<+>) #-}++instance MFunctor (EitherT s) where+ hoist f (EitherT m) = EitherT $ f m++-- | Zip sinks together. This function may be used multiple times:+--+-- >>> let mySink s await => resolve await () $ \() x -> liftIO $ print $ s <> show x+-- >>> zipSinks sinkList (zipSinks (mySink "foo") (mySink "bar")) $ yieldMany [1,2,3]+-- "foo: 1"+-- "bar: 1"+-- "foo: 2"+-- "bar: 2"+-- "foo: 3"+-- "bar: 3"+-- ([1,2,3],((),())) zipSinks :: Monad m => (Source (StateT (r', s) m) i s -> StateT (r', s) m r) -> (Source (StateT (r', s) m) i s' -> StateT (r', s) m r')- -> (forall t. Source (StateT (r', s) m) i t)- -> m (r, r')+ -> Source m i (s, s') -> m (r, r') zipSinks x y await = do- let i = (error "accessing r", error "accessing r'")+ let i = (error "accessing r'", error "accessing s") flip evalStateT i $ do r <- x $ \rx yieldx -> do r' <- lift $ y $ \ry yieldy -> EitherT $ do- eres <- runEitherT $ await (rx, ry) $ \(rx', ry') u -> do- x' <- rewrap (, ry') $ yieldx rx' u- y' <- rewrap (rx' ,) $ yieldy ry' u+ st <- get+ eres <- lift $ runEitherT $ await (rx, ry) $ \(rx', ry') u -> do+ x' <- stripS st $ rewrap (, ry') $ yieldx rx' u+ y' <- stripS st $ rewrap (rx' ,) $ yieldy ry' u return (fst x', snd y') let (s, s') = either id id eres modify (\(b, _) -> (b, s))@@ -789,6 +857,9 @@ gets snd r' <- gets fst return (r, r')+ where+ stripS :: (MFunctor t, Monad n) => b1 -> t (StateT b1 n) b -> t n b+ stripS s = hoist (`evalStateT` s) newtype ZipSink i m r s = ZipSink { getZipSink :: Source m i r -> m s } @@ -797,22 +868,120 @@ instance Monad m => Applicative (ZipSink i m r) where pure x = ZipSink $ \_ -> return x- ZipSink f <*> ZipSink x =- ZipSink $ \await -> f await `ap` x await+ ZipSink f <*> ZipSink x = ZipSink $ \await -> f await `ap` x await -- | Send incoming values to all of the @Sink@ providing, and ultimately--- coalesce together all return values.+-- coalesce together all return values. -- -- Implemented on top of @ZipSink@, see that data type for more details.------ Since 1.0.13 sequenceSinks :: (Traversable f, Monad m) => f (Source m i r -> m s) -> Source m i r -> m (f s) sequenceSinks = getZipSink . sequenceA . fmap ZipSink +-- infixr 3 <*>+-- (<*>) :: Monad m+-- => (Source (StateT (r', s) m) i s -> StateT (r', s) m r)+-- -> (Source (StateT (r', s) m) i s' -> StateT (r', s) m r')+-- -> Source m i (s, s') -> m (r, r')+-- (<*>) = zipSinks+-- {-# INLINE (<*>) #-}++-- zipConduitApp :: Monad m => Conduit a m (x -> y) r -> Conduit a m x r -> Conduit a m y r+-- zipConduitApp f arg z yield = f z $ \r x -> arg r $ \_ y -> yield z (x y)++-- newtype ZipConduit a m r b = ZipConduit { getZipConduit :: Conduit a m b r }++-- instance Monad m => Functor (ZipConduit a m r) where+-- fmap f (ZipConduit p) = ZipConduit $ \z yield -> p z $ \r x -> yield r (f x)++-- instance Monad m => Applicative (ZipConduit a m r) where+-- pure x = ZipConduit $ yieldOne x+-- ZipConduit l <*> ZipConduit r = ZipConduit (zipConduitApp l r)++-- -- | Sequence a collection of sources.+-- --+-- -- >>> sinkList $ sequenceConduits [yieldOne 1, yieldOne 2, yieldOne 3]+-- -- [[1,2,3]]+-- sequenceConduits :: (Traversable f, Monad m)+-- => f (Conduit a m b r) -> Conduit a m (f b) r+-- sequenceConduits = getZipConduit . sequenceA . fmap ZipConduit+ asyncC :: (MonadBaseControl IO m, Monad m) => (a -> m b) -> Conduit a m (Async (StM m b)) r asyncC f await k yield = do res <- async $ await k $ \r x -> yield r =<< lift (async (f x)) wait res++-- | Convert a 'Control.Foldl.FoldM' fold abstraction into a Sink.+--+-- NOTE: This requires ImpredicativeTypes in the code that uses it.+--+-- >>> fromFoldM (FoldM ((return .) . (+)) (return 0) return) $ yieldMany [1..10]+-- 55+fromFoldM :: Monad m => FoldM m a b -> (forall r. Source m a r) -> m b+fromFoldM (FoldM step initial final) await =+ initial >>= flip (resolve await) ((lift .) . step) >>= final++-- | Convert a Sink into a 'Control.Foldl.FoldM', passing it into a+-- continuation.+--+-- >>> toFoldM sumC (\f -> Control.Foldl.foldM f [1..10])+-- 55+toFoldM :: Monad m+ => Sink a m r -> (FoldM (EitherT r m) a r -> EitherT r m r) -> m r+toFoldM sink f = sink $ \k yield -> f $ FoldM yield (return k) return++-- | A Source for exhausting a TChan, but blocks if it is initially empty.+sourceTChan :: TChan a -> Source STM a r+sourceTChan chan z yield = go z+ where+ go r = do+ x <- lift $ readTChan chan+ r' <- yield r x+ mt <- lift $ isEmptyTChan chan+ if mt+ then return r'+ else go r'++sourceTQueue :: TQueue a -> Source STM a r+sourceTQueue chan z yield = go z+ where+ go r = do+ x <- lift $ readTQueue chan+ r' <- yield r x+ mt <- lift $ isEmptyTQueue chan+ if mt+ then return r'+ else go r'++sourceTBQueue :: TBQueue a -> Source STM a r+sourceTBQueue chan z yield = go z+ where+ go r = do+ x <- lift $ readTBQueue chan+ r' <- yield r x+ mt <- lift $ isEmptyTBQueue chan+ if mt+ then return r'+ else go r'++untilMC :: Monad m => m a -> m Bool -> Source m a r+untilMC m f z yield = go z+ where+ go r = do+ x <- lift m+ r' <- yield r x+ cont <- lift f+ if cont+ then go r'+ else return r'++whileMC :: Monad m => m Bool -> m a -> Source m a r+whileMC f m z yield = go z+ where+ go r = do+ cont <- lift f+ if cont+ then lift m >>= yield r >>= go+ else return r
simple-conduit.cabal view
@@ -1,5 +1,5 @@ Name: simple-conduit-Version: 0.1.0+Version: 0.2.0 Synopsis: A simple streaming library based on composing monadic folds. Description: @simple-conduit@ follows a similar UI to the more capable @conduit@ library, but reduces the scope of what it can solve donw to what can be expressed by chaining monadic folds that allow for early termination. This allows for more predictable resource management behavior, at the cost of not allowing scenarios that @conduit@ is better designed.@@ -22,6 +22,7 @@ , either , exceptions , filepath+ , foldl , lifted-async , lifted-base >= 0.1 , mmorph@@ -30,6 +31,7 @@ , mtl , mwc-random , primitive+ , stm , streaming-commons , text , transformers >= 0.2.2 && < 0.5