streaming-utils (empty) → 0.1.0.0
raw patch · 6 files changed
+672/−0 lines, 6 filesdep +attoparsecdep +basedep +bytestringsetup-changed
Dependencies added: attoparsec, base, bytestring, http-client, http-client-tls, mtl, pipes, streaming, streaming-bytestring, transformers
Files
- Data/Attoparsec/ByteString/Streaming.hs +104/−0
- Data/ByteString/Streaming/HTTP.hs +132/−0
- LICENSE +30/−0
- Setup.hs +2/−0
- Streaming/Pipes.hs +368/−0
- streaming-utils.cabal +36/−0
+ Data/Attoparsec/ByteString/Streaming.hs view
@@ -0,0 +1,104 @@+module Data.Attoparsec.ByteString.Streaming+ (Message+ , parse+ , parsed+ , module Data.Attoparsec.ByteString+ + )+ where++import qualified Data.ByteString as B+import qualified Data.Attoparsec.ByteString as A+import qualified Data.Attoparsec.Internal.Types as T+import Data.Attoparsec.ByteString+ hiding (IResult(..), Result, eitherResult, maybeResult,+ parse, parseWith, parseTest)++import Streaming hiding (concats, unfold)+import Streaming.Internal (Stream (..))+import Data.ByteString.Streaming+import Data.ByteString.Streaming.Internal+import Data.Monoid ++type Message = ([String], String)++{- | The result of a parse (@Either a ([String], String)@), with the unconsumed byte stream.++>>> (r,rest1) <- parse (A.scientific <* A.many' A.space) $ "12.3 4.56 78." >> "3"+>>> print r+Left 12.3+>>> (s,rest2) <- parse (A.scientific <* A.many' A.space) rest1+>>> print s+Left 4.56+>>> (t,rest3) <- parse (A.scientific <* A.many' A.space) rest2+>>> print t+Left 78.3+>>> Q.putStrLn rest3++-}+parse :: Monad m + => A.Parser a + -> ByteString m x -> m (Either a Message, ByteString m x)+parse p s = case s of+ Chunk x xs -> go (A.parse p x) xs+ Empty r -> go (A.parse p B.empty) (Empty r)+ Go m -> m >>= parse p+ where+ go (T.Fail x stk msg) ys = return $ (Right (stk, msg), Chunk x ys)+ go (T.Done x r) ys = return $ (Left r, Chunk x ys)+ go (T.Partial k) (Chunk y ys) = go (k y) ys+ go (T.Partial k) (Go m) = m >>= go (T.Partial k)+ go (T.Partial k) blank = go (k B.empty) blank+++{-| Parse a succession of values from a stream of bytes, ending when the parser fails.or+ the bytes run out.++>>> S.print $ AS.parsed (A.scientific <* A.many' A.space) $ "12.3 4.56 78." >> "9 18.282"+12.3+4.56+78.9+18.282++-}+parsed+ :: Monad m+ => A.Parser a -- ^ Attoparsec parser+ -> ByteString m r -- ^ Raw input+ -> Stream (Of a) m (Either (Message, ByteString m r) r)+parsed parser = go+ where+ go p0 = do+ x <- lift (nextChunk p0)+ case x of+ Left r -> Return (Right r)+ Right (bs,p1) -> step (chunk bs >>) (A.parse parser bs) p1+ step diffP res p0 = case res of+ A.Fail _ c m -> Return (Left ((c,m), diffP p0))+ A.Done bs b -> Step (b :> go (chunk bs >> p0))+ A.Partial k -> do+ x <- lift (nextChunk p0)+ case x of+ Left e -> step diffP (k mempty) (return e)+ Right (a,p1) -> step (diffP . (chunk a >>)) (k a) p1+{-# INLINABLE parsed #-}++-- | Run a parser and return its result, using @StateT (ByteString m x)@ in the style+-- of pipes parse+-- atto :: Monad m => A.Parser a -> StateT (ByteString m x) m (Result a)+-- atto p = StateT (parse p)++-- atto_ :: Monad m => A.Parser a -> ExceptT ([String], String) (StateT (ByteString m x) m) a+-- atto_ p = ExceptT $ StateT loop where+-- loop s = case s of+-- Chunk x xs -> go (A.parse p x) xs+-- Empty r -> go (A.parse p B.empty) (Empty r)+-- Go m -> m >>= loop+--+-- go (T.Fail x stk msg) ys = return $ (Left (stk, msg), Chunk x ys)+-- go (T.Done x r) ys = return $ (Right r, Chunk x ys)+-- go (T.Partial k) (Chunk y ys) = go (k y) ys+-- go (T.Partial k) (Go m) = m >>= go (T.Partial k)+-- go (T.Partial k) blank = go (k B.empty) blank++
+ Data/ByteString/Streaming/HTTP.hs view
@@ -0,0 +1,132 @@+-- | This module replicates `pipes-http` as closely as will type-check.+-- +-- Here is an example GET request that streams the response body to standard+-- output:+--+-- > import qualified Data.ByteString.Streaming as S+-- > import Data.ByteString.Streaming.HTTP+-- >+-- > main = do+-- > req <- parseUrl "https://www.example.com"+-- > m <- newManager tlsManagerSettings +-- > withHTTP req m $ \resp -> S.stdout (responseBody resp) +-- > +--+-- Here is an example POST request that also streams the request body from+-- standard input:+--+-- > {-#LANGUAGE OverloadedStrings #-}+-- > import qualified Data.ByteString.Streaming as S+-- > import Data.ByteString.Streaming.HTTP+-- > +-- > main = do+-- > req <- parseUrl "https://www.example.com"+-- > let req' = req+-- > { method = "POST"+-- > , requestBody = stream S.stdin+-- > }+-- > m <- newManager tlsManagerSettings+-- > withHTTP req' m $ \resp -> S.stdout (responseBody resp)+--+-- For non-streaming request bodies, study the 'RequestBody' type, which also+-- accepts strict \/ lazy bytestrings or builders.+++module Data.ByteString.Streaming.HTTP (+ -- * http-client+ -- $httpclient+ module Network.HTTP.Client+ , module Network.HTTP.Client.TLS++ -- * Streaming Interface+ , withHTTP+ , streamN+ , stream++ ) where++import Control.Monad (unless)+import qualified Data.ByteString as B+import Data.Int (Int64)+import Data.IORef (newIORef, readIORef, writeIORef)+import Network.HTTP.Client+import Network.HTTP.Client.TLS+import Data.ByteString.Streaming+import Data.ByteString.Streaming.Internal+import Control.Monad.Trans++{- $httpclient+ This module is a thin @streaming-bytestring@ wrapper around the @http-client@ and+ @http-client-tls@ libraries.++ Read the documentation in the "Network.HTTP.Client" module of the+ @http-client@ library to learn about how to:++ * manage connections using connection pooling,++ * use more advanced request\/response features,++ * handle exceptions, and:+ + * manage cookies.++ @http-client-tls@ provides support for TLS connections (i.e. HTTPS).+-}++-- | Send an HTTP 'Request' and wait for an HTTP 'Response'+withHTTP+ :: Request+ -- ^+ -> Manager+ -- ^+ -> (Response (ByteString IO ()) -> IO a)+ -- ^ Handler for response+ -> IO a+withHTTP r m k = withResponse r m k'+ where+ k' resp = do+ let p = (from . brRead . responseBody) resp+ k (resp { responseBody = p})+{-# INLINABLE withHTTP #-}++-- | Create a 'RequestBody' from a content length and an effectful 'ByteString'+streamN :: Int64 -> ByteString IO () -> RequestBody+streamN n p = RequestBodyStream n (to p)+{-# INLINABLE streamN #-}++{-| Create a 'RequestBody' from an effectful 'ByteString'++ 'stream' is more flexible than 'streamN', but requires the server to support+ chunked transfer encoding.+-}+stream :: ByteString IO () -> RequestBody+stream p = RequestBodyStreamChunked (to p)+{-# INLINABLE stream #-}++to :: ByteString IO () -> (IO B.ByteString -> IO ()) -> IO ()+to p0 k = do+ ioref <- newIORef p0+ let readAction :: IO B.ByteString+ readAction = do+ p <- readIORef ioref+ case p of+ Empty () -> do+ writeIORef ioref (return ())+ return B.empty+ Go m -> do + p' <- m+ writeIORef ioref p'+ readAction+ Chunk bs p' -> do+ writeIORef ioref p'+ return bs+ k readAction ++from :: IO B.ByteString -> ByteString IO ()+from io = go+ where+ go = do+ bs <- lift io+ unless (B.null bs) $ do+ chunk bs+ go
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2015, Michael Thompson, 2014 Gabriel Gonzalez, 2014 Renzo Carbonara++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of michaelt nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ Streaming/Pipes.hs view
@@ -0,0 +1,368 @@+{-| "Pipes.Group.Tutorial" is the correct introduction to the use of this module,+ which is mostly just an optimized @Pipes.Group@, replacing @FreeT@ with @Stream@. + (See the introductory documentation for this package. The @pipes-group@ tutorial + is framed as a hunt for a genuinely streaming+ @threeGroups@. The formulation it opts for in the end would + be expressed here thus:++> import Pipes+> import Streaming.Pipes +> import qualified Pipes.Prelude as P+>+> threeGroups :: (Monad m, Eq a) => Producer a m () -> Producer a m ()+> threeGroups = concats . takes 3 . groups++ The only difference is that this simple module omits the detour via lenses.+ The program splits the initial producer into a connected stream of+ producers containing "equal" values; it takes three of those; and then+ erases the effects of splitting. So for example++>>> runEffect $ threeGroups (each "aabccoooooo") >-> P.print+'a'+'a'+'b'+'c'+'c'++ For the rest, only part of the tutorial that would need revision is + the bit at the end about writing explicit @FreeT@ programs. + Its examples use pattern matching, but the constructors of the + @Stream@ type are necessarily hidden, so one would have replaced + by the various inspection combinators provided by the @streaming@ library.+ +-}++{-#LANGUAGE RankNTypes, BangPatterns #-}++++module Streaming.Pipes (+ -- * @Streaming@ \/ @Pipes@ interoperation+ produce,+ stream,+ + -- * Transforming a connected stream of 'Producer's+ takes,+ takes',+ maps,+ + -- * Streaming division of a 'Producer' into two+ span,+ splitAt,+ group,+ groupBy,+ + -- * Splitting a 'Producer' into a connected stream of 'Producer's+ groupsBy,+ groupsBy',+ groups,+ + -- * Rejoining a connected stream of 'Producer's+ concats, + intercalates,+ + -- * Folding over the separate layers of a connected stream of 'Producer's+ folds,+ foldsM,+ + ) where++import Pipes+import Streaming hiding (concats)+import qualified Streaming.Internal as SI+import qualified Pipes.Internal as PI+import qualified Pipes.Prelude as P+import qualified Pipes as P+++import qualified Streaming.Prelude as S+import Control.Monad (liftM)+import Prelude hiding (span, splitAt)++-- | Construct an ordinary pipes 'Producer' from a 'Stream' of elements+produce :: Monad m => Stream (Of a) m r -> Producer' a m r+produce = loop where+ loop stream = case stream of -- this should be rewritten without constructors+ SI.Return r -> PI.Pure r+ SI.Delay m -> PI.M (liftM loop m)+ SI.Step (a:>rest) -> PI.Respond a (\_ -> loop rest)+{-# INLINABLE produce #-}++-- | Construct a 'Stream' of elements from a @pipes@ 'Producer'+stream :: Monad m => Producer a m r -> Stream (Of a) m r+stream = loop where+ loop stream = case stream of+ PI.Pure r -> SI.Return r + PI.M m -> SI.Delay (liftM loop m)+ PI.Respond a f -> SI.Step (a :> loop (f ()))+ PI.Request x g -> PI.closed x+{-# INLINABLE stream #-}++{-| 'span' splits a 'Producer' into two 'Producer's; the outer 'Producer' + is the longest consecutive group of elements that satisfy the predicate.+ Its inverse is 'Control.Monad.join'+-}+span :: Monad m => (a -> Bool) -> Producer a m r -> Producer a m (Producer a m r)+span predicate = loop where+ loop p = do+ e <- lift (next p)+ case e of+ Left r -> return (return r)+ Right (a, p') ->+ if predicate a+ then yield a >> loop p'+ else return (yield a >> p')+{-# INLINABLE span #-}++{-| 'splitAt' divides a 'Producer' into two 'Producer's + after a fixed number of elements. Its inverse is 'Control.Monad.join'++-}+splitAt+ :: Monad m+ => Int -> Producer a m r -> Producer a m (Producer a m r)+splitAt = loop where + loop n p | n <= 0 = return p+ loop n p = do+ e <- lift (next p)+ case e of+ Left r -> return (return r)+ Right (a, p') -> yield a >> loop (n - 1) p'+{-# INLINABLE splitAt #-}++{-| 'groupBy' splits a 'Producer' into two 'Producer's; the second+ producer begins where we meet an element that is different+ according to the equality predicate. Its inverse is 'Control.Monad.join'+-}+groupBy+ :: Monad m+ => (a -> a -> Bool) -> Producer a m r -> Producer a m (Producer a m r)+groupBy equals = loop where+ loop p = do+ x <- lift (next p)+ case x of+ Left r -> return (return r)+ Right (a, p') -> span (equals a) (yield a >> p') +{-# INLINABLE groupBy #-}++-- | Like 'groupBy', where the equality predicate is ('==')+group+ :: (Monad m, Eq a) => Producer a m r -> Producer a m (Producer a m r)+group = groupBy (==)+{-# INLINABLE group #-}+++groupsBy+ :: Monad m+ => (a -> a -> Bool)+ -> Producer a m r -> Stream (Producer a m) m r +groupsBy equals = loop where+ loop p = SI.Delay $ do+ e <- next p+ return $ case e of+ Left r -> SI.Return r+ Right (a, p') -> SI.Step (fmap loop (yield a >> span (equals a) p'))+{-# INLINABLE groupsBy #-}++{-| `groupsBy'` splits a 'Producer' into a 'Stream' of 'Producer's grouped using+ the given equality predicate++ This differs from `groupsBy` by comparing successive elements for equality+ instead of comparing each element to the first member of the group++>>> import Pipes (yield, each)+>>> import Pipes.Prelude (toList)+>>> let cmp c1 c2 = succ c1 == c2+>>> (toList . intercalates (yield '|') . groupsBy' cmp) (each "12233345")+"12|23|3|345"+>>> (toList . intercalates (yield '|') . groupsBy cmp) (each "12233345")+"122|3|3|34|5"+-}+groupsBy'+ :: Monad m+ => (a -> a -> Bool) -> Producer a m r -> Stream (Producer a m) m r+groupsBy' equals = loop where+ loop p = SI.Delay $ do+ e <- next p+ return $ case e of+ Left r -> SI.Return r+ Right (a, p') -> SI.Step (fmap loop (loop0 (yield a >> p')))+ loop0 p1 = do+ e <- lift (next p1)+ case e of+ Left r -> return (return r)+ Right (a2, p2) -> do+ yield a2+ let loop1 a p = do+ e' <- lift (next p)+ case e' of+ Left r -> return (return r)+ Right (a', p') ->+ if equals a a'+ then do+ yield a'+ loop1 a' p'+ else return (yield a' >> p')+ loop1 a2 p2+{-# INLINABLE groupsBy' #-}++groups:: (Monad m, Eq a)+ => Producer a m r -> Stream (Producer a m) m r+groups = groupsBy (==)++chunksOf+ :: Monad m => Int -> Producer a m r -> Stream (Producer a m) m r+chunksOf n = loop where+ loop p = SI.Delay $ do+ e <- next p+ return $ case e of+ Left r -> SI.Return r+ Right (a, p') -> SI.Step (fmap loop (splitAt n (yield a >> p')))+{-# INLINABLE chunksOf #-}++-- | Join a stream of 'Producer's into a single 'Producer'+concats :: Monad m => Stream (Producer a m) m r -> Producer a m r+concats = loop where+ loop stream = case stream of+ SI.Return r -> return r+ SI.Delay m -> PI.M $ liftM loop m+ SI.Step p -> do + rest <- p+ loop rest +{-# INLINABLE concats #-}++-- {-| Join a 'FreeT'-delimited stream of 'Producer's into a single 'Producer' by+-- intercalating a 'Producer' in between them+-- -}+-- intercalates+-- :: Monad m => Producer a m () -> Stream (Producer a m) m r -> Producer a m r+-- intercalates sep = loop where+-- loop stream = case stream of+--+-- x <- lift (runFreeT f)+-- case x of+-- Pure r -> return r+-- Free p -> do+-- f' <- p+-- go1 f'+-- go1 f = do+-- x <- lift (runFreeT f)+-- case x of+-- Pure r -> return r+-- Free p -> do+-- sep+-- f' <- p+-- go1 f'+-- {-# INLINABLE intercalates #-}++{- $folds+ These folds are designed to be compatible with the @foldl@ library. See+ the 'Control.Foldl.purely' and 'Control.Foldl.impurely' functions from that+ library for more details.++ For example, to count the number of 'Producer' layers in a 'Stream', you can+ write:++> import Control.Applicative (pure)+> import qualified Control.Foldl as L+> import Pipes.Group+> import qualified Pipes.Prelude as P+>+> count :: Monad m => Stream (Producer a m) m () -> m Int+> count = P.sum . L.purely folds (pure 1)+-}++{-| Fold each 'Producer' in a producer 'Stream'++> purely folds+> :: Monad m => Fold a b -> Stream (Producer a m) m r -> Producer b m r+-}++folds+ :: Monad m+ => (x -> a -> x)+ -- ^ Step function+ -> x+ -- ^ Initial accumulator+ -> (x -> b)+ -- ^ Extraction function+ -> Stream (Producer a m) m r+ -- ^+ -> Producer b m r+folds step begin done = loop where+ loop stream = case stream of + SI.Return r -> return r+ SI.Delay m -> PI.M $ liftM loop m+ SI.Step p -> do+ (stream', b) <- lift (fold p begin)+ yield b+ loop stream'+ fold p x = do+ y <- next p+ case y of+ Left f -> return (f, done x)+ Right (a, p') -> fold p' $! step x a+{-# INLINABLE folds #-}++++{-| Fold each 'Producer' in a 'Producer' stream, monadically++> impurely foldsM+> :: Monad m => FoldM a b -> Stream (Producer a m) m r -> Producer b m r+-}+foldsM+ :: Monad m+ => (x -> a -> m x)+ -- ^ Step function+ -> m x+ -- ^ Initial accumulator+ -> (x -> m b)+ -- ^ Extraction function+ -> Stream (Producer a m) m r+ -- ^+ -> Producer b m r+foldsM step begin done = loop where+ loop stream = case stream of + SI.Return r -> return r+ SI.Delay m -> PI.M (liftM loop m)+ SI.Step p -> do+ (f', b) <- lift $ begin >>= foldM p + yield b+ loop f'++ foldM p x = do+ y <- next p+ case y of+ Left f -> do+ b <- done x+ return (f, b)+ Right (a, p') -> do+ x' <- step x a+ foldM p' $! x'+{-# INLINABLE foldsM #-}++{-| @(takes' n)@ only keeps the first @n@ 'Producer's of a linked 'Stream' of 'Producers'++ Unlike 'takes', 'takes'' is not functor-general - it is aware that a 'Producer'+ can be /drained/, as functors cannot generally be. Here, then, we drain + the unused 'Producer's in order to preserve the return value. + This makes it a suitable argument for 'maps'.+-}+takes' :: Monad m => Int -> Stream (Producer a m) m r -> Stream (Producer a m) m r+takes' = loop where+ + loop !n stream | n <= 0 = drain_loop stream+ loop n stream = case stream of+ SI.Return r -> SI.Return r+ SI.Delay m -> SI.Delay (liftM (loop n) m)+ SI.Step p -> SI.Step (fmap (loop (n - 1)) p)++ drain_loop stream = case stream of+ SI.Return r -> SI.Return r+ SI.Delay m -> SI.Delay (liftM drain_loop m)+ SI.Step p -> SI.Delay $ do + stream' <- runEffect (P.for p P.discard)+ return $ drain_loop stream'+{-# INLINABLE takes' #-}+
+ streaming-utils.cabal view
@@ -0,0 +1,36 @@+name: streaming-utils+version: 0.1.0.0+synopsis: utilities for http-client, attoparsec, pipes etc with streaming and streaming-bytestring+description: Experimental http-client, attoparsec and pipes utilities streaming and streaming-bytestring++license: BSD3+license-file: LICENSE+author: michaelt+maintainer: what_is_it_to_do_anything@yahoo.com+-- copyright: +category: Data+build-type: Simple+-- extra-source-files: +cabal-version: >=1.10++library+ exposed-modules: Data.Attoparsec.ByteString.Streaming,+ Data.ByteString.Streaming.HTTP,+ Streaming.Pipes+ -- other-modules: + other-extensions: CPP, Trustworthy+ + build-depends: base >=4.6 && <4.9, + transformers >=0.4 && <0.5, + mtl >=2.2 && <2.3,+ attoparsec,+ streaming > 0.1.0.15 && < 0.1.1,+ streaming-bytestring > 0.1.0.5 && < 0.1.1,+ bytestring, + pipes >= 4.0 && < 4.2,+ http-client >=0.2 && <0.5, + http-client-tls <0.3+ + + -- hs-source-dirs: + default-language: Haskell2010