streaming-utils 0.1.4.2 → 0.1.4.3
raw patch · 3 files changed
+400/−15 lines, 3 filesdep +networkdep +network-simpledep ~aesondep ~basedep ~json-stream
Dependencies added: network, network-simple
Dependency ranges changed: aeson, base, json-stream, transformers
Files
- Data/ByteString/Streaming/Aeson.hs +15/−5
- Streaming/Network/TCP.hs +369/−0
- streaming-utils.cabal +16/−10
Data/ByteString/Streaming/Aeson.hs view
@@ -20,8 +20,9 @@ Here we use a long top level array of objects from <https://raw.githubusercontent.com/ondrap/json-stream/master/benchmarks/json-data/buffer-builder.json a file> - @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+ @json-streams@ benchmarking directory. Each object in the top level array + has a \"friends\" field with an assocated array of friends; each of these has a \"name\".+ Here, we extract the name of each friend of each person recorded in the level array, and enumerate them all: > {-#LANGUAGE OverloadedStrings #-}@@ -54,7 +55,8 @@ 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. + friends\' names as soon as they come. With appropriate instances, we could+ of course just stream the objects in the top-level array instead. -} @@ -184,11 +186,19 @@ If the parser is fitted to recognize only one thing, then zero or one item will be yielded; if it uses combinators like @arrayOf@, - it will stream many values as they arise. + it will stream many values as they arise. See the example at the top of this module,+ in which values inside a top level array are emitted as they are parsed. Aeson would+ accumulate the whole bytestring before declaring on the contents of the array.+ This of course makes sense, since attempt to parse a json array may end with + a bad parse, invalidating the json as a whole. With @json-streams@, a bad + parse will also of course emerge in the end, but only after the initial good parses+ are streamed. This too makes sense though, but in a smaller range of contexts + -- for example, where one is folding over the parsed material.+ This function is closely modelled on 'Data.JsonStream.Parser.parseByteString' and - 'Data.JsonStream.Parser.parseLazyByteString'+ 'Data.JsonStream.Parser.parseLazyByteString' from @Data.JsonStream.Parser@. -} streamParse
+ Streaming/Network/TCP.hs view
@@ -0,0 +1,369 @@+{-# LANGUAGE RankNTypes #-}++-- | This hyper-minimal module closely follows the corresponding module +-- in Renzo Carbonara' 'pipes-network' package. It is meant to be used together with+-- the "Network.Simple.TCP" module from Carbonara\'s @network-simple@ package, which is+-- completely re-exported from this module. +++module Streaming.Network.TCP (+ -- * Receiving+ fromSocket++ -- * Sending+ , toSocket+ + -- * Simple demos+ -- $demos+ + -- * Source+ -- $source + + -- * Re-exports+ -- $exports+ , module Network.Simple.TCP++ ) where++import qualified Data.ByteString as B+import qualified Data.ByteString.Lazy as BL+import qualified Data.ByteString.Streaming as Q++import Foreign.C.Error (errnoToIOError, eTIMEDOUT)+import qualified Network.Socket.ByteString as NSB+import Network.Simple.TCP+ (connect, serve, listen, accept, acceptFork,+ bindSock, connectSock, closeSock, recv, send, sendLazy,+ sendMany, withSocketsDo, HostName,+ HostPreference(HostAny, HostIPv4, HostIPv6, Host),+ ServiceName, SockAddr, Socket)+import System.Timeout (timeout)+import Control.Monad (when)+import Control.Monad.IO.Class+--------------------------------------------------------------------------------+++{- | Receives bytes from a connected socket with a maximum chunk size.+ The bytestream ends if the remote peer closes its side of the connection+ or EOF is received. The implementation is as follows:++> fromSocket sock nbytes = loop where+> loop = do+> bs <- liftIO (NSB.recv sock nbytes)+> if B.null bs +> then return ()+> else Q.chunk bs >> loop+-}+fromSocket+ :: MonadIO m+ => Socket -- ^Connected socket.+ -> Int -- ^Maximum number of bytes to receive and send+ -- dowstream at once. Renzo recommends+ -- using @4096@ if you don't have a special purpose.+ -> Q.ByteString m ()+fromSocket sock nbytes = loop where+ loop = do+ bs <- liftIO (NSB.recv sock nbytes)+ if B.null bs + then return ()+ else Q.chunk bs >> loop+{-# INLINABLE fromSocket #-}+ ++--------------------------------------------------------------------------------+++{- | Connect a stream of bytes to the remote end. The implementation+ is again very simple:++> toSocket sock = loop where+> loop bs = do+> e <- Q.nextChunk bs+> case e of+> Left r -> return r+> Right (b,rest) -> send sock b >> loop rest+-}+toSocket+ :: MonadIO m+ => Socket -- ^Connected socket.+ -> Q.ByteString m r+ -> m r+toSocket sock = loop where+ loop bs = do+ e <- Q.nextChunk bs+ case e of+ Left r -> return r+ Right (b,rest) -> send sock b >> loop rest+{-# INLINABLE toSocket #-}++{- $demos++Here are a collection of little @hello telnet world@ programs, following Michael Snoyberg's <http://www.yesodweb.com/blog/2014/03/network-conduit-async post> on using @network-conduit@ +together with @async@ I hope the reader will find that they are a bit more intelligible when+we think naively of 'byte streams' as ordinary Haskell entities, rather than conduits trapped in+a framework. (In fact they're pretty straightforward either way, of course.) The complete source is appended to this module below.++- `serverToUpper` - a server on 4000 that sends back input sent e.g. with telnet upper-cased+- `serverDouble` - a server on 4001 that sends back input doubled, `Char8` by `Char8`+- `clientToUpper` - a client through which the user interacts directly with the upper-caser+- `clientPipeline` - a client that sends material to the uppercasing server and then the doubling server and returns it to the user+- `proxyToUpper` - a proxy on 4002 that sends input to the uppercasing server on 4000+- `proxyAuth` - a proxy on 4003 that asks for demands authorization before condescending to send user input to the upper-casing server on 4000++The following remarks will require that eight+instances of a terminal all be opened; a+crude option parser will make the examples usable with+one executable:++> $ streaming-network-tcp-examples --help+> Usage: streaming-network-tcp-examples COMMAND++> Available options:+> -h,--help Show this help text++> Available commands:+> ClientPipeline +> ClientToUpper +> ProxyAuth +> ProxyToUpper +> ServePipes +> ServerDouble +> ServerToUpper++Since most examples use the uppercasing service,+which looks like this:+++> serverToUpper :: IO ()+> serverToUpper = do+> putStrLn "Opening upper-casing service on 4000"+> serve (Host "127.0.0.1") "4000" $ \(client,_) -> +> fromSocket client 4096 -- raw bytes are received from a telnet user or the like+> & Q.map toUpper -- we map them to uppercase+> & toSocket client -- and send them back+++we start it in one terminal:++> term1$ streaming-network-tcp-examples ServerToUpper+> Opening upper-casing service on 4000+ +then in another terminal we can telnet to it:++> term2$ telnet localhost 4000+> Trying 127.0.0.1...+> Connected to localhost.+> Escape character is '^]'.+> hello -- <-- our input+> HELLO+> ...++or we can scrap telnet and use a dedicated Haskell client. This is a little subtler:++> clientToUpper :: IO ()+> clientToUpper = connect "127.0.0.1" "4000" $ \(socket,_) -> do+> let act1 = toSocket socket Q.stdin -- we send our stdin to the service+> act2 = Q.stdout (fromSocket socket 4096) -- we read our stdout from the service+> concurrently act1 act2 -- but we do each on a separate thread+> return ()+ +Here, we stream standard input to the remote end indefinitely, and we stream +news from the remote end to standard output indefinitely. The two open ended processes are run them together+with @Control.Concurrent.Async.concurrently@, so we see:++> term3$ streaming-network-tcp-examples ClientToUpper+> el pueblo unido jamas sera vencido! -- our input+> EL PUEBLO UNIDO JAMAS SERA VENCIDO!+> el pueblo unido jamas sera vencido! -- our input+> EL PUEBLO UNIDO JAMAS SERA VENCIDO!+> ...+ +To complicate the system of connections, we can also start a second server,+which again just makes a trivial alteration in the bytestream, doubling each @Char8@:++> serverDoubler :: IO ()+> serverDoubler = do +> putStrLn "Double server available on 4001"+> serve (Host "127.0.0.1") "4001" $ \(socket, remoteAddr) -> +> fromSocket socket 4096 -- raw bytes from a client+> & Q.toChunks -- are munged ...+> & S.map (B.concatMap (\x -> B.pack [x,x])) -- with standard bytestream materials+> & Q.fromChunks -- ...+> & toSocket socket -- and sent back+++starting it up thus:++> term4$ streaming-network-tcp-examples ServerDouble++we see:++> term5$ telnet localhost 4001+> Trying 127.0.0.1...+> Connected to localhost.+> Escape character is '^]'.+> hello+> hheelllloo++Now we complicate our use of the @async@ library with a +Haskell client that interacts with @4000@ and @4001@ together:++> clientPipeline :: IO ()+> clientPipeline = do+> putStrLn "We will connect stdin to 4000 and 4001 in succession."+> putStrLn "Input will thus be uppercased and doubled char-by-char.\n"+> connect "127.0.0.1" "4000" $ \(socket1,_) ->+> connect "127.0.0.1" "4001" $ \(socket2,_) ->+> do let act1 = toSocket socket1 Q.stdin+> -- we send out stdin to the uppercaser+> act2 = toSocket socket2 (fromSocket socket1 4096)+> -- we send the results from the uppercase to the doubler+> act3 = Q.stdout (fromSocket socket2 4096)+> -- we send the doubler's output to stdout+> runConcurrently $ Concurrently act1 *> -- all this simultaneously+> Concurrently act2 *>+> Concurrently act3++Note the use of the `Applicative` instance for `Concurrently` from the+`async` library to make the three stream operations simultaneous. Then we see:++> term6$ streaming-network-tcp-examples ClientPipeline+> hello+> HHEELLLLOO++The upper-caser is open on @4000@ but don\'t tell the children. +The last program does a little manipulation of the bytestream to +demand authorization on 4003 ++> term7$ streaming-network-tcp-examples ProxyAuth++which then elsewhere permits++> term8$ telnet localhost 4003+> Trying 127.0.0.1...+> Connected to localhost.+> Escape character is '^]'.+> Username: spaceballs+> Password: 12345+> Successfully authenticated.+> hello+> HELLO+> hello!+> HELLO!+ +-}+++{- $source++> -- streaming-network-tcp-examples.hs++>{-#LANGUAGE OverloadedStrings #-}+>module Main where+>+>import Streaming+>import Streaming.Network.TCP+>import qualified Streaming.Prelude as S+>import qualified Data.ByteString.Streaming as Q+>+>import Control.Concurrent.Async -- cabal install async+>import qualified Data.ByteString as B+>import Data.ByteString (ByteString)+>import Data.Word8 (toUpper, _cr) -- cabal install word8+>import Data.Function ((&))+>import Options.Applicative -- cabal install optparse-applicative+>import Control.Applicative+>import Control.Monad+>import Data.Monoid+>+>serverToUpper :: IO ()+>serverToUpper = do+> putStrLn "Opening upper-casing service on 4000"+> serve (Host "127.0.0.1") "4000" $ \(client,_) -> +> toSocket client $ Q.map toUpper $ fromSocket client 4096 +>+>serverDoubler :: IO ()+>serverDoubler = do +> putStrLn "Double server available on 4001"+> serve (Host "127.0.0.1") "4001" $ \(client, remoteAddr) -> +> fromSocket client 4096+> & Q.toChunks+> & S.map (B.concatMap (\x -> B.pack [x,x]))+> & Q.fromChunks+> & toSocket client+>+>clientToUpper :: IO ()+>clientToUpper = connect "127.0.0.1" "4000" $ \(server,_) -> do+> let act1 = toSocket server Q.stdin +> act2 = Q.stdout (fromSocket server 4096) +> concurrently act1 act2 +> return ()+>+>clientPipeline :: IO ()+>clientPipeline = do+> putStrLn "We will connect stdin to 4000 and 4001 in succession."+> putStrLn "Input will thus be uppercased and doubled char-by-char.\n"+> connect "127.0.0.1" "4000" $ \(socket1,_) ->+> connect "127.0.0.1" "4001" $ \(socket2,_) ->+> do let act1 = toSocket socket1 (Q.stdin)+> act2 = toSocket socket2 (fromSocket socket1 4096)+> act3 = Q.stdout (fromSocket socket2 4096)+> runConcurrently $ Concurrently act1 *>+> Concurrently act2 *>+> Concurrently act3+>+>proxyToUpper :: IO ()+>proxyToUpper = +> serve (Host "127.0.0.1") "4002" $ \(client, _) ->+> connect "127.0.0.1" "4000" $ \(server, _) -> +> do let act1 = toSocket server (fromSocket client 4096)+> act2 = toSocket client (fromSocket server 4096)+> concurrently act1 act2+> return ()+>+>proxyAuth :: IO ()+>proxyAuth = serve (Host "127.0.0.1") "4003" process +> where+> process (client, _) =+> do from_client <- toSocket client (checkAuth (fromSocket client 4096))+> connect "127.0.0.1" "4000" $ \(server,_) ->+> do let pipe_forward = toSocket server from_client +> pipe_back = toSocket client (fromSocket server 4096) +> concurrently pipe_forward pipe_back+> return ()+>+> checkAuth :: MonadIO m => Q.ByteString m r -> Q.ByteString m (Q.ByteString m r)+> checkAuth p = do +> Q.chunk "Username: "+> (username,p1) <- lift $ shortLineInput 80 p+> Q.chunk "Password: "+> (password,p2) <- lift $ shortLineInput 80 p1+> if (username, password) `elem` creds+> then Q.chunk "Successfully authenticated.\n"+> else do Q.chunk "Invalid username/password.\n"+> error "Invalid authentication, please log somewhere..."+> return p2 -- when using `error`+> +> shortLineInput n bs = do+> (bs:>rest) <- Q.toStrict $ Q.break (==10) $ Q.splitAt n bs+> return $ (B.filter (/= _cr) bs, Q.drop 1 $ rest >>= id) +> +> creds :: [(ByteString, ByteString)]+> creds = [ ("spaceballs", "12345") ]+>+>main :: IO ()+>main = join $ execParser (info opts idm) where+>+> opts :: Parser (IO ())+> opts = helper <*> subparser stuff where +> stuff = mconcat+> [ command "ClientPipeline" (info (pure clientPipeline) idm)+> , command "ClientToUpper" (info (pure clientToUpper) idm)+> , command "ProxyAuth" (info (pure proxyAuth) idm)+> , command "ProxyToUpper" (info (pure proxyToUpper) idm)+> , command "ServerDouble" (info (pure serverDoubler) idm)+> , command "ServerToUpper" (info (pure serverToUpper) idm)+> ]+>++-}
streaming-utils.cabal view
@@ -1,5 +1,5 @@ name: streaming-utils-version: 0.1.4.2+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 the <http://hackage.haskell.org/package/streaming streaming> and @@ -53,25 +53,31 @@ location: https://github.com/michaelt/streaming-utils library- exposed-modules: Data.Attoparsec.ByteString.Streaming,- Data.ByteString.Streaming.HTTP,- Data.ByteString.Streaming.Aeson,- Streaming.Pipes+ exposed-modules: Data.Attoparsec.ByteString.Streaming+ , Data.ByteString.Streaming.HTTP+ , Data.ByteString.Streaming.Aeson+ , Streaming.Pipes+ , Streaming.Network.TCP+ -- , Streaming.Pipes.Concurrent+ -- other-modules: other-extensions: CPP, Trustworthy - build-depends: base >=4.7 && <4.9, - transformers >=0.4 && <0.5, + 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,+ streaming-bytestring >= 0.1.4.0 && < 0.1.4.5, bytestring > 0.10.0 && < 0.11.0, pipes >= 4.0 && < 4.2,+ network-simple,+ network, + -- pipes-concurrency >= 2.0 && < 2.1, http-client >=0.2 && <0.5, http-client-tls <0.3,- aeson > 0.8 && <0.9.1.0,- json-stream == 0.4.0.*,+ aeson > 0.8 && <0.11.3,+ json-stream > 0.4.0 && < 0.4.2, resourcet > 1.0 && < 1.2 -- hs-source-dirs: