diff --git a/ChangeLog.md b/ChangeLog.md new file mode 100644 --- /dev/null +++ b/ChangeLog.md @@ -0,0 +1,10 @@ +# Revision history for attoparsec-framer + +`attoparsec-framer` uses [PVP Versioning][1]. + + +## 0.1.0.0 -- 2023-03-01 + +* Initial version. + +[1]: https://pvp.haskell.org diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright (c) 2022, Tim Emiola + +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 Tim Emiola 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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,4 @@ +import Distribution.Simple + + +main = defaultMain diff --git a/attoparsec-framer.cabal b/attoparsec-framer.cabal new file mode 100644 --- /dev/null +++ b/attoparsec-framer.cabal @@ -0,0 +1,115 @@ +cabal-version: 3.0 +name: attoparsec-framer +version: 0.1.0.0 +synopsis: Use Attoparsec to parse framed protocol bytestreams +description: + A library that simplifies the use of + [Attoparsec](https://hackage.haskell.org/package/attoparsec) when processing + framed protocol bytestreams. + + As well as reading the haddocks, please take a look at a + [demo server](https://github.com/adetokunbo/attoparsec-framer/blob/main/toy/Server.hs) + and [client](https://github.com/adetokunbo/attoparsec-framer/blob/main/toy/Client.hs) + as working examples. + + +license: BSD-3-Clause +license-file: LICENSE +author: Tim Emiola +maintainer: adetokunbo@emio.la +category: Parsing, Attoparsec, Network API +homepage: https://github.com/adetokunbo/attoparsec-framer#readme +bug-reports: https://github.com/adetokunbo/attoparsec-framer/issues +build-type: Simple +extra-source-files: ChangeLog.md + +source-repository head + type: git + location: https://github.com/adetokunbo/attoparsec-framer.git + +library + exposed-modules: + Data.Attoparsec.Framer + Data.Attoparsec.Framer.Testing + + hs-source-dirs: src + build-depends: + , attoparsec >=0.14.4 && <0.15 + , base >=4.10 && <5 + , bytestring >=0.10.8.2 && <0.12.0.0 + , exceptions >= 0.10.4 && < 0.11 + , text >=1.2.3 && <2.2 + + default-language: Haskell2010 + ghc-options: + -Wall -Wincomplete-uni-patterns -Wpartial-fields -fwarn-tabs + +-- toy-server serves a simple framed protocol +executable toy-server + main-is: Server.hs + other-modules: + Attoparsec.ToyFrame + build-depends: + , attoparsec >=0.14.4 && <0.15 + , attoparsec-binary >=0.2 && <0.3 + , attoparsec-framer + , exceptions + , base >=4.10 && <5 + , bytestring >=0.10.8.2 && <0.12.0.0 + , network >=3.1.0 && <4 + , network-run >=0.2.2 && <0.3 + , text >=1.2.3 && <2.2 + , QuickCheck + + hs-source-dirs: toy + default-language: Haskell2010 + ghc-options: -threaded -rtsopts -with-rtsopts=-N + ghc-options: + -Wall -Wincomplete-uni-patterns -Wpartial-fields -fwarn-tabs + ghc-options: -fprof-auto + +-- toy-client accesses a simple framed protocol +executable toy-client + main-is: Client.hs + other-modules: + Attoparsec.ToyFrame + build-depends: + , attoparsec >=0.14.4 && <0.15 + , attoparsec-binary >=0.2 && <0.3 + , attoparsec-framer + , base >=4.10 && <5 + , bytestring >=0.10.8.2 && <0.12.0.0 + , network >=3.1.0 && <4 + , network-run >=0.2.2 && <0.3 + , text >=1.2.3 && <2.2 + , QuickCheck + + hs-source-dirs: toy + default-language: Haskell2010 + ghc-options: -threaded -rtsopts -with-rtsopts=-N + ghc-options: + -Wall -Wincomplete-uni-patterns -Wpartial-fields -fwarn-tabs + ghc-options: -fprof-auto + +test-suite unittests + type: exitcode-stdio-1.0 + main-is: Spec.hs + other-modules: + Attoparsec.ToyFrame + Attoparsec.ToyFrameSpec + + hs-source-dirs: test toy + build-depends: + , attoparsec + , attoparsec-binary + , attoparsec-framer + , exceptions + , base + , bytestring + , hspec + , QuickCheck + , text + + default-language: Haskell2010 + ghc-options: + -Wall -Wincomplete-uni-patterns -Wpartial-fields -fwarn-tabs diff --git a/src/Data/Attoparsec/Framer.hs b/src/Data/Attoparsec/Framer.hs new file mode 100644 --- /dev/null +++ b/src/Data/Attoparsec/Framer.hs @@ -0,0 +1,296 @@ +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE OverloadedStrings #-} +{-# OPTIONS_HADDOCK prune not-home #-} + +{- | +Module : Data.Attoparsec.Framer +Copyright : (c) 2022 Tim Emiola +Maintainer : Tim Emiola +SPDX-License-Identifier: BSD3 + +Provides the 'Framer' data type that combines an @Attoparsec 'A.Parser'@ with a +a few additional combinators that allow the parser to be used to process frames +from the framed byte streams commonly used in network protocol implementations. + +A @'Framer'@ specifies how the processing function @'runFramer'@ should +parse a byte stream. + +Minimally, a @Framer@ specifies + +* An @'A.Parser'@, used to extract frames from the byte stream +* a @'FrameHandler'@ responsible using the parsed frames +* the bytestream source, represented a 'ByteSource' + + +@'runFramer'@ the 'FrameHandler' is invoked repeatedly; on each +invocation it returns a 'Progression', which indicates if processing should +continue. This makes it possible to terminate for the 'FrameHandler' to signal +that frame processing should terminate. +-} +module Data.Attoparsec.Framer ( + -- * Framer + ByteSource, + Framer, + FrameHandler, + Progression (..), + mkFramer, + mkFramer', + + -- * query/update a @'Framer'@ + setChunkSize, + setOnBadParse, + setOnClosed, + setOnFrame, + chunkSize, + + -- * Run the @Framer@ + runFramer, + runOneFrame, + + -- * Exception handling + -- $exceptions + + -- * exceptions + BrokenFrame (..), + NoMoreInput (..), +) where + +import Control.Exception (Exception) +import Control.Monad.Catch (MonadThrow (..)) +import qualified Data.Attoparsec.ByteString as A +import Data.ByteString (ByteString) +import qualified Data.ByteString as BS +import Data.Maybe (fromMaybe) +import Data.Text (Text) +import qualified Data.Text as Text +import Data.Word (Word32) + + +-- | Handles a parsed @frame@, returning a @Progression@ that indicates if further @frames@ should be parsed. +type FrameHandler m frame = frame -> m Progression + + +-- | A byte stream from which chunks are to be repeatedly retrieved. +type ByteSource m = Word32 -> m ByteString + + +-- | Used by 'FrameHandler' to indicate if additional frames should be parsed. +data Progression + = Stop + | StopUnlessExtra + | Continue + deriving (Eq, Show) + + +-- | Use 'A.Parser' to parse a stream of @frames@ from a bytestream +data Framer m frame = Framer + { framerChunkSize :: !Word32 + , frameByteSource :: !(ByteSource m) + , framerOnFrame :: !(FrameHandler m frame) + , framerParser :: !(A.Parser frame) + , framerOnClosed :: !(m ()) + , framerOnBadParse :: !(Text -> m ()) + } + + +{- | Construct @'Framer'@ that will handle @frames@ repeatedly until a returned + @'Progression'@ stops it. +-} +mkFramer' :: + MonadThrow m => + A.Parser frame -> + FrameHandler m frame -> + ByteSource m -> + Framer m frame +mkFramer' framerParser framerOnFrame frameByteSource = + Framer + { framerParser + , framerOnFrame + , frameByteSource + , framerOnBadParse = \_err -> pure () + , framerOnClosed = throwM NoMoreInput + , framerChunkSize = defaultChunkSize + } + + +-- | Construct @'Framer'@ that loops continuously. +mkFramer :: + MonadThrow m => + A.Parser a -> + (a -> m ()) -> + (Word32 -> m ByteString) -> + Framer m a +mkFramer parser onFrame fetchBytes = + let onFrameContinue x = do + onFrame x + pure Continue + in mkFramer' parser onFrameContinue fetchBytes + + +-- | Repeatedly parse and handle frames until the configured @FrameHandler@ ends handling. +runFramer :: + MonadThrow m => + Framer m a -> + m () +runFramer f = + let Framer + { framerChunkSize = fetchSize + , framerOnBadParse = onErr + , frameByteSource = fetchBytes + , framerOnFrame = onFrame + , framerParser = parser + , framerOnClosed = onClosed + } = f + in runFramer' fetchSize parser fetchBytes onFrame onErr onClosed + + +{- | Parse and handle a single frame. + +The result is tuple of the outstanding unparsed bytes from the bytestream if +any, and a value indicating if the bytestream has terminated. +-} +runOneFrame :: + MonadThrow m => + Maybe ByteString -> + Framer m a -> + m ((Maybe ByteString), Bool) +runOneFrame restMb f = + let Framer + { framerChunkSize = fetchSize + , framerOnBadParse = onErr + , frameByteSource = fetchBytes + , framerOnFrame = onFrame + , framerParser = parser + , framerOnClosed = onClose + } = f + in runOneFrame' restMb fetchSize parser fetchBytes onFrame onErr onClose + + +-- | The chunk size of a @Framer@. +chunkSize :: Framer m a -> Word32 +chunkSize = framerChunkSize + + +-- | Update the chunk size of a @Framer@. +setChunkSize :: Word32 -> Framer m a -> Framer m a +setChunkSize size f = f {framerChunkSize = size} + + +-- | Update the parse error handler of a @Framer@. +setOnBadParse :: (Text -> m ()) -> Framer m a -> Framer m a +setOnBadParse onErr f = f {framerOnBadParse = onErr} + + +-- | Update the @FrameHandler@ of a @Framer@. +setOnFrame :: FrameHandler m frame -> Framer m frame -> Framer m frame +setOnFrame onFrame f = f {framerOnFrame = onFrame} + + +-- | Update the end-of-input handler of a @Framer@. +setOnClosed :: (m ()) -> Framer m a -> Framer m a +setOnClosed onClose f = f {framerOnClosed = onClose} + + +runFramer' :: + MonadThrow m => + Word32 -> + A.Parser a -> + (Word32 -> m ByteString) -> + (a -> m Progression) -> + (Text -> m ()) -> + m () -> + m () +runFramer' fetchSize parser fetchBytes handleFrame onErr onClosed = do + let loop x = do + (next, closed) <- runOneFrame' x fetchSize parser fetchBytes handleFrame onErr onClosed + if not closed then loop next else pure () + loop Nothing + + +runOneFrame' :: + MonadThrow m => + Maybe ByteString -> + Word32 -> + A.Parser a -> + (Word32 -> m ByteString) -> + (a -> m Progression) -> + (Text -> m ()) -> + m () -> + m ((Maybe ByteString), Bool) +runOneFrame' restMb fetchSize parser fetchBytes handleFrame onErr onClose = do + let pullChunk = fetchBytes fetchSize + initial = fromMaybe BS.empty restMb + onParse (A.Fail _ ctxs reason) = do + let errMessage = parsingFailed ctxs reason + if reason == closedReason + then -- WANTED: a typed way of detecting this condition, i.e, + -- it is possible not to rely on a specific error message ? + do + onClose + pure (Nothing, True) + else do + onErr errMessage + throwM $ BrokenFrame reason + onParse (A.Done i r) = do + let extraMb = if BS.null i then Nothing else Just i + doMore <- handleFrame r + case (doMore, extraMb) of + (Stop, _) -> pure (extraMb, True) + (StopUnlessExtra, Nothing) -> pure (extraMb, True) + (_, _) -> pure (extraMb, False) + onParse (A.Partial continue) = pullChunk >>= onParse . continue + A.parseWith pullChunk parser initial >>= onParse + + +parsingFailed :: [String] -> String -> Text +parsingFailed context reason = + let contexts = Text.intercalate "-" (Text.pack <$> context) + cause = if null reason then Text.empty else ":" <> Text.pack reason + in "bad parse:" <> contexts <> cause + + +{- $exceptions + +On failures, @'runFramer'@ throws @'Exception's@ using @'MonadThrow'@ rather +than using an @Either@ or @MonadError@ + +This is because it is intended to be used to parse framed protocol byte streams; +where parsing or connection errors here are typically not recoverable. In haskell +non-recoverable failures are better modelled using @Exceptions@. + +Although it throws 'NoMoreInput' or 'BrokenFrame' when appropriate, it provides +hooks to override these when constructing a 'Framer'. + +By use of 'setOnClosed' and 'setOnBadParse', the caller of @runFramer@ can +completely override the exception type that is raised when @runFramer@ encounters +any failure. +-} + + +{- | Thrown by 'runFramer' or 'runOneFrame' if parsing fails and there is no + handler installed using 'setOnBadParse', or it does not throw an exception. +-} +newtype BrokenFrame = BrokenFrame String + deriving (Eq, Show) + + +instance Exception BrokenFrame + + +{- | Thrown by 'runFramer' or 'runOneFrame' when no further input is available and + no end of input handler is set using 'setOnClosed'. +-} +data NoMoreInput = NoMoreInput + deriving (Eq, Show) + + +instance Exception NoMoreInput + + +closedReason :: String +closedReason = "not enough input" + + +defaultChunkSize :: Word32 +defaultChunkSize = 2048 diff --git a/src/Data/Attoparsec/Framer/Testing.hs b/src/Data/Attoparsec/Framer/Testing.hs new file mode 100644 --- /dev/null +++ b/src/Data/Attoparsec/Framer/Testing.hs @@ -0,0 +1,72 @@ +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# OPTIONS_HADDOCK prune not-home #-} + +{- | +Module : Data.Attoparsec.Framer.Testing +Copyright : (c) 2022 Tim Emiola +Maintainer : Tim Emiola +SPDX-License-Identifier: BSD3 + +This module provides combinators that simplify unit tests of code that +use @'Framer's@. +-} +module Data.Attoparsec.Framer.Testing ( + -- * testing combinators + parsesFromFramerOk, + chunksOfN, +) where + +import Control.Exception (catch) +import qualified Data.Attoparsec.ByteString as A +import Data.Attoparsec.Framer +import Data.ByteString (ByteString) +import qualified Data.ByteString as BS +import Data.IORef ( + IORef, + modifyIORef', + newIORef, + readIORef, + writeIORef, + ) +import Data.List (unfoldr) +import Data.Word (Word32) + + +{- | Creates a 'Framer' and uses 'runFramer to confirm that the expect frames + are received ' +-} +parsesFromFramerOk :: Eq a => (a -> ByteString) -> A.Parser a -> Word32 -> [a] -> IO Bool +parsesFromFramerOk asBytes parser chunkSize' wanted = do + chunkStore <- newIORef Nothing + dst <- newIORef [] + let updateDst x = modifyIORef' dst ((:) x) + mkChunks n = mconcat $ map (chunksOfN n . asBytes) wanted + src = nextFrom' mkChunks chunkStore + frames = setChunkSize chunkSize' $ mkFramer parser updateDst src + runFramer frames `catch` (\(_e :: NoMoreInput) -> pure ()) + + got <- readIORef dst + pure $ got == reverse wanted + + +-- | Split a 'ByteString' into chunks of given size +chunksOfN :: Int -> ByteString -> [ByteString] +chunksOfN x b = + let go y = + let taken = BS.take x y + in if BS.null taken then Nothing else Just (taken, BS.drop x y) + in unfoldr go b + + +nextFrom' :: + (Int -> [ByteString]) -> IORef (Maybe [ByteString]) -> Word32 -> IO ByteString +nextFrom' initChunks chunkStore chunkSize' = do + readIORef chunkStore >>= \case + Nothing -> do + writeIORef chunkStore $ Just $ initChunks $ fromIntegral chunkSize' + nextFrom' initChunks chunkStore chunkSize' + Just [] -> pure BS.empty + Just (x : xs) -> do + writeIORef chunkStore $ Just xs + pure x diff --git a/test/Attoparsec/ToyFrameSpec.hs b/test/Attoparsec/ToyFrameSpec.hs new file mode 100644 --- /dev/null +++ b/test/Attoparsec/ToyFrameSpec.hs @@ -0,0 +1,67 @@ +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# OPTIONS_HADDOCK prune not-home #-} + +{- | +Module : Attoparsec.ToyFrameSpec +Copyright : (c) 2022 Tim Emiola +Maintainer : Tim Emiola +SPDX-License-Identifier: BSD3 +-} +module Attoparsec.ToyFrameSpec (spec) where + +import Attoparsec.ToyFrame +import Control.Exception (ArithException (..), throwIO) +import Data.Attoparsec.Framer +import Data.Attoparsec.Framer.Testing +import qualified Data.ByteString as BS +import Data.Word (Word32) +import Test.Hspec +import Test.QuickCheck +import Test.QuickCheck.Monadic (forAllM, monadicIO, run) + + +spec :: Spec +spec = describe "ToyFrame" $ do + context "parser" $ + it "should roundtrip with 'builder'" prop_trip + + mapM_ receivesWithChunksOf [16, 256, 1024, 2048, 4096, 8192] + + context "when input ends, receivesFrames" $ do + let basic = mkFramer parser (const $ pure ()) $ const $ pure BS.empty + otherError = setOnClosed (throwIO Underflow) basic + noError = setOnClosed (pure ()) basic + + it "should use the default close handler" $ do + runFramer basic `shouldThrow` (\NoMoreInput -> True) + + context "when a throwing closed handler is installed" $ do + it "should throw" $ do + runFramer otherError `shouldThrow` (\x -> x == Underflow) + + context "when a closed handler does not throw an exception" $ do + it "should finish without throwing" $ do + runFramer noError `shouldReturn` () + + +receivesWithChunksOf :: Word32 -> SpecWith () +receivesWithChunksOf chunkSize' = do + context ("when chunk size is " ++ show chunkSize') $ + context "runFramer" $ + it "should parse into frames" $ + prop_runFramer chunkSize' + + +prop_trip :: Property +prop_trip = + withMaxSuccess 15000 $ + forAll genFullFrame $ + \p -> parse (asBytes p) == Just p + + +prop_runFramer :: Word32 -> Property +prop_runFramer chunkSize' = monadicIO $ + forAllM (listOf1 genFullFrame) $ + \ps -> run $ parsesFromFramerOk asBytes parser chunkSize' ps diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -0,0 +1,23 @@ +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module Main where + +import qualified Attoparsec.ToyFrameSpec as ToyFrame +import System.IO ( + BufferMode (..), + hSetBuffering, + stderr, + stdout, + ) +import Test.Hspec + + +main :: IO () +main = do + hSetBuffering stdout NoBuffering + hSetBuffering stderr NoBuffering + + hspec $ do + ToyFrame.spec diff --git a/toy/Attoparsec/ToyFrame.hs b/toy/Attoparsec/ToyFrame.hs new file mode 100644 --- /dev/null +++ b/toy/Attoparsec/ToyFrame.hs @@ -0,0 +1,169 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE OverloadedStrings #-} +{-# OPTIONS_HADDOCK prune not-home #-} + +{- | +Module : ToyFrame +Copyright : (c) 2022 Tim Emiola +Maintainer : Tim Emiola +SPDX-License-Identifier: BSD3 + +Provides the 'Header' and 'FullFrame' data types used by the demo Client and +Server along with useful common functions. +-} +module Attoparsec.ToyFrame ( + -- * data types + Header (..), + Payload (Payload), + FullFrame, + + -- * functions + asBytes, + builder, + buildFrameHeader, + parse, + parser, + parseHeader, + + -- * sample data + genPayload, + genAscFullFrames, + genHeader, + genFullFrame, + someTriggers, +) where + +import qualified Data.Attoparsec.Binary as A +import qualified Data.Attoparsec.ByteString as A +import Data.ByteString (ByteString) +import qualified Data.ByteString as BS +import Data.ByteString.Builder +import qualified Data.ByteString.Lazy as LBS +import Data.Word (Word32, Word8) +import Test.QuickCheck ( + Arbitrary (arbitrary), + Gen, + chooseEnum, + generate, + vectorOf, + ) + + +-- | Class for datastructures that specify the frame size of a payload; used by 'parseSizedFrame' +class FrameSize a where + frameSize :: a -> Word32 + + +-- | Creates an 'A.Parser' that parses a datastructure specifying a frame size, and then a separate framed one with the given size +parseSizedFrame :: FrameSize h => A.Parser h -> A.Parser b -> A.Parser (h, b) +parseSizedFrame parseHead parseBody = do + h <- parseHead + let size = frameSize h + body <- fixed (fromIntegral size) parseBody + pure (h, body) + + +fixed :: Word32 -> A.Parser a -> A.Parser a +fixed i p = do + intermediate <- A.take $ fromIntegral i + case A.parseOnly (p <* A.endOfInput) intermediate of + Left x -> fail x + Right x -> pure x + + +-- | @FullFrame@ is a header followed by a payload. +type FullFrame = (Header, Payload) + + +-- | @Header@ indicates a message index and the size of the payload +data Header = Header + { hResponseSize :: !Word32 + , hMaxPayloadSize :: !Word32 + } + deriving (Eq, Show) + + +instance FrameSize Header where + frameSize = hMaxPayloadSize + + +newtype Payload = Payload ByteString + deriving (Eq, Show) + + +parseHeader :: A.Parser Header +parseHeader = Header <$> A.anyWord32be <*> A.anyWord32be + + +buildFrameHeader :: Header -> Builder +buildFrameHeader fh = word32BE (hResponseSize fh) <> word32BE (hMaxPayloadSize fh) + + +parseFrame :: A.Parser Payload +parseFrame = fmap Payload $ A.takeByteString + + +parser :: A.Parser FullFrame +parser = parseSizedFrame parseHeader parseFrame + + +builder' :: Word32 -> Payload -> Builder +builder' hResponseSize (Payload b) = + let hMaxPayloadSize = fromIntegral $ BS.length b + header = Header {hResponseSize, hMaxPayloadSize} + in buildFrameHeader header <> byteString b + + +builder :: FullFrame -> Builder +builder (header, body) = builder' (hResponseSize header) body + + +asBytes :: FullFrame -> BS.ByteString +asBytes = LBS.toStrict . toLazyByteString . builder + + +parse :: BS.ByteString -> Maybe FullFrame +parse = A.maybeResult . A.parse parser + + +genPrintable :: Gen Word8 +genPrintable = chooseEnum (32, 127) + + +genPayload :: Word32 -> Gen Payload +genPayload size = fmap (Payload . BS.pack) $ vectorOf (fromIntegral size) genPrintable + + +genHeader :: Gen Header +genHeader = Header <$> arbitrary <*> chooseEnum (2, 32) + + +genClientTrigger :: Gen Header +genClientTrigger = Header <$> (chooseEnum (32, 4096)) <*> (chooseEnum (32, 1024)) + + +someTriggers :: Int -> IO [Header] +someTriggers count = generate $ vectorOf count $ genClientTrigger + + +genFullFrame :: Gen FullFrame +genFullFrame = do + header <- genHeader + body <- genPayload $ hMaxPayloadSize header + pure (header, body) + + +genEnumPayload' :: Word32 -> Word32 -> Gen [(Word32, Payload)] +genEnumPayload' count maxSize = vectorOf (fromIntegral count) $ do + aSize <- chooseEnum (1, maxSize) + payload <- genPayload aSize + pure (aSize, payload) + + +genAscFullFrames :: Word32 -> Word32 -> IO [FullFrame] +genAscFullFrames count maxSize = generate $ do + xs <- genEnumPayload' count maxSize + let toFullFrame (hResponseSize, (hMaxPayloadSize, p)) = (Header {hMaxPayloadSize, hResponseSize}, p) + pure $ map toFullFrame $ zip [1 ..] xs diff --git a/toy/Client.hs b/toy/Client.hs new file mode 100644 --- /dev/null +++ b/toy/Client.hs @@ -0,0 +1,151 @@ +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE OverloadedStrings #-} +{-# OPTIONS_HADDOCK prune not-home #-} + +{- | +Module : Client +Copyright : (c) 2022 Tim Emiola +Maintainer : Tim Emiola +SPDX-License-Identifier: BSD3 + +A demo client that contacts the demo server via a framed protocol. + +* it generates a sequence of @'Header's@ that represent server requests + * each request specifies the number of frames the server should send in response +* it connects to the server and begins sending the generated requests +* between each request, it consumes the frames sent by the server in response + * it tracks them and confirms that the requested number of frames are received +* finally, it sends a special @Header@ that signals to the server that it should close the connection +-} +module Main (main) where + +import Attoparsec.ToyFrame ( + FullFrame, + Header (..), + Payload (..), + buildFrameHeader, + parser, + someTriggers, + ) +import Data.Attoparsec.Framer ( + Framer, + mkFramer, + runFramer, + setOnBadParse, + setOnClosed, + ) +import qualified Data.ByteString as BS +import Data.ByteString.Builder (toLazyByteString) +import qualified Data.ByteString.Lazy as LBS +import Data.IORef ( + IORef, + newIORef, + readIORef, + writeIORef, + ) +import Data.Text (Text) +import qualified Data.Text.IO as Text +import Network.Run.TCP (runTCPClient) +import Network.Socket (Socket) +import Network.Socket.ByteString (recv, sendAll) + + +main :: IO () +main = runTCPClient "127.0.0.1" "3927" $ \s -> do + -- generate the sequence of headers to send to the server, save in the tracking IORef + trackingRef <- someTriggers 1024 >>= newTrackingRef + -- trigger an initial response from the server + trackFrames trackingRef (socketSink s) Nothing + -- get the Framer; mkClientFramer uses trackFrames as its FrameHandler + runFramer $ mkClientFramer trackingRef s + -- print a summary after completing + tracking <- readIORef trackingRef + putStrLn $ "Received " ++ (show $ trackingFrames tracking) ++ " of total size " ++ (show $ trackingBytes tracking) + + +mkClientFramer :: IORef Tracking -> Socket -> Framer IO FullFrame +mkClientFramer ref s = + let sink = socketSink s + onFullFrame' f = trackFrames ref sink $ Just f + in setOnClosed onClosed $ + setOnBadParse (onFailedParse' sink) $ + mkFramer parser onFullFrame' (recv s . fromIntegral) + + +data Tracking = Tracking + { trackingLeft :: ![Header] + , trackingBytes :: !Int + , trackingFrames :: !Int + , trackingCountdown :: !(Int, Int) + } + + +newTrackingRef :: [Header] -> IO (IORef Tracking) +newTrackingRef xs = newIORef $ Tracking xs 0 0 (0, 0) + + +trackFrames :: IORef Tracking -> ByteSink -> Maybe FullFrame -> IO () +trackFrames trackingRef sink frameMb = do + t <- readIORef trackingRef + let (target, lastCount) = trackingCountdown t + nextCount = lastCount + 1 + nextFrames = trackingFrames t + 1 + incrWithPayload p = + t + { trackingCountdown = (target, nextCount) + , trackingFrames = nextFrames + , trackingBytes = trackingBytes t + BS.length p + } + countedUp = nextCount == target + incrOr p' action = + if not countedUp + then writeIORef trackingRef $ incrWithPayload p' + else action + + case (frameMb, trackingLeft t) of + (Just (_, Payload p'), []) -> incrOr p' $ do + writeIORef trackingRef $ incrWithPayload p' + sink bye + (Just (_, Payload p'), x : xs) -> incrOr p' $ do + let updatedTracking = + (incrWithPayload p') + { trackingCountdown = (fromIntegral $ hResponseSize x, 0) + , trackingLeft = xs + } + writeIORef trackingRef updatedTracking + sink $ asBytes x + (Nothing, x : xs) -> do + writeIORef + trackingRef + t + { trackingCountdown = (fromIntegral $ hResponseSize x, 0) + , trackingLeft = xs + } + sink $ asBytes x + (Nothing, []) -> sink bye + + +type ByteSink = BS.ByteString -> IO () + + +socketSink :: Socket -> ByteSink +socketSink = sendAll + + +bye :: BS.ByteString +bye = asBytes $ Header 0 0 + + +onFailedParse' :: ByteSink -> Text -> IO () +onFailedParse' sink cause = do + -- if does not parse as a full frame immediately terminate the connection + Text.putStrLn $ "parse error ended a connection to a toy server: " <> cause + sink bye + + +onClosed :: IO () +onClosed = Text.putStrLn "finished at the server too!" + + +asBytes :: Header -> BS.ByteString +asBytes = LBS.toStrict . toLazyByteString . buildFrameHeader diff --git a/toy/Server.hs b/toy/Server.hs new file mode 100644 --- /dev/null +++ b/toy/Server.hs @@ -0,0 +1,81 @@ +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE OverloadedStrings #-} +{-# OPTIONS_HADDOCK prune not-home #-} + +{- | +Module : Server +Copyright : (c) 2022 Tim Emiola +Maintainer : Tim Emiola +SPDX-License-Identifier: BSD3 + +A demo server that responds to the demo client via a framed protocol. + +* once started, it waits for client connections + * it loops forever, eventually it needs to be shutdown using CTRL-C +* when a client connects, it responds to valid client requests, on invalid + requests the connection is closed +* valid requests are @'Header's@ that indicate the number of frames to send and + their maximum size + * it sends a series of responses that match the request specification + * if the requested response size is 0, it closes the connection +-} +module Main (main) where + +import Attoparsec.ToyFrame (Header (..), asBytes, genAscFullFrames, parseHeader) +import Data.Attoparsec.Framer ( + Framer, + Progression (..), + mkFramer', + runFramer, + setOnBadParse, + setOnClosed, + ) +import qualified Data.ByteString as BS +import Data.Text (Text) +import qualified Data.Text.IO as Text +import Network.Run.TCP (runTCPServer) +import Network.Socket (Socket) +import Network.Socket.ByteString (recv, sendAll) + + +main :: IO () +main = runTCPServer Nothing "3927" $ \s -> do + Text.putStrLn "a toy client connected" + runFramer $ mkServerFramer s + + +type ByteSink = BS.ByteString -> IO () + + +mkServerFramer :: Socket -> Framer IO Header +mkServerFramer s = + let onHeader' = onHeader $ sendAll s + in setOnClosed onClosed $ + setOnBadParse onFailedParse $ + mkFramer' parseHeader onHeader' (recv s . fromIntegral) + + +onHeader :: ByteSink -> Header -> IO Progression +onHeader sink Header {hResponseSize, hMaxPayloadSize} = do + if (hResponseSize == 0) + then -- hResponseSize is 0; the client means 'bye', stop waiting for input + do + Text.putStrLn "a toy client sent bye" + pure Stop + else do + -- hResponseSize > 0; starting from 1, send a frame with a body whose max size is hMaxPayloadSize + -- generate a list of frames counting up to the index provided in the header + toSend <- genAscFullFrames hResponseSize hMaxPayloadSize + mapM_ sink $ map asBytes toSend + pure Continue + + +onFailedParse :: Text -> IO () +onFailedParse cause = do + -- if does not parse as a frame header terminate the connection + -- no explicit exception is raised here, so runFramer throws + Text.putStrLn $ "parse error ended a connection from a toy client: " <> cause + + +onClosed :: IO () +onClosed = Text.putStrLn "a toy client closed a connection"