conduit-network-stream 0.1 → 0.2

raw patch · 6 files changed

+259/−130 lines, 6 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• LICENSE +1/−1
• conduit-network-stream.cabal +1/−2
• src/Data/Conduit/Network/Stream.hs +188/−99
• src/Data/Conduit/Network/Stream/Exceptions.hs +1/−0
• src/Data/Conduit/Network/Stream/Header.hs +20/−20
• src/Data/Conduit/Network/Stream/Internal.hs +48/−8

@@ -13,7 +13,7 @@       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+    * Neither the name of Nils Schweinsberg nor the names of other       contributors may be used to endorse or promote products derived       from this software without specific prior written permission. 

@@ -1,5 +1,5 @@ Name:                conduit-network-stream-Version:             0.1+Version:             0.2  Synopsis:            A base layer for network protocols using Conduits Description:         A base layer for network protocols using Conduits@@ -20,7 +20,6 @@ source-repository head   type:             git   location:         git://github.com/mcmaniac/conduit-network-stream.git-  tag:              0.1  Library 

@@ -1,21 +1,77 @@ {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE FlexibleInstances #-} +{- |++Module      :  Data.Conduit.Network.Stream+Copyright   :  Nils Schweinsberg+License     :  BSD-style++Maintainer  :  Nils Schweinsberg <mail@nils.cc>+Stability   :  experimental+++Easy to use network streaming with conduits. This library properly encodes+conduit blocks over a network connection such that++ - each `await` corresponds to exactly one `yield` and++ - each `receive` corresponds to exactly one `send`.++It also supports sending and receiving of custom data types via the+`Sendable` and `Receivable` instances.++A simple server/client example (using @-XOverloadedStrings@):++> import           Control.Monad.Trans+> import qualified Data.ByteString             as Strict+> import qualified Data.ByteString.Lazy        as Lazy+> import           Data.Conduit+> import qualified Data.Conduit.List           as CL+> import           Data.Conduit.Network+> import           Data.Conduit.Network.Stream+>+> client :: IO ()+> client = runResourceT $ runTCPClient (clientSettings ..) $ \appData -> do       +>+>     streamData <- toStreamData appData+>+>     send streamData $ mapM_ yield (["ab", "cd", "ef"] :: [Strict.ByteString])+>     send streamData $ mapM_ yield (["123", "456"]     :: [Strict.ByteString])+>+>     closeStream streamData+>+> server :: IO ()+> server = runResourceT $ runTCPServer (serverSettings ..) $ \appData -> do+>+>     streamData <- toStreamData appData+>+>     bs  <- receive streamData $ CL.consume+>     liftIO $ print (bs  :: [Lazy.ByteString])+>+>     bs' <- receive streamData $ CL.consume+>     liftIO $ print (bs' :: [Lazy.ByteString])+> +>     closeStream streamData++-}+ module Data.Conduit.Network.Stream   ( -- * Network streams-    Stream+    StreamData, toStreamData, closeStream     -- ** Sending-  , Sendable, send1, sendList+  , send+  , Sendable(..), EncodedBS     -- ** Receiving-  , Streamable (receive), receiveLast, close-    -- ** Manual sending/receiving-  , next-  , (~~)-  , sink1, sinkList, sinkList'-  --, sinkListStart, sinkListElems, sinkListEnd+  , receive+  , Receivable(..)+    -- ** Bi-directional conversations+  , streamSink+  , withElementSink   ) where -import Control.Monad.Trans+import Control.Concurrent.MVar+import Control.Monad.Reader import Control.Monad.Trans.Resource import Data.ByteString (ByteString) import Data.Conduit hiding (($$))@@ -23,7 +79,7 @@  import qualified Data.Conduit         as C  import qualified Data.Conduit.List    as CL-import qualified Data.Conduit.Binary  as CB+import qualified Data.Conduit.Internal as CI import qualified Data.ByteString      as BS import qualified Data.ByteString.Lazy as BL @@ -31,109 +87,142 @@ 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+sinkCondStart, sinkCondEnd+  :: Monad m+  => StreamData m -> Sink a m ()+sinkCondStart sd = yield (BS.pack condStart) =$ streamDataSink sd+sinkCondEnd   sd = yield (BS.pack condEnd)   =$ streamDataSink sd -class Sendable a m where-  encode :: Conduit a (Stream m) ByteString+sinkCondElems+  :: (Monad m, Sendable m a)+  => StreamData m -> Sink a m ()+sinkCondElems sd = encode =$ CL.map (\(EncodedBS bs) -> bs) =$ streamDataSink sd -instance Monad m => Sendable ByteString m where-  encode = encodeBS+toStreamData :: MonadIO n => AppData m -> n (StreamData m)+toStreamData ad = do+  src <- liftIO $ newMVar (NewSource ad)+  let sd = StreamData src (appSink ad)+  --register $ closeStream sd+  return sd -instance Monad m => Sendable (Int, BL.ByteString) m where-  encode = encodeLazyBS+-- | Close current stream. In order to guarantee process resource finalization,+-- you /must/ use this operator after using `receive`.+closeStream+  :: MonadResource m+  => StreamData m+  -> m ()+closeStream sd = do+  src <- liftIO $ takeMVar (streamDataSource sd)+  case src of+       OpenSource s -> s $$+- return ()+       _            -> return () --- | 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+--------------------------------------------------------------------------------+-- Receiving data --- | 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+-- | `decode` is used after receiving the individual conduit block elements.+-- It is therefore not necessary to reuse other `decode` instances (in+-- contrast to `Sendable` instance definitions).+class Receivable a m where+  decode :: Conduit BL.ByteString m a -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)+-- | Instance for strict bytestrings. Note that this uses `BL.toStrict` for the+-- conversion from lazy bytestrings, which is rather expensive. Try to use lazy+-- bytestrings if possible.+instance Monad m => Receivable ByteString m where+  decode = CL.map BL.toStrict -sinkList :: (Monad m, Sendable a m) => AppData m -> Sink a (Stream m) ()-sinkList ad = do-  sinkListStart ad-  sinkListElems ad-  sinkListEnd   ad+-- | For lazy bytestrings, `decode` is the identity conduit.+instance Monad m => Receivable BL.ByteString m where+  decode = CI.ConduitM $ CI.idP -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)+-- | Receive the next conduit block. Might fail with the `ClosedStream`+-- exception if used on a stream that has been closed by `closeStream`.+receive :: (MonadResource m, Receivable a m) => StreamData m -> Sink a m b -> m b+receive sd sink = do+  -- get current source (and block MVar, just in case)+  src <- liftIO $ takeMVar (streamDataSource sd)+  (next,a) <- case src of+    NewSource ad    -> appSource ad $$+  decodeCondBlock =$= decode =$ sink+    OpenSource rsrc -> rsrc         $$++ decodeCondBlock =$= decode =$  sink+    ClosedSource    -> monadThrow $ ClosedStream+  liftIO $ putMVar (streamDataSource sd) (OpenSource next)+  return a -instance MonadResource m => Streamable (AppData m) m where-  receive ad sink = stream_base $-    transPipe lift (appSource ad) $$+ next =$ sink+--------------------------------------------------------------------------------+-- Sending data -instance MonadResource m => Streamable (ResumableSource (Stream m) ByteString) m where-  receive src sink = stream_base $-    src $$++ next =$ sink+-- | Newtype for properly encoded bytestrings.+newtype EncodedBS = EncodedBS ByteString --- | 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+-- | To define your own `Sendable` instances, reuse the instances for strict and+-- lazy bytestrings, for example for "Data.Text":+--+-- > instance (Monad m, Sendable m Data.ByteString.ByteString) => Sendable m Text where+-- >     encode = Data.Conduit.List.map encodeUtf8 =$= encode+class Sendable m a where+  -- | `encode` is called before sending out conduit block elements. Each+  -- element has to be encoded either as strict `ByteString` or as lazy `BL.ByteString`+  -- with a known length.+  encode :: Conduit a m EncodedBS --- | Close a resumable source-close-  :: MonadResource m-  => ResumableSource (Stream m) ByteString-  -> m ()-close src = stream_base $-  src $$+- return ()+-- | Instance for strict bytestrings, using a specialized version of `encode`.+instance Monad m => Sendable m ByteString where+  encode = encodeBS =$= CL.map EncodedBS --- | 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+-- | Instance for lazy bytestrings with a known length, using a specialized+-- version of `encode`.+instance Monad m => Sendable m (Int, BL.ByteString) where+  encode = encodeLazyBS =$= CL.map EncodedBS -  list = do-    h <- decodeHeader-    case h of-         VarInt l -> single l >> list-         ListEND  -> return ()-         _        -> monadThrow $ UnexpectedHeader h+-- | Instance for lazy bytestrings which calculates the length of the+-- `BL.ByteString` before calling the @(Int, Data.ByteString.Lazy.ByteString)@+-- instance of `Sendable`.+instance Monad m => Sendable m BL.ByteString where+  encode = CL.map (\bs -> (len bs, bs)) =$= encode+   where+    len :: BL.ByteString -> Int+    len bs = fromIntegral $ BL.length bs --- | 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+-- | Send one conduit block.+send :: (Monad m, Sendable m a) => StreamData m -> Source m a -> m ()+send sd src = src C.$$ streamSink sd -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)+--------------------------------------------------------------------------------+-- Bi-directional conversations++-- | For bi-directional conversations you sometimes need the sink of the current+-- stream, since you can't use `send` within another `receive`.+--+-- A simple example:+--+-- > receive streamData $+-- >     myConduit =$ streamSink streamData+--+-- Note, that each `streamSink` marks its own conduit block. If you want to sink+-- single block elements, use `withElementSink` instead.+streamSink+  :: (Monad m, Sendable m a)+  => StreamData m+  -> Sink a m ()+streamSink sd = do+  sinkCondStart sd+  sinkCondElems sd+  sinkCondEnd   sd++-- | Sink single elements inside the same conduit block. Example:+--+-- > receive streamData $ withElementSink $ \sinkElem -> do+-- >     yield singleElem =$ sinkElem+-- >     mapM_ yield moreElems =$ sinkElem+withElementSink+  :: (Monad m, Sendable m a)+  => StreamData m+  -> (Sink a m () -> Sink b m c)+  -> Sink b m c+withElementSink sd run = do+  sinkCondStart sd+  res <- run (sinkCondElems sd)+  sinkCondEnd   sd+  return res

@@ -12,6 +12,7 @@  data StreamException   = UnexpectedHeader Header+  | ClosedStream   deriving (Show, Typeable)  instance Exception StreamException

@@ -3,8 +3,8 @@ -- 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]@+--  * The beginning of a conduit block is encoded as @[0, MSB]@+--  * The end of a conduit block 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@@ -35,14 +35,14 @@ -- > 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":+-- Distinction between sepcial headers (such as the start or end of conduit+-- blocks) 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 0000  0000 0001  --  special header: conduit block start+-- > 0000 0010  0000 0001  --  special header: conduit block end -- > 0000 0011             --  block of length 1 -- > 0000 0101             --  block of length 2 -- > 0000 1001             --  block of length 4@@ -63,8 +63,8 @@ --import qualified Data.ByteString.Lazy as BL  data Header-  = ListSTART-  | ListEND+  = ConduitSTART+  | ConduitEND   | VarInt Int   | InvalidHeader [Word8]   | EndOfInput@@ -102,9 +102,9 @@ 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]+condStart, condEnd :: [Word8]+condStart = [0, mkMSB 0]+condEnd   = [1, mkMSB 0]  -- | A decode 'ByteString' sink which returns the current header decodeHeader :: Monad m => Consumer BS.ByteString m Header@@ -119,15 +119,15 @@                  | otherwise             -> go   (w8s ++ [w8])    -- special header decoding-  spec [0] = return ListSTART-  spec [1] = return ListEND+  spec [0] = return ConduitSTART+  spec [1] = return ConduitEND   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+encodeHeader ConduitSTART = Just $ BS.pack condStart+encodeHeader ConduitEND   = Just $ BS.pack condEnd+encodeHeader (VarInt vi)  = Just $ BS.pack (varint vi)+encodeHeader _            = Nothing

@@ -3,30 +3,47 @@ module Data.Conduit.Network.Stream.Internal where  import Control.Applicative-import Control.Monad.Trans+import Control.Concurrent.MVar+import Control.Monad.Reader+--import Control.Monad.Trans import Control.Monad.Trans.Resource import Data.Conduit+import Data.Conduit.Network import Data.ByteString (ByteString)  import qualified Data.ByteString      as BS import qualified Data.ByteString.Lazy as BL+import qualified Data.Conduit.Binary  as CB +import Data.Conduit.Network.Stream.Exceptions import Data.Conduit.Network.Stream.Header +data StreamSource m+  = NewSource (AppData m)+  | OpenSource (ResumableSource m ByteString)+  | ClosedSource++data StreamData m = StreamData+  { streamDataSource  :: MVar (StreamSource m)+  , streamDataSink    :: Sink ByteString m () }+ -- | 'BL.ByteString' stream-newtype Stream m a = Stream { stream_base :: m a }+newtype StreamT m a = StreamT { stream_base :: ReaderT (StreamData m) m a }   deriving (Monad, MonadIO, Functor, Applicative) -instance MonadTrans Stream where-  lift f = Stream f+instance MonadTrans StreamT where+  lift f = StreamT $ lift f -instance (MonadThrow m) => MonadThrow (Stream m) where+instance (MonadThrow m) => MonadThrow (StreamT m) where   monadThrow e = lift $ monadThrow e -instance (MonadResource m, MonadIO m) => MonadResource (Stream m) where+instance (MonadResource m, MonadIO m) => MonadResource (StreamT m) where   liftResourceT t = lift $ liftResourceT t -encodeBS :: Monad m => Conduit ByteString (Stream m) ByteString +--------------------------------------------------------------------------------+-- Stream encoding++encodeBS :: Monad m => Conduit ByteString m ByteString  encodeBS = awaitForever $ \bs -> do   yield $ BS.pack (varint $ BS.length bs)   mapM_ yield $ blocks bs@@ -36,7 +53,30 @@               let (f,r) = BS.splitAt 4096 bs                in f : blocks r -encodeLazyBS :: Monad m => Conduit (Int, BL.ByteString) (Stream m) ByteString+encodeLazyBS :: Monad m => Conduit (Int, BL.ByteString) m ByteString encodeLazyBS = awaitForever $ \(l,bs) -> do   yield $ BS.pack (varint l)   mapM_ yield $ BL.toChunks bs++--------------------------------------------------------------------------------+-- Stream decoding++-- | Get the next package from the stream (whether it's a single 'BL.ByteString' or+-- a list)+decodeCondBlock :: MonadResource m => Conduit ByteString m BL.ByteString+decodeCondBlock = do+  h <- decodeHeader+  case h of+       VarInt l     -> single l+       ConduitSTART -> 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+         ConduitEND -> return ()+         _          -> monadThrow $ UnexpectedHeader h