diff --git a/Data/Attoparsec/ByteString/Streaming.hs b/Data/Attoparsec/ByteString/Streaming.hs --- a/Data/Attoparsec/ByteString/Streaming.hs +++ b/Data/Attoparsec/ByteString/Streaming.hs @@ -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 - - diff --git a/Data/ByteString/Streaming/HTTP.hs b/Data/ByteString/Streaming/HTTP.hs --- a/Data/ByteString/Streaming/HTTP.hs +++ b/Data/ByteString/Streaming/HTTP.hs @@ -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))} + diff --git a/Streaming/Pipes.hs b/Streaming/Pipes.hs --- a/Streaming/Pipes.hs +++ b/Streaming/Pipes.hs @@ -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 diff --git a/Streaming/Zip.hs b/Streaming/Zip.hs new file mode 100644 --- /dev/null +++ b/Streaming/Zip.hs @@ -0,0 +1,217 @@ +-- | This module modifies material in Renzo Carbonara\'s 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 #-} + diff --git a/streaming-utils.cabal b/streaming-utils.cabal --- a/streaming-utils.cabal +++ b/streaming-utils.cabal @@ -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 and 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