pipes-attoparsec-streaming (empty) → 0.1.0.0

raw patch · 5 files changed

+435/−0 lines, 5 filesdep +attoparsecdep +basedep +bytestringsetup-changed

Dependencies added: attoparsec, base, bytestring, pipes-attoparsec-streaming, pipes-core, transformers

Files

• Control/Pipe/Attoparsec/Stream.hs +193/−0
• LICENSE +30/−0
• Setup.hs +2/−0
• examples/MimeParser.hs +167/−0
• pipes-attoparsec-streaming.cabal +43/−0

@@ -0,0 +1,193 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE DeriveDataTypeable #-}+------------------------------------------------------------------------------+-- | Module: Control.Pipe.Attoparsec.Stream+-- Copyright: Martin Grabmueller+-- License: BSD3+--+-- Maintainer: martin@grabmueller.de+-- Stability: provisional+-- Portability: GHC and Linux only+--+-- This module exports the single function parse, which can be used to+-- run an Attoparsec parser in a streaming fashion, which means that+-- the parser is not only run incrementally across the input (which+-- can be done with plain Attoparsec or packages like+-- pipes-attoparsec), but that the parse results are delivered+-- incrementally.  This package can be seen as a kind of dual to+-- pipes-attoparsec: the latter runs parser incrementally over their+-- input, whereas the former incrementally delivers output.+--+-- The following example (to be found with comments in+-- examples/MimeParser.hs in the source tarball) shows an example for+-- parsing e-mail messages incrementally.+--+-- > import Control.Pipe.Attoparsec.Stream+-- > +-- > import Data.Word+-- > import Control.Pipe+-- > import Control.Pipe.Combinators as Combinators+-- > import Data.Attoparsec(Parser)+-- > import qualified Data.Attoparsec as A+-- > import qualified Data.Attoparsec.Char8 as A8+-- > import Data.ByteString(ByteString)+-- > import qualified Data.ByteString as B+-- > import Control.Applicative+-- > import Control.Monad.Trans.Class(lift)+-- >+-- > isToken :: Word8 -> Bool+-- > isToken w = w <= 127 && A.notInClass "\0-\31()<>@,;:\\\"/[]?={} \t" w+-- > +-- > takeSpaces :: Parser ByteString+-- > takeSpaces = B.cons <$> A.satisfy A8.isHorizontalSpace <*> A.takeWhile A8.isHorizontalSpace+-- > +-- > endOfLine :: Parser ByteString+-- > endOfLine = (A.word8 10 >> return "\n") <|> A.string "\r\n"+-- > +-- > takeAtMost :: Int -> Parser ByteString+-- > takeAtMost n = A.scan n (\ n' c -> if n' == 0 then Nothing else (Just (n' - 1)))+-- > +-- > data Event+-- >   = Header ByteString ByteString [(ByteString, ByteString, ByteString)]+-- >   | EndOfHeader ByteString+-- >   | BodyChunk ByteString+-- >   | EndOfInput+-- >   deriving (Show)+-- > +-- > messageHeader :: Parser (PartialResult Event)+-- > messageHeader = do+-- >   header <- B.cons <$> A.satisfy isToken <*> A.takeWhile isToken+-- >   delim <- B.snoc <$> A.takeWhile A8.isHorizontalSpace <*> A8.char8 ':'+-- >   body <-  (,,) <$> A.takeWhile A8.isHorizontalSpace <*>+-- >                     A.takeTill A8.isEndOfLine <*> endOfLine+-- >   bodies <- many $ (,,) <$> takeSpaces <*> A.takeTill A8.isEndOfLine <*> endOfLine+-- >   return (PartialResult (Just (Header header delim (body:bodies))) (Just messageHeader))+-- >  <|>+-- >   return (PartialResult Nothing (Just headerEnd))+-- > +-- > headerEnd :: Parser (PartialResult Event)+-- > headerEnd = do+-- >   s <- endOfLine+-- >   return $! PartialResult (Just (EndOfHeader s)) (Just bodyChunk)+-- > +-- > bodyChunk :: Parser (PartialResult Event)+-- > bodyChunk = do+-- >   s <- takeAtMost 10+-- >   if B.null s+-- >     then return $ PartialResult (Just EndOfInput) Nothing+-- >     else return $ PartialResult (Just (BodyChunk s)) (Just bodyChunk)+-- > +-- > msg :: ByteString+-- > msg =+-- >   "Received  : foo\r\n  cont'd\r\nDate: now\r\n\r\nblabla\r\nsecond line - longer"+-- > +-- > dump :: Pipe Event Void IO ()+-- > dump = go+-- >  where+-- >    go = do+-- >      e <- await+-- >      lift $ print e+-- >      go+-- > +-- > main :: IO ()+-- > main = do+-- >   runPipe $ fromList [msg] >+> parse messageHeader >+> dump+--+------------------------------------------------------------------------------+module Control.Pipe.Attoparsec.Stream+  (+    -- * Types+    StreamException(..),+    PartialResult(..),+    -- * Parse function+    parse+  ) where++import Control.Exception+import Control.Pipe+import Control.Pipe.Combinators as Combinators+import Control.Pipe.Exception as E+import Data.Attoparsec(Parser, Result)+import qualified Data.Attoparsec as A+import Data.ByteString(ByteString)+import qualified Data.ByteString as B+import Data.Typeable++-- | This is the type of exceptions that may be thrown by the+-- streaming parser.+--+-- Currently, only one exception i defined: it reports Attoparsec+-- parse errors.  The values carried by the exceptions are the parse+-- context and the error message, just as in the @Fail@ case of+-- Attoparsec's @Result@ type.+--+data StreamException+  = ParseException [String] String -- ^ Parsing failed.+  deriving (Show, Typeable)++instance Exception StreamException++-- | Parsers which are to be used with the streaming parser Pipe must+-- return values of the following type:+--+-- * The first parameter of type @Maybe a@ is the parser result. When+-- it is a @Just@ value, the carried value is delivered to the next+-- Pipe in the Pipeline.  When it is @Nothing@, no value is yield.+--+-- * When the second parameter is a @Just p@, the parser @p@ is used+-- to continue the parsing process. When it is @Nothing@, the parsing+-- process ends and the Pipeline terminates.+--+data PartialResult a = PartialResult (Maybe a) (Maybe (Parser (PartialResult a)))+++-- | This function converts an Attoparsec parser over bytestrings+-- which returns a @PartialResult a@ into a Pipe that consumes a+-- stream of bytestrings and delivers a stream of @a@ values.+--+parse :: Monad m => Parser (PartialResult a) -> Pipe ByteString a m ()+parse parser = do+  chunk <- await+  go (A.parse parser chunk)+ where+   go :: Monad m => (Result (PartialResult a)) -> Pipe ByteString a m ()+   go cont =+     case cont of+       -- Convert a parse error into an exception.+       A.Fail _ ctxt err -> +         E.throw (ParseException ctxt err)++       -- Partial parse: check whether more input is available, and+       -- feed it to the continuation.  Note that the following is not+       -- sufficient:+       --+       -- > do chunk <- await+       -- >    go (c chunk)+       --+       -- The reason is that the await will terminate the pipeline+       -- when no more input is available, so the parse will not see+       -- the end-of-input.  Instead, we have to feed an empty+       -- bytestring to the continuation, so that Attoparsec can+       -- finish the parsing.+       --+       A.Partial c -> do+         chunk <- tryAwait+         case chunk of+           Nothing -> go (c B.empty)+           Just chunk' -> go (c chunk')+       +       -- We are finished with the current parser. Yield the parsed+       -- value (if there is any) and run the continuation parser, if+       -- there is one.+       --+       A.Done rest (PartialResult mbResult mbNext) -> do+         case mbResult of+           Nothing -> return ()+           Just a -> yield a+         case mbNext of+           Nothing -> return ()+           Just p -> go (A.parse p rest)++

@@ -0,0 +1,30 @@+Copyright (c) 2012, Martin Grabmueller++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 Martin Grabmueller 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.

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

@@ -0,0 +1,167 @@+{-# LANGUAGE OverloadedStrings #-}+------------------------------------------------------------------------------+-- | Module: Main+-- Copyright: Martin Grabmueller+-- License: BSD3+--+-- Maintainer: martin@grabmueller.de+-- Stability: provisional+-- Portability: GHC and Linux only+--+-- This file contains an example on using the+-- pipes-attoparsec-streaming package.+--+-- Part of this code was adapted from Bryan O'Sullivan's HTTP parser+-- example on his blog.+--+-- Currently, this parser only recognizes headers (like in e-mails),+-- the end-of-header empty line and the parser chunks the remainder of+-- the input into suitable pieces.+------------------------------------------------------------------------------+module Main(main) where++import Control.Pipe.Attoparsec.Stream++import Data.Word+import Control.Pipe+import Control.Pipe.Combinators as Combinators+import Data.Attoparsec(Parser)+import qualified Data.Attoparsec as A+import qualified Data.Attoparsec.Char8 as A8+import Data.ByteString(ByteString)+import qualified Data.ByteString as B+import Control.Applicative+import Control.Monad.Trans.Class(lift)+++------------------------------------------------------------------------------+-- First, we start with some useful utility parsers.+------------------------------------------------------------------------------++-- Predicate for testing whether a byte may appear in a MIME token.+--+isToken :: Word8 -> Bool+isToken w = w <= 127 && A.notInClass "\0-\31()<>@,;:\\\"/[]?={} \t" w++-- Skip horizontal space.  At least one space must be+-- available. Return the skipped whitespace.+--+takeSpaces :: Parser ByteString+takeSpaces = B.cons <$> A.satisfy A8.isHorizontalSpace <*> A.takeWhile A8.isHorizontalSpace++-- Recognize and return an end-of-line sequence, which must be either+-- a single newline or a carriage-newline combination.+--+endOfLine :: Parser ByteString+endOfLine = (A.word8 10 >> return "\n") <|> A.string "\r\n"++-- Return @n@ bytes, or less if the end of input is reached.+takeAtMost :: Int -> Parser ByteString+takeAtMost n = A.scan n (\ n' _ -> if n' == 0 then Nothing else (Just (n' - 1)))+++------------------------------------------------------------------------------+-- Events.  These are the parse results that are streamed.+------------------------------------------------------------------------------++-- | The generated events. +--+-- Each event contains exactly the bytes which were parsed, so that+-- the original input can be reconstructed from the events.+--+data Event+  = Header ByteString ByteString [(ByteString, ByteString, ByteString)]+  | EndOfHeader ByteString+  | BodyChunk ByteString+  | EndOfInput+  deriving (Show)+++------------------------------------------------------------------------------+-- The parser.+------------------------------------------------------------------------------++-- | Parse a single message header, including continuation lines.+--+messageHeader :: Parser (PartialResult Event)+messageHeader = do+  -- Header name.+  header <- B.cons <$> A.satisfy isToken <*> A.takeWhile isToken+  +  -- Delimiter.+  delim <- B.snoc <$> A.takeWhile A8.isHorizontalSpace <*> A8.char8 ':'+  +  -- Parse the first line of the header value.+  body <-  (,,) <$> A.takeWhile A8.isHorizontalSpace <*>+                    A.takeTill A8.isEndOfLine <*> endOfLine+                    +  -- Parse the remaining lines of the header value. Each one must+  -- begin with at least one horizontal space.+  bodies <- many $ (,,) <$> takeSpaces <*> A.takeTill A8.isEndOfLine <*> endOfLine+  +  -- Return the parsed header and return ourself as the continuation,+  -- to parse more headers.+  return (PartialResult (Just (Header header delim (body:bodies))) (Just messageHeader))++ <|>++  -- When no header can be parsed, return no result and set the+  -- @headerEnd@ parser as the continuation.+  return (PartialResult Nothing (Just headerEnd))+++-- | Parse the end-of-header marker (an empty line).+--+headerEnd :: Parser (PartialResult Event)+headerEnd = do+  -- This one is easy.+  s <- endOfLine+  +  -- Continue with a body chunk.+  return $! PartialResult (Just (EndOfHeader s)) (Just bodyChunk)+++-- | Parse a chunk of message body data.+--+bodyChunk :: Parser (PartialResult Event)+bodyChunk = do+  -- Parse a small chunk.+  s <- takeAtMost 10+  if B.null s+    -- When end-of-input is reached, indicate that with the correct+    -- event and don't return a continuation.+    then return $ PartialResult (Just EndOfInput) Nothing+         +    -- Return the chunk and try again.+    else return $ PartialResult (Just (BodyChunk s)) (Just bodyChunk)+++------------------------------------------------------------------------------+-- Example invocation.+------------------------------------------------------------------------------++-- | Tiny example message.+--+msg :: ByteString+msg =+  "Received  : foo\r\n  cont'd\r\nDate: now\r\n\r\nblabla\r\nsecond line - longer"+++-- | Small helper for printing the parsed events.+--+dump :: Pipe Event Void IO ()+dump = go+ where+   go = do+     e <- await+     lift $ print e+     go+++-- | Main program. Run the parser on a small example and print the+-- streamed parsing results.+--+main :: IO ()+main = do+  runPipe $ fromList [msg] >+> parse messageHeader >+> dump+

@@ -0,0 +1,43 @@+name: pipes-attoparsec-streaming+version: 0.1.0.0+synopsis: Streaming parsing in the pipes-core framework with Attoparsec.+description: This module exports the single function parse, which can be used to+  run an Attoparsec parser in a streaming fashion, which means that+  the parser is not only run incrementally across the input (which+  can be done with plain Attoparsec or packages like+  pipes-attoparsec), but that the parse results are delivered+  incrementally.  This package can be seen as a kind of dual to+  pipes-attoparsec: the latter runs parser incrementally over their+  input, whereas the former incrementally delivers output.++license: BSD3+license-file: LICENSE+author: Martin Grabmueller+maintainer: martin@grabmueller.de+copyright: Copyrght (c) 2012 Martin Grabmueller+category: Control+build-type: Simple+cabal-version: >= 1.8++library+  exposed-modules:+    Control.Pipe.Attoparsec.Stream+  -- other-modules:+  build-depends:+    base == 4.5.*,+    pipes-core == 0.1.*,+    attoparsec == 0.10.*,+    transformers >= 0.3,+    bytestring >= 0.9++executable MimeParser+  main-is: MimeParser.hs+  hs-source-dirs: examples+  build-depends:+    base == 4.5.*,+    pipes-core == 0.1.*,+    attoparsec == 0.10.*,+    transformers >= 0.3,+    bytestring >= 0.9,+    pipes-attoparsec-streaming+