diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -13,7 +13,7 @@ disclaimer in the documentation and/or other materials provided with the distribution. - * Neither the name of Nils Schweinsberg 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. diff --git a/conduit-network-stream.cabal b/conduit-network-stream.cabal --- a/conduit-network-stream.cabal +++ b/conduit-network-stream.cabal @@ -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 diff --git a/src/Data/Conduit/Network/Stream.hs b/src/Data/Conduit/Network/Stream.hs --- a/src/Data/Conduit/Network/Stream.hs +++ b/src/Data/Conduit/Network/Stream.hs @@ -1,21 +1,77 @@ {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE FlexibleInstances #-} +{- | + +Module : Data.Conduit.Network.Stream +Copyright : Nils Schweinsberg +License : BSD-style + +Maintainer : Nils Schweinsberg +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 diff --git a/src/Data/Conduit/Network/Stream/Exceptions.hs b/src/Data/Conduit/Network/Stream/Exceptions.hs --- a/src/Data/Conduit/Network/Stream/Exceptions.hs +++ b/src/Data/Conduit/Network/Stream/Exceptions.hs @@ -12,6 +12,7 @@ data StreamException = UnexpectedHeader Header + | ClosedStream deriving (Show, Typeable) instance Exception StreamException diff --git a/src/Data/Conduit/Network/Stream/Header.hs b/src/Data/Conduit/Network/Stream/Header.hs --- a/src/Data/Conduit/Network/Stream/Header.hs +++ b/src/Data/Conduit/Network/Stream/Header.hs @@ -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 diff --git a/src/Data/Conduit/Network/Stream/Internal.hs b/src/Data/Conduit/Network/Stream/Internal.hs --- a/src/Data/Conduit/Network/Stream/Internal.hs +++ b/src/Data/Conduit/Network/Stream/Internal.hs @@ -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