streaming-utils 0.1.4.3 → 0.1.4.4
raw patch · 5 files changed
+326/−69 lines, 5 filesdep +streaming-commonsdep ~attoparsecdep ~http-clientdep ~http-client-tls
Dependencies added: streaming-commons
Dependency ranges changed: attoparsec, http-client, http-client-tls, pipes, streaming, streaming-bytestring
Files
- Data/Attoparsec/ByteString/Streaming.hs +39/−41
- Data/ByteString/Streaming/HTTP.hs +41/−10
- Streaming/Pipes.hs +18/−7
- Streaming/Zip.hs +217/−0
- streaming-utils.cabal +11/−11
Data/Attoparsec/ByteString/Streaming.hs view
@@ -44,7 +44,7 @@ (Message , parse , parsed- , module Data.Attoparsec.ByteString+-- , module Data.Attoparsec.ByteString ) where @@ -56,7 +56,7 @@ parse, parseWith, parseTest) import Streaming hiding (concats, unfold)-import Streaming.Internal (Stream (..))+import Streaming.Internal (Stream (..)) import Data.ByteString.Streaming import Data.ByteString.Streaming.Internal import Data.Monoid @@ -81,18 +81,35 @@ 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 parser bs = do + (e,rest) <- apply parser bs+ return (either Right Left e, rest)+{-#INLINE parse #-} +apply :: Monad m+ => A.Parser a+ -> ByteString m x -> m (Either Message a, ByteString m x)+apply parser = begin where+ + begin p0 = case p0 of+ Go m -> m >>= begin+ Empty r -> step id (A.parse parser mempty) (return r)+ Chunk bs p1 -> if B.null bs -- attoparsec understands "" + then begin p1 -- as eof.+ else step (chunk bs >>) (A.parse parser bs) p1 + step diff res p0 = case res of+ T.Fail _ c m -> return (Left (c,m), diff p0)+ T.Done a b -> return (Right b, chunk a >> p0)+ T.Partial k -> do+ let clean p = case p of -- inspect for null chunks before+ Go m -> m >>= clean -- feeding attoparsec + Empty r -> step diff (k mempty) (return r)+ Chunk bs p1 | B.null bs -> clean p1+ | otherwise -> step (diff . (chunk bs >>)) (k bs) p1+ clean p0+{-#INLINABLE apply #-} + {-| Apply a parser repeatedly to a stream of bytes, streaming the parsed values, but ending when the parser fails.or the bytes run out. @@ -101,46 +118,27 @@ 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+parsed parser = begin 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+ begin p0 = case p0 of -- inspect for null chunks before+ Go m -> lift m >>= begin -- feeding attoparsec + Empty r -> Return (Right r)+ Chunk bs p1 | B.null bs -> begin p1+ | otherwise -> 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.Done bs a | B.null bs -> Step (a :> begin p0) + | otherwise -> Step (a :> begin (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+ Right (bs,p1) | B.null bs -> step diffP res p1+ | otherwise -> step (diffP . (chunk bs >>)) (k bs) 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
@@ -1,9 +1,11 @@ {-#LANGUAGE OverloadedStrings #-}--- | 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:+-- | This module replicates `pipes-http` as closely as will type-check, adding a+-- conduit-like @http@ in @ResourceT@ and a primitive @simpleHTTP@ that emits+-- a streaming bytestring rather than a lazy one. --+-- +-- Here is an example GET request that streams the response body to standard output:+-- -- > import qualified Data.ByteString.Streaming as Q -- > import Data.ByteString.Streaming.HTTP -- >@@ -29,18 +31,36 @@ -- > m <- newManager tlsManagerSettings -- > withHTTP req' m $ \resp -> Q.stdout (responseBody resp) --+-- Here is the GET request modified to use @http@ and write to a file. @runResourceT@+-- manages the file handle and the interaction.+--+-- > import qualified Data.ByteString.Streaming as Q+-- > import Data.ByteString.Streaming.HTTP+-- >+-- > main = do+-- > req <- parseUrl "https://www.example.com"+-- > m <- newManager tlsManagerSettings +-- > runResourceT $ do+-- > resp <- http request manager +-- > Q.writeFile "example.html" (responseBody resp) +--+-- +-- @simpleHTTP@ can be used in @ghci@ like so:+--+-- > ghci> runResourceT $ Q.stdout $ Q.take 137 $ simpleHTTP "http://lpaste.net/raw/13"+-- > -- Adaptation and extension of a parser for data definitions given in+-- > -- appendix of G. Huttons's paper - Monadic Parser Combinators.+-- > --+ -- 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+ withHTTP+ , http , streamN , stream @@ -48,6 +68,8 @@ , simpleHTTP -- * re-exports+ , module Network.HTTP.Client+ , module Network.HTTP.Client.TLS , ResourceT (..) , MonadResource (..) , runResourceT@@ -63,7 +85,7 @@ import Data.ByteString.Streaming.Internal import Control.Monad.Trans import Control.Monad.Trans.Resource-+import qualified Data.ByteString.Streaming.Char8 as Q {- $httpclient This module is a thin @streaming-bytestring@ wrapper around the @http-client@ and @http-client-tls@ libraries.@@ -189,4 +211,13 @@ responseClose ( from . liftIO . responseBody) ++http :: MonadResource m+ => Request+ -> Manager+ -> m (Response (ByteString m ()))+http req man = do+ (key, res) <- allocate (responseOpen req man) responseClose+ return res {responseBody = from (liftIO (responseBody res))} +
Streaming/Pipes.hs view
@@ -1,14 +1,22 @@ {-| "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@. + The module also includes optimized functions for interoperation:++> fromStream :: Monad m => Stream (Of a) m r -> Producer' a m r+> toStream :: Monad m => Producer a m r -> Stream (Of a) m r .- The only systematic difference is that this simple module omits lenses, which + It is not a drop in replacement for @Pipes.Group@. The only systematic difference+ is that this simple module omits lenses. It is hoped that this will may make elementary usage easier to grasp. The lenses exported the pipes packages- come into their own with the simple @StateT@ parsing procedure pipes promotes.- I hope to make a corresponding @Streaming.Pipes.Lens@ soon.+ only come into their own with the simple @StateT@ parsing procedure pipes promotes. + We are not attempting here to replicate this advanced procedure, but only to make + elementary forms of breaking and splitting possible in the simplest possible way. . The @pipes-group@ tutorial is framed as a hunt for a genuinely streaming- @threeGroups@. The formulation it opts for in the end would + @threeGroups@, which would collect the first three groups of matching items while+ never holding more than the present item in memory. + The formulation it opts for in the end would be expressed here thus: > import Pipes@@ -28,7 +36,11 @@ 'b' 'c' 'c'-+ + The new user might look at the examples of splitting, breaking and joining+ in @Streaming.Prelude@ keeping in mind that @Producer a m r@ is equivalent+ to @Stream (Of a) m r@.+ . For the rest, only part of the tutorial that would need revision is the bit at the end about writing explicit @FreeT@ programs. Here one does not proceed by pattern matching, but uses `inspect` in place of `runFreeT`@@ -48,8 +60,6 @@ {-#LANGUAGE RankNTypes, BangPatterns #-} -- module Streaming.Pipes ( -- * @Streaming@ \/ @Pipes@ interoperation fromStream,@@ -185,6 +195,7 @@ -} break :: Monad m => (a -> Bool) -> Producer a m r -> Producer a m (Producer a m r) break predicate = span (not . predicate)+{-#INLINE break #-} split :: (Eq a, Monad m) => a -> Producer a m r -> Stream (Producer a m) m r
+ Streaming/Zip.hs view
@@ -0,0 +1,217 @@+-- | This module modifies material in Renzo Carbonara\'s <http://hackage.haskell.org/package/pipes-zlib pipes-zlib> package. ++module Streaming.Zip (+ -- * Streams+ decompress+ , decompress'+ , compress+ , gunzip+ , gunzip'+ , gzip ++ -- * Compression levels+ , CompressionLevel+ , defaultCompression+ , noCompression+ , bestSpeed+ , bestCompression+ , compressionLevel++ -- * Window size+ -- $ccz-re-export+ , Z.defaultWindowBits+ , windowBits+ ) where + +import Data.Streaming.Zlib as Z+import Control.Exception (throwIO)+import Control.Monad (unless)+import qualified Data.ByteString as B+import Data.ByteString.Streaming +import Streaming+import qualified Data.ByteString.Streaming.Internal as I +import Data.ByteString.Streaming.Internal (ByteString (..)) ++++--------------------------------------------------------------------------------++-- | Decompress a streaming bytestring. 'Z.WindowBits' is from "Codec.Compression.Zlib" +--+-- @+-- 'decompress' 'defaultWindowBits' :: 'MonadIO' m => 'ByteString' m r -> 'ByteString' m r+-- @++decompress+ :: MonadIO m+ => Z.WindowBits+ -> ByteString m r -- ^ Compressed stream+ -> ByteString m r -- ^ Decompressed stream+decompress wbits p0 = do+ inf <- liftIO $ Z.initInflate wbits+ r <- for p0 $ \bs -> do+ popper <- liftIO (Z.feedInflate inf bs)+ fromPopper popper+ bs <- liftIO $ Z.finishInflate inf+ unless (B.null bs) (chunk bs)+ return r+{-# INLINABLE decompress #-}++-- | Decompress a zipped byte stream, returning any leftover input+-- that follows the compressed material.+decompress'+ :: MonadIO m+ => Z.WindowBits+ -> ByteString m r -- ^ Compressed byte stream+ -> ByteString m (Either (ByteString m r) r)+ -- ^ Decompressed byte stream, ending with either leftovers or a result+decompress' wbits p0 = go p0 =<< liftIO (Z.initInflate wbits)+ where+ flush inf = do+ bs <- liftIO $ Z.flushInflate inf+ unless (B.null bs) (chunk bs)+ go p inf = do+ res <- lift (nextChunk p)+ case res of+ Left r -> return $ Right r+ Right (bs, p') -> do+ fromPopper =<< liftIO (Z.feedInflate inf bs)+ flush inf+ leftover <- liftIO $ Z.getUnusedInflate inf+ if B.null leftover+ then go p' inf+ else return $ Left (chunk leftover >> p')+{-# INLINABLE decompress' #-}++-- | Compress a byte stream.+--+-- See the "Codec.Compression.Zlib" module for details about+-- 'Z.CompressionLevel' and 'Z.WindowBits'.+--+-- @+-- 'compress' :: 'MonadIO' m+-- => 'Z.CompressionLevel'+-- -> 'Z.WindowBits'+-- -> 'ByteString' m r+-- -> 'ByteString' m r+-- @+compress+ :: MonadIO m+ => CompressionLevel+ -> Z.WindowBits+ -> ByteString m r -- ^ Decompressed stream+ -> ByteString m r -- ^ Compressed stream+compress (CompressionLevel clevel) wbits p0 = do+ def <- liftIO $ Z.initDeflate clevel wbits+ let loop bs = case bs of + I.Chunk c rest -> do+ popper <- liftIO (Z.feedDeflate def c)+ fromPopper popper+ loop rest+ I.Go m -> I.Go (liftM loop m)+ I.Empty r -> return r+ r <- loop p0+ fromPopper $ Z.finishDeflate def+ return r+{-# INLINABLE compress #-}++--------------------------------------------------------------------------------++-- $ccz-re-export+--+-- The following are re-exported from "Codec.Compression.Zlib" for your+-- convenience.++--------------------------------------------------------------------------------+-- Compression Levels++-- | How hard should we try to compress?+newtype CompressionLevel = CompressionLevel Int+ deriving (Show, Read, Eq, Ord)++defaultCompression, noCompression, bestSpeed, bestCompression :: CompressionLevel+defaultCompression = CompressionLevel (-1)+noCompression = CompressionLevel 0+bestSpeed = CompressionLevel 1+bestCompression = CompressionLevel 9++-- | A specific compression level between 0 and 9.+compressionLevel :: Int -> CompressionLevel+compressionLevel n+ | n >= 0 && n <= 9 = CompressionLevel n+ | otherwise = error "CompressionLevel must be in the range 0..9"++windowBits :: Int -> WindowBits+windowBits = WindowBits++-- | Decompress a gzipped byte stream.+--+-- @+-- 'gunzip' :: 'MonadIO' m+-- => 'ByteString' m r+-- -> 'ByteString' m r+-- @+gunzip+ :: MonadIO m+ => ByteString m r -- ^ Compressed stream+ -> ByteString m r -- ^ Decompressed stream+gunzip = decompress gzWindowBits+{-# INLINABLE gunzip #-}++-- | Decompress a gzipped byte stream, returning any leftover input+-- that follows the compressed stream.+gunzip'+ :: MonadIO m+ => ByteString m r -- ^ Compressed byte stream+ -> ByteString m (Either (ByteString m r) r)+ -- ^ Decompressed bytes stream, returning either a 'ByteString' of + -- the leftover input or the return value from the input 'ByteString'.+gunzip' = decompress' gzWindowBits+{-# INLINE gunzip' #-}+++-- | Compress a byte stream in the gzip format.+--+-- @+-- 'gzip' :: 'MonadIO' m+-- => 'ZC.CompressionLevel'+-- -> 'ByteString' m r+-- -> 'ByteString' m r+-- @+gzip+ :: MonadIO m+ => CompressionLevel+ -> ByteString m r -- ^ Decompressed stream+ -> ByteString m r -- ^ Compressed stream+gzip clevel = compress clevel gzWindowBits+{-# INLINE gzip #-}++gzWindowBits :: Z.WindowBits+gzWindowBits = Z.WindowBits 31+++--------------------------------------------------------------------------------+-- Internal stuff+++for bs0 op = loop bs0 where+ loop bs = case bs of + I.Chunk c rest -> op c >> loop rest+ I.Go m -> I.Go (liftM loop m)+ I.Empty r -> return r+{-# INLINABLE for #-}++-- | Produce values from the given 'Z.Popper' until exhausted.+fromPopper :: MonadIO m+ => Z.Popper+ -> ByteString m ()+fromPopper pop = loop+ where+ loop = do + mbs <- liftIO pop+ case mbs of+ PRDone -> I.Empty ()+ PRError e -> I.Go (liftIO (throwIO e))+ PRNext bs -> I.Chunk bs loop+{-# INLINABLE fromPopper #-}+
streaming-utils.cabal view
@@ -1,7 +1,6 @@ name: streaming-utils-version: 0.1.4.3-synopsis: http, attoparsec, pipes and conduit utilities for the streaming libraries-description: Experimental http-client, aeson, attoparsec and pipes utilities for use with+version: 0.1.4.4+synopsis: Experimental http-client, aeson, attoparsec, zlib and pipes utilities for use with the <http://hackage.haskell.org/package/streaming streaming> and <http://hackage.haskell.org/package/streaming-bytestring streaming bytestring> libraries. They generally closely follow similarly named modules in the pipes @@ -58,7 +57,7 @@ , Data.ByteString.Streaming.Aeson , Streaming.Pipes , Streaming.Network.TCP- -- , Streaming.Pipes.Concurrent+ , Streaming.Zip -- other-modules: other-extensions: CPP, Trustworthy@@ -66,19 +65,20 @@ build-depends: base >=4.7 && <5.0, transformers >=0.4 && <0.5.3, mtl >=2.2 && <2.3,- attoparsec >=0.13.0.1,- streaming >= 0.1.4.0 && < 0.1.4.5,- streaming-bytestring >= 0.1.4.0 && < 0.1.4.5,+ attoparsec > 0.13.0.0 && < 0.13.2.0,+ streaming >= 0.1.4.0 && < 0.1.4.8,+ streaming-bytestring >= 0.1.4.0 && < 0.1.4.8, bytestring > 0.10.0 && < 0.11.0,- pipes >= 4.0 && < 4.2,+ pipes >= 4.0 && < 4.3, network-simple, network, -- pipes-concurrency >= 2.0 && < 2.1, - http-client >=0.2 && <0.5, - http-client-tls <0.3,+ http-client >=0.2 && <0.6, + http-client-tls, aeson > 0.8 && <0.11.3, json-stream > 0.4.0 && < 0.4.2,- resourcet > 1.0 && < 1.2+ resourcet > 1.0 && < 1.2,+ streaming-commons > 0.1.0 && < 0.1.17 -- hs-source-dirs: default-language: Haskell2010