@@ -0,0 +1,112 @@
1+{-# LANGUAGE FlexibleContexts #-}
2+{-# LANGUAGE DeriveDataTypeable #-}
3+module Data.Conduit
4+ ( -- * Helper types
5+ Stream (..)
6+ , SinkResult (..)
7+ , ConduitResult (..)
8+ -- * Main types
9+ , Source (..)
10+ , Sink (..)
11+ , Conduit
12+ -- * Connect pieces together
13+ , connect
14+ , connectLeft
15+ , connectRight
16+ -- * Helper functions
17+ , mkSource
18+ ) where
19+
20+import Control.Monad.Trans.Resource
21+import Control.Monad.Trans.Control (MonadBaseControl)
22+import Control.Monad.Base (liftBase)
23+import Control.Exception.Lifted (throwIO, Exception)
24+import qualified Data.IORef as I
25+import Data.Typeable (Typeable)
26+import Control.Monad (liftM)
27+
28+data Stream a = EOF | Chunks [a]
29+ deriving Show
30+instance Functor Stream where
31+ fmap _ EOF = EOF
32+ fmap f (Chunks l) = Chunks (map f l)
33+
34+data SinkResult input output = SinkResult [input] (Maybe output)
35+instance Functor (SinkResult input) where
36+ fmap f (SinkResult input moutput) = SinkResult input (fmap f moutput)
37+
38+data ConduitResult input output = ConduitResult [input] (Stream output)
39+
40+data SourceInside m a = SourceInside
41+ { sourcePull :: ResourceT m (Stream a)
42+ , sourcePush :: [a] -> ResourceT m ()
43+ }
44+newtype Source m a = Source (ResourceT m (SourceInside m a))
45+
46+type SinkInside a m b = (Stream a -> ResourceT m (SinkResult a b))
47+newtype Sink input m output = Sink (ResourceT m (SinkInside input m output))
48+instance Monad m => Functor (Sink input m) where
49+ fmap f (Sink msink) = Sink $ do
50+ sink <- msink
51+ return $ \stream -> liftM (fmap f) (sink stream)
52+
53+type Conduit a m b = Stream a -> ResourceT m (ConduitResult a b)
54+
55+mkSource :: MonadBaseControl IO m
56+ => ResourceT m (Stream a)
57+ -> Source m a
58+mkSource pull = Source $ do
59+ buffer <- liftBase $ I.newIORef []
60+ return SourceInside
61+ { sourcePull = do
62+ x <- liftBase $ I.atomicModifyIORef buffer (\a -> ([], a))
63+ if null x
64+ then pull
65+ else return $ Chunks x
66+ , sourcePush = \s ->
67+ if null s
68+ then return ()
69+ else liftBase $
70+ I.atomicModifyIORef buffer (\a -> (s ++ a, ()))
71+ }
72+
73+connect :: MonadBaseControl IO m => Source m a -> Sink a m b -> ResourceT m b
74+connect (Source msource) (Sink msink) = do
75+ sink <- msink
76+ source <- msource
77+ connect' source sink
78+ where
79+ connect' source sink = do
80+ stream <- sourcePull source
81+ SinkResult leftover mres <- sink stream
82+ sourcePush source leftover
83+ case (stream, mres) of
84+ (_, Just res) -> return res
85+ (Chunks _, Nothing) -> connect' source sink
86+ (EOF, Nothing) -> liftBase (throwIO NothingAfterEOF)
87+
88+data NothingAfterEOF = NothingAfterEOF
89+ deriving (Show, Typeable)
90+instance Exception NothingAfterEOF
91+
92+connectLeft :: MonadBaseControl IO m
93+ => Source m a
94+ -> Conduit a m b
95+ -> Source m b
96+connectLeft (Source msource) pipe = Source $ do
97+ source <- msource
98+ source' <- unSource $ mkSource $ do
99+ ConduitResult leftover result <- sourcePull source >>= pipe
100+ sourcePush source leftover
101+ return result
102+ return source'
103+ where
104+ unSource (Source s) = s
105+
106+connectRight :: MonadBaseControl IO m => Conduit a m b -> Sink b m c -> Sink a m c
107+connectRight pipe (Sink msink) = Sink $ do
108+ sink <- msink
109+ return $ \stream -> do
110+ ConduitResult leftover stream' <- pipe stream
111+ SinkResult _thisislost mres <- sink stream'
112+ return $ SinkResult leftover mres