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 @@ -1,9 +1,50 @@ +{- | Here is a simple use of 'parsed' and standard @Streaming@ segmentation devices to + parse a file in which groups of numbers are separated by blank lines. Such a problem + of \'nesting streams\' is described in the @conduit@ context in + + +> -- $ cat nums.txt +> -- 1 +> -- 2 +> -- 3 +> -- +> -- 4 +> -- 5 +> -- 6 +> -- +> -- 7 +> -- 8 + + We will sum the groups and stream the results to standard output: + +> import Streaming +> import qualified Streaming.Prelude as S +> import qualified Data.ByteString.Streaming.Char8 as Q +> import qualified Data.Attoparsec.ByteString.Char8 as A +> import qualified Data.Attoparsec.ByteString.Streaming as A +> import Data.Function ((&)) +> +> main = Q.getContents -- raw bytes +> & A.parsed lineParser -- stream of parsed `Maybe Int`s; blank lines are `Nothing` +> & void -- drop any unparsed nonsense at the end +> & S.split Nothing -- split on blank lines +> & S.maps S.concat -- keep `Just x` values in the sub-streams (cp. catMaybes) +> & S.mapped S.sum -- sum each substream +> & S.print -- stream results to stdout +> +> lineParser = Just <$> A.scientific <* A.endOfLine <|> Nothing <$ A.endOfLine + +> -- $ cat nums.txt | ./atto +> -- 6.0 +> -- 15.0 +> -- 15.0 + +-} module Data.Attoparsec.ByteString.Streaming (Message , parse , parsed , module Data.Attoparsec.ByteString - ) where @@ -24,13 +65,14 @@ {- | 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" +>>> :set -XOverloadedStrings -- the string literal below is a streaming bytestring +>>> (r,rest1) <- AS.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 +>>> (s,rest2) <- AS.parse (A.scientific <* A.many' A.space) rest1 >>> print s Left 4.56 ->>> (t,rest3) <- parse (A.scientific <* A.many' A.space) rest2 +>>> (t,rest3) <- AS.parse (A.scientific <* A.many' A.space) rest2 >>> print t Left 78.3 >>> Q.putStrLn rest3 @@ -51,10 +93,10 @@ 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. +{-| Apply a parser repeatedly to a stream of bytes, streaming the parsed values, but + 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" +>>> S.print $ AS.parsed (A.scientific <* A.many' A.space) $ "12.3 4.56 78.9" 12.3 4.56 78.9 diff --git a/Data/ByteString/Streaming/Aeson.hs b/Data/ByteString/Streaming/Aeson.hs --- a/Data/ByteString/Streaming/Aeson.hs +++ b/Data/ByteString/Streaming/Aeson.hs @@ -1,13 +1,63 @@ {-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE RankNTypes #-} --- | The @encode@, @decode@ and @decoded@ functions replicate --- the similar functions in Renzo Carbonara's `pipes-aeson`. --- @streamParse@ accepts parsers from the 'json-streams' library. --- Aeson must consume a whole top level json array or --- object before coming to any conclusion. The 'json-streams' --- parsers use aeson types, but will stream suitable elements --- as they arise. +{- | The @encode@, @decode@ and @decoded@ functions replicate + the similar functions in Renzo Carbonara's + . + Note that @aeson@ accumulates a whole top level json array or object before coming to any conclusion. + This is the only default that could cover all cases. + + The @streamParse@ function accepts parsers from the + library. + The 'json-streams' parsers use aeson types, but will stream suitable elements + as they arise. For this reason, of course, it cannot validate the entire json + entity before acting, but carries on validation as it moves along, reporting + failure when it comes. Though it is generally faster and accumulates less memory than + the usual aeson parsers, it is by no means a universal replacement for + aeson\'s behavior. It will certainly be sensible, for example, wherever + you are executing a left fold over the subordinate elements, e.g. gathering certain + statistics about them. + + Here we use a long top level array of objects from + + @json-streams@ benchmarking directory. Each object has a friends field with + an array of friends; we extract the name of each friend of each person recorded in the level array, and + enumerate them all: + +> {-#LANGUAGE OverloadedStrings #-} +> import Streaming +> import qualified Streaming.Prelude as S +> import Data.ByteString.Streaming.HTTP +> import Data.ByteString.Streaming.Aeson (streamParse) +> import Data.JsonStream.Parser (string, arrayOf, (.:), Parser) +> import Data.Text (Text) +> import Data.Function ((&)) +> main = do +> req <- parseUrl "https://raw.githubusercontent.com/ondrap/json-stream/master/benchmarks/json-data/buffer-builder.json" +> m <- newManager tlsManagerSettings +> withHTTP req m $ \resp -> do +> let names, friend_names :: Parser Text +> names = arrayOf ("name" .: string) +> friend_names = arrayOf ("friends" .: names) +> responseBody resp -- raw bytestream from http-client +> & streamParse friend_names -- find name fields in each sub-array of friends +> & void -- drop material after any bad parse +> & S.zip (S.each [1..]) -- number the friends' names +> & S.print -- successively print to stdout +> +> -- (1,"Joyce Jordan") +> -- (2,"Ophelia Rosales") +> -- (3,"Florine Stark") +> -- ... +> -- (287,"Hilda Craig") +> -- (288,"Leola Higgins") + + This program does not accumulate the whole byte stream, as an aeson parser + for a top-level json entity would. Rather it streams and enumerates + friends\' names as soon as they come. + +-} + module Data.ByteString.Streaming.Aeson ( DecodingError(..) 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 @@ -4,30 +4,30 @@ -- Here is an example GET request that streams the response body to standard -- output: -- --- > import qualified Data.ByteString.Streaming as S +-- > import qualified Data.ByteString.Streaming as Q -- > import Data.ByteString.Streaming.HTTP -- > -- > main = do -- > req <- parseUrl "https://www.example.com" -- > m <- newManager tlsManagerSettings --- > withHTTP req m $ \resp -> S.stdout (responseBody resp) +-- > withHTTP req m $ \resp -> Q.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 qualified Data.ByteString.Streaming as Q -- > import Data.ByteString.Streaming.HTTP -- > -- > main = do -- > req <- parseUrl "https://www.example.com" -- > let req' = req -- > { method = "POST" --- > , requestBody = stream S.stdin +-- > , requestBody = stream Q.stdin -- > } -- > m <- newManager tlsManagerSettings --- > withHTTP req' m $ \resp -> S.stdout (responseBody resp) +-- > withHTTP req' m $ \resp -> Q.stdout (responseBody resp) -- -- For non-streaming request bodies, study the 'RequestBody' type, which also -- accepts strict \/ lazy bytestrings or builders. diff --git a/streaming-utils.cabal b/streaming-utils.cabal --- a/streaming-utils.cabal +++ b/streaming-utils.cabal @@ -1,5 +1,5 @@ name: streaming-utils -version: 0.1.4.1 +version: 0.1.4.2 synopsis: http, attoparsec, pipes and conduit utilities for the streaming libraries description: Experimental http-client, aeson, attoparsec and pipes utilities for use with the and @@ -11,20 +11,18 @@ of a remote document thus: . > import Streaming - > import Streaming.Prelude (with, each, next, for) + > import Streaming.Prelude (with, each) > import qualified Streaming.Prelude as S > import Data.ByteString.Streaming.HTTP > import qualified Data.ByteString.Streaming.Char8 as Q > - > main = runResourceT - > $ Q.putStrLn - > $ Q.unlines - > $ interleaves numbers - > & Q.lines - > $ simpleHTTP "http://lpaste.net/raw/146542" + > main = runResourceT $ do + > let output = numbers <|> Q.lines (simpleHTTP "http://lpaste.net/raw/146542") + > Q.putStrLn $ Q.unlines output > - > numbers :: Monad m => Stream (Q.ByteString m) m r - > numbers = with (S.enumFrom (1::Int)) $ \n -> Q.pack (each (show n ++ ". ")) + > numbers :: Monad m => Stream (Q.ByteString m) m () + > numbers = with (each [1..]) $ \n -> Q.pack (each (show n ++ ". ")) + > -- ["1. ", "2. " ..] . The memory requirements of this @Prelude@-ish program will not be affected by the fact that, say, the third \'line\' is 10 terabytes long.