conduit-network-stream (empty) → 0.1
raw patch · 7 files changed
+406/−0 lines, 7 filesdep +basedep +bytestringdep +conduitsetup-changed
Dependencies added: base, bytestring, conduit, mtl, network-conduit, resourcet
Files
- LICENSE +30/−0
- Setup.hs +6/−0
- conduit-network-stream.cabal +39/−0
- src/Data/Conduit/Network/Stream.hs +139/−0
- src/Data/Conduit/Network/Stream/Exceptions.hs +17/−0
- src/Data/Conduit/Network/Stream/Header.hs +133/−0
- src/Data/Conduit/Network/Stream/Internal.hs +42/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c)2012, Nils Schweinsberg <mail@nils.cc>++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 Nils Schweinsberg <mail@nils.cc> 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.
+ Setup.hs view
@@ -0,0 +1,6 @@+#!/usr/bin/env runhaskell++import Distribution.Simple++main :: IO ()+main = defaultMainWithHooks simpleUserHooks
+ conduit-network-stream.cabal view
@@ -0,0 +1,39 @@+Name: conduit-network-stream+Version: 0.1++Synopsis: A base layer for network protocols using Conduits+Description: A base layer for network protocols using Conduits++License: BSD3+License-file: LICENSE+Author: Nils Schweinsberg <mail@nils.cc>+Maintainer: mail@nils.cc+-- Copyright: ++Category: Network++Build-type: Simple+Cabal-version: >=1.6++-- Extra-source-files: ++source-repository head+ type: git+ location: git://github.com/mcmaniac/conduit-network-stream.git+ tag: 0.1++Library++ HS-source-dirs: src+ GHC-options: -Wall++ Build-depends: + base == 4.*, bytestring >= 0.10, mtl,+ resourcet, conduit, network-conduit++ Exposed-modules:+ Data.Conduit.Network.Stream+ Data.Conduit.Network.Stream.Exceptions+ Other-modules: + Data.Conduit.Network.Stream.Header+ Data.Conduit.Network.Stream.Internal
+ src/Data/Conduit/Network/Stream.hs view
@@ -0,0 +1,139 @@+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleInstances #-}++module Data.Conduit.Network.Stream+ ( -- * Network streams+ Stream+ -- ** Sending+ , Sendable, send1, sendList+ -- ** Receiving+ , Streamable (receive), receiveLast, close+ -- ** Manual sending/receiving+ , next+ , (~~)+ , sink1, sinkList, sinkList'+ --, sinkListStart, sinkListElems, sinkListEnd+ ) where++import Control.Monad.Trans+import Control.Monad.Trans.Resource+import Data.ByteString (ByteString)+import Data.Conduit hiding (($$))+import Data.Conduit.Network++import qualified Data.Conduit as C +import qualified Data.Conduit.List as CL+import qualified Data.Conduit.Binary as CB+import qualified Data.ByteString as BS+import qualified Data.ByteString.Lazy as BL++import Data.Conduit.Network.Stream.Exceptions+import Data.Conduit.Network.Stream.Header+import Data.Conduit.Network.Stream.Internal++-- | Lifted version of @($$)@+infixr 0 ~~+(~~) :: Monad m => Source (Stream m) a -> Sink a (Stream m) b -> m b+src ~~ sink = stream_base $ src C.$$ sink++class Sendable a m where+ encode :: Conduit a (Stream m) ByteString++instance Monad m => Sendable ByteString m where+ encode = encodeBS++instance Monad m => Sendable (Int, BL.ByteString) m where+ encode = encodeLazyBS++-- | Send one single 'ByteString' over the network connection (if there is more+-- than one 'ByteString' in the pipe it will be discarded)+send1 :: (Monad m, Sendable a m) => AppData m -> Source (Stream m) a -> m ()+send1 ad src = src ~~ sink1 ad++-- | Send all 'ByteString's in the pipe as a list over the network connection+sendList :: (Monad m, Sendable a m) => AppData m -> Source (Stream m) a -> m ()+sendList ad src = src ~~ sinkList ad++sink1 :: (Monad m, Sendable a m) => AppData m -> Sink a (Stream m) ()+sink1 ad = do+ CL.isolate 1 =$ encode =$ sink+ CL.sinkNull+ where+ sink = transPipe lift (appSink ad)++sinkList :: (Monad m, Sendable a m) => AppData m -> Sink a (Stream m) ()+sinkList ad = do+ sinkListStart ad+ sinkListElems ad+ sinkListEnd ad++class Streamable source m where+ -- | Get the next package from a streamable source (calls 'next'+ -- automatically)+ receive :: source -> Sink BL.ByteString (Stream m) b -> m (ResumableSource (Stream m) ByteString, b)++instance MonadResource m => Streamable (AppData m) m where+ receive ad sink = stream_base $+ transPipe lift (appSource ad) $$+ next =$ sink++instance MonadResource m => Streamable (ResumableSource (Stream m) ByteString) m where+ receive src sink = stream_base $+ src $$++ next =$ sink++-- | Get the next package from the stream and close the source afterwards+receiveLast+ :: MonadResource m+ => ResumableSource (Stream m) ByteString+ -> Sink BL.ByteString (Stream m) a+ -> m a+receiveLast src sink = stream_base $+ src $$+- next =$ sink++-- | Close a resumable source+close+ :: MonadResource m+ => ResumableSource (Stream m) ByteString+ -> m ()+close src = stream_base $+ src $$+- return ()++-- | Get the next package from the stream (whether it's a single 'BL.ByteString' or+-- a list)+next :: MonadResource m => Conduit ByteString (Stream m) BL.ByteString+next = do+ h <- decodeHeader+ case h of+ VarInt l -> single l+ ListSTART -> list+ EndOfInput -> return ()+ _ -> monadThrow $ UnexpectedHeader h+ where+ single l = CB.take l >>= yield++ list = do+ h <- decodeHeader+ case h of+ VarInt l -> single l >> list+ ListEND -> return ()+ _ -> monadThrow $ UnexpectedHeader h++-- | Send multiple sinks in the same list+sinkList'+ :: (Monad m, Sendable a m)+ => AppData m+ -> (Sink a (Stream m) () -> Sink b (Stream m) c)+ -> Sink b (Stream m) c+sinkList' ad f = do+ sinkListStart ad+ b <- f (sinkListElems ad)+ sinkListEnd ad+ return b++sinkListStart, sinkListEnd+ :: Monad m => AppData m -> Sink a (Stream m) ()+sinkListStart ad = yield (BS.pack listStart) =$ transPipe lift (appSink ad)+sinkListEnd ad = yield (BS.pack listEnd) =$ transPipe lift (appSink ad)++sinkListElems+ :: (Monad m, Sendable a m) => AppData m -> Sink a (Stream m) ()+sinkListElems ad = encode =$ transPipe lift (appSink ad)
+ src/Data/Conduit/Network/Stream/Exceptions.hs view
@@ -0,0 +1,17 @@+{-# LANGUAGE DeriveDataTypeable #-}++module Data.Conduit.Network.Stream.Exceptions+ ( StreamException (..)+ , Header (..)+ ) where++import Control.Exception (Exception)+import Data.Typeable++import Data.Conduit.Network.Stream.Header++data StreamException+ = UnexpectedHeader Header+ deriving (Show, Typeable)++instance Exception StreamException
+ src/Data/Conduit/Network/Stream/Header.hs view
@@ -0,0 +1,133 @@+-- | The network stream header design is inspired by the variable length+-- integers used in Googles Protocol Buffers (protobuf): Each stream block is+-- preceded by a binary header of variable length. There are currently 3 types+-- of headers:+--+-- * The beginning of a list is encoded as @[0, MSB]@+-- * The end of a list is encoded as @[1, MSB]@+-- * Length definition of the current block, encoded as variable length integer+--+-- Each header byte consists of 7 bits + the "most significant bit". For+-- example, the number 256 in its binary representation is:+--+-- > 1 0000 0000+--+-- First, split the number into 7-bit packages (add 0s to the front if+-- necessary):+--+-- > 0000 010 0000 000+--+-- These 7-bit packages are then reversed in order and encoded as 8-bit bytes+-- where the last bit marks the end of our header if set, for example:+--+-- > 0000 000 0000 010 -- reverse order+-- > 0000 0000 0000 0101 -- set 8th bit+-- > _ X -- (unset = _, set = X):+--+-- Another example would be the representation of the number 65536, or+--+-- > 1 0000 0000 0000 0000+--+-- as+--+-- > 0000 100 0000 000 0000 000 -- 7 bit packages+-- > 0000 000 0000 000 0000 100 -- reverse order+-- > 0000 0000 0000 0000 0000 1001 -- set 8th bit+-- > _ _ X+--+-- Distinction between sepcial headers (such as the start or end of lists) and+-- regular variable length integers is done by checking the most significant+-- (i.e. last) byte. In a variable length integer, the first 7 bits of the MSB+-- are always bigger than 0, while special headers are ended by the byte+-- "00000001":+--+-- > 0000 0000 0000 0001 -- special header: list start+-- > 0000 0010 0000 0001 -- special header: list end+-- > 0000 0011 -- block of length 1+-- > 0000 0101 -- block of length 2+-- > 0000 1001 -- block of length 4+-- > 1111 1111 -- block of length 127+-- > 0000 0000 0000 0011 -- block of length 128 (note the 7 bit shift)+-- > 0000 0010 0000 0011 -- block of length 129++module Data.Conduit.Network.Stream.Header where++import Data.Bits+--import Data.Enum+import Data.Word+import Data.Conduit+-- import Data.Conduit.Network.Stream.Exceptions++import qualified Data.Conduit.Binary as CB+import qualified Data.ByteString as BS+--import qualified Data.ByteString.Lazy as BL++data Header+ = ListSTART+ | ListEND+ | VarInt Int+ | InvalidHeader [Word8]+ | EndOfInput+ deriving (Show)++-- | Make the current \"7-bit byte\" the most significant byte in the header+mkMSB :: Word8 -> Word8+mkMSB = setBit `flip` 7++specialHeaderMSB :: Word8+specialHeaderMSB = mkMSB 0++-- | Test wether or not a byte is the most significant byte of a special header+-- (i.e. 8th bit = 1, rest = 0)+isSpecialHeaderMSB :: Word8 -> Bool+isSpecialHeaderMSB = (specialHeaderMSB ==)++-- | Test wether or not a byte is the most significant byte (i.e. the last byte+-- of a header block)+isMSB :: Word8 -> Bool+isMSB = testBit `flip` 7++varint :: (Show int, Integral int) => int -> [Word8]+varint int = go (fromIntegral int :: Integer) []+ where+ go i l =+ let w8 = fromIntegral $ 127 .&. i -- take the first 7 bits+ r = shiftR i 7 -- shift the rest of the bits to the right+ in if r == 0+ then l ++ [mkMSB w8]+ else go r (l ++ [w8])++fromVarint :: (Show int, Integral int, Bits int) => [Word8] -> int+fromVarint [] = 0+fromVarint [x] = fromIntegral $ x `clearBit` 7+fromVarint (w:r) = fromIntegral w + shiftL (fromVarint r) 7++listStart, listEnd :: [Word8]+listStart = [0, mkMSB 0]+listEnd = [1, mkMSB 0]++-- | A decode 'ByteString' sink which returns the current header+decodeHeader :: Monad m => Consumer BS.ByteString m Header+decodeHeader = go []+ where+ go w8s = do+ h <- CB.head+ case h of+ Nothing -> return EndOfInput+ Just w8 | isSpecialHeaderMSB w8 -> spec w8s+ | isMSB w8 -> var (w8s ++ [w8])+ | otherwise -> go (w8s ++ [w8])++ -- special header decoding+ spec [0] = return ListSTART+ spec [1] = return ListEND+ spec w8s = return $ InvalidHeader w8s++ -- var int decoding+ var vi = return $ VarInt (fromVarint vi)++encodeHeader :: Header -> Maybe BS.ByteString+encodeHeader ListSTART = Just $ BS.pack listStart+encodeHeader ListEND = Just $ BS.pack listEnd+encodeHeader (VarInt vi) = Just $ BS.pack (varint vi)+encodeHeader _ = Nothing
+ src/Data/Conduit/Network/Stream/Internal.hs view
@@ -0,0 +1,42 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++module Data.Conduit.Network.Stream.Internal where++import Control.Applicative+import Control.Monad.Trans+import Control.Monad.Trans.Resource+import Data.Conduit+import Data.ByteString (ByteString)++import qualified Data.ByteString as BS+import qualified Data.ByteString.Lazy as BL++import Data.Conduit.Network.Stream.Header++-- | 'BL.ByteString' stream+newtype Stream m a = Stream { stream_base :: m a }+ deriving (Monad, MonadIO, Functor, Applicative)++instance MonadTrans Stream where+ lift f = Stream f++instance (MonadThrow m) => MonadThrow (Stream m) where+ monadThrow e = lift $ monadThrow e++instance (MonadResource m, MonadIO m) => MonadResource (Stream m) where+ liftResourceT t = lift $ liftResourceT t++encodeBS :: Monad m => Conduit ByteString (Stream m) ByteString +encodeBS = awaitForever $ \bs -> do+ yield $ BS.pack (varint $ BS.length bs)+ mapM_ yield $ blocks bs+ where+ blocks bs | BS.null bs = []+ | otherwise =+ let (f,r) = BS.splitAt 4096 bs+ in f : blocks r++encodeLazyBS :: Monad m => Conduit (Int, BL.ByteString) (Stream m) ByteString+encodeLazyBS = awaitForever $ \(l,bs) -> do+ yield $ BS.pack (varint l)+ mapM_ yield $ BL.toChunks bs