pipes-p2p (empty) → 0.1
raw patch · 4 files changed
+383/−0 lines, 4 filesdep +asyncdep +basedep +binarysetup-changed
Dependencies added: async, base, binary, bytestring, errors, exceptions, mtl, network, network-simple-sockaddr, pipes, pipes-concurrency, pipes-network
Files
- LICENSE +30/−0
- Setup.hs +2/−0
- pipes-p2p.cabal +36/−0
- src/Pipes/Network/P2P.hs +315/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2014, Danny Navarro++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 Danny Navarro 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,2 @@+import Distribution.Simple+main = defaultMain
+ pipes-p2p.cabal view
@@ -0,0 +1,36 @@+name: pipes-p2p+version: 0.1+cabal-version: >=1.10+tested-with: GHC == 7.6.3+build-type: Simple+homepage: https://github.com/jdnavarro/pipes-p2p+license: BSD3+license-file: LICENSE+author: Danny Navarro+maintainer: j@dannynavarro.net+category: Network, Pipes+synopsis: P2P network nodes with pipes+description:+ Toy library to facilitate the creation of custom P2P networks using `pipes`.++source-repository head+ type: git+ location: git@github.com:jdnavarro/pipes-p2p.git++library+ exposed-modules: Pipes.Network.P2P+ hs-source-dirs: src+ default-language: Haskell2010+ ghc-options: -Wall+ build-depends: base >=4.6 && <4.8,+ async >=2,+ binary >=0.7,+ mtl >=2,+ bytestring >=0.10,+ errors >=1.4,+ exceptions >=0.3,+ pipes >=0.4,+ pipes-concurrency >=2.0,+ pipes-network >=0.6,+ network >=2.4,+ network-simple-sockaddr
+ src/Pipes/Network/P2P.hs view
@@ -0,0 +1,315 @@+{-# OPTIONS_HADDOCK ignore-exports #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}++{-| Use 'node' to create a 'Node' with your desired settings and then 'launch'+ it.+-}++module Pipes.Network.P2P+ (+ -- * Nodes and Connections+ Node(..)+ , node+ , NodeConn(..)+ , NodeConnT+ , Connection(..)+ , Handlers(..)+ , launch+ , runNodeConn+ -- * Handshaking utilities+ , deliver+ , expect+ , fetch+ -- * Messaging+ , Relay(..)+ , serialize+ -- * Re-exports+ , MonadIO+ , liftIO+ , MonadCatch+ ) where++import Control.Applicative (Applicative, (<$>))+import Control.Monad (forever, void, unless, guard)+import Data.Monoid ((<>))+import Data.Foldable (for_, forM_)+import Control.Concurrent (ThreadId, myThreadId)+import Control.Concurrent.Async (async, link)+import GHC.Generics (Generic)+import Data.ByteString (ByteString)+import qualified Data.ByteString as B+import Data.ByteString.Lazy (toStrict, fromStrict)+import Data.Binary (Binary)+import qualified Data.Binary as Binary(encode,decodeOrFail)+import Control.Monad.Reader (ReaderT(..), MonadReader, runReaderT, ask)+import Control.Error (MaybeT, hoistMaybe, hush)+import Control.Monad.Catch (MonadCatch, bracket_)+import Pipes+ ( Pipe+ , Producer+ , Consumer+ , (>->)+ , runEffect+ , yield+ , await+ , MonadIO+ , liftIO+ )+import Pipes.Core (Proxy, (>+>), request, respond)+import qualified Pipes.Prelude as P+import Pipes.Network.TCP (fromSocketN)+import qualified Pipes.Concurrent+import Pipes.Concurrent+ ( Buffer(Unbounded)+ , Output+ , Input+ , spawn+ , toOutput+ , fromInput+ , atomically+ , performGC+ )+import Network.Simple.SockAddr+ ( Socket+ , SockAddr+ , serve+ , connectFork+ , send+ , recv+ )++-- * Node and Connections++{-| A P2P node.++ The constructor is exported for pattern matching purposes. Under normal+ circumstances, you should use 'node' for 'Node' creation.+-}+data Node a = Node+ { magic :: Int+ -- ^ Magic bytes.+ , address :: SockAddr+ -- ^ Listening address.+ , handlers :: Handlers a+ -- ^ Functions to define the behavior of the 'Node'.+ , broadcaster :: Mailbox a+ -- ^ 'MailBox' to relay internal messages.+ }++-- | Smart constructor to create a 'Node'.+node :: (Functor m, Applicative m, MonadIO m, Binary a, Show a)+ => Int+ -- ^ Magic bytes.+ -> SockAddr+ -- ^ Listening address.+ -> Handlers a+ -- ^ Functions to define the behavior of the 'Node'.+ -> m (Node a)+node magic addr handlers =+ Node magic addr handlers <$> liftIO (spawn Unbounded)+{-# INLINABLE node #-}++-- | Functions to define the behavior of a 'Node'.+data Handlers a = Handlers+ { ohandshake :: HandShaker a+ -- ^ What to do for an outgoing connection handshake.+ , ihandshake :: HandShaker a+ -- ^ What to do for an incoming connection handshake.+ , onConnect :: Handler a+ -- ^ Action to perform after a connection has been established.+ , onDisconnect :: Handler a+ -- ^ Action to perform after a connection has ended.+ , msgConsumer :: forall m . (MonadIO m, MonadCatch m)+ => a -> Consumer (Either (Relay a) a) (NodeConnT a m) ()+ -- ^ This consumes incoming messages either from other connections in the+ -- node, as @'Left' ('Relay' a)@, or from the current connected socket,+ -- as @'Right' a@.+ -- This is only used after a handshake has been successful.+ }++-- | Convenient data type to put together a 'Node' and a 'Connection'.+data NodeConn a = NodeConn (Node a) (Connection)++{-| Convenient data type to put together a network address and its+ corresponding socket.+-}+data Connection = Connection SockAddr Socket++-- | Convenient wrapper for a 'ReaderT' monad containing a 'NodeConn'.+newtype NodeConnT a m r = NodeConnT+ { runNodeConnT :: ReaderT (NodeConn a) m r+ } deriving ( Functor+ , Applicative+ , Monad+ , MonadIO+ , MonadCatch+ , MonadReader (NodeConn a)+ )++-- | Launch a 'Node'.+launch :: (Functor m, Applicative m, MonadIO m, MonadCatch m, Binary a)+ => Node a+ -- ^+ -> [SockAddr]+ -- ^ Addresses to try to connect on launch.+ -> m ()+launch n@Node{address} addrs = do+ for_ addrs $ \addr -> connectFork addr $ runNodeConn n True addr+ serve address $ runNodeConn n False+{-# INLINABLE launch #-}++-- | Connect a 'Node' to the given pair of 'SockAddr', 'Socket'.+runNodeConn :: (Functor m, MonadIO m, MonadCatch m, Binary a)+ => Node a+ -- ^+ -> Bool+ -- ^ Whether this is an outgoing connection.+ -> SockAddr+ -- ^ Address to connect to.+ -> Socket+ -- ^ Socket to connect to.+ -> m ()+runNodeConn n isOut addr sock =+ runReaderT (runNodeConnT go) (NodeConn n $ Connection addr sock)+ where+ go = do NodeConn Node{handlers} _ <- ask+ (if isOut+ then ohandshake handlers+ else ihandshake handlers) >>= maybe (return ()) handle+{-# INLINABLE runNodeConn #-}++-- * Handshaking utilities++{-| Send an expected message.++ The message is automatically serialized and prepended with the magic+ bytes.+-}+deliver :: (Binary a, MonadIO m)+ => a+ -- ^ Message+ -> MaybeT (NodeConnT a m) ()+deliver msg = do NodeConn (Node{magic}) (Connection _ sock) <- ask+ liftIO . send sock $ serialize magic msg+{-# INLINABLE deliver #-}++{-| Receive a message and make sure it's the same as the expected message.++ The message is automatically deserialized and checked for the correct+ magic bytes.+-}+expect :: (MonadIO m, Binary a, Eq a)+ => a+ -- ^ Message+ -> MaybeT (NodeConnT a m) ()+expect msg = do+ msg' <- fetch+ guard $ msg == msg'+{-# INLINABLE expect #-}++{-| Fetch next message.++ The message is automatically deserialized and checked for the correct magic+ bytes. Uses the length bytes in the header to pull the exact number of+ bytes of the message.+-}+fetch :: (MonadIO m, Binary a) => MaybeT (NodeConnT a m) a+fetch = do+ NodeConn (Node{magic}) (Connection _ sock) <- ask+ headerBS <- recv sock hSize+ (Header magic' nbytes) <- hoistMaybe $ decode headerBS+ guard $ magic == magic'+ bs <- liftIO $ recv sock nbytes+ hoistMaybe $ decode bs+{-# INLINABLE fetch #-}++-- * Messaging++-- | Internal message to relay to the rest of connections in the node.+data Relay a = Relay ThreadId a deriving (Show)++-- | Serializes and prepends a 'Header' to a message.+serialize :: Binary a+ => Int -- ^ Magic bytes.+ -> a -- ^ Message.+ -> ByteString+serialize magic msg = encode (Header magic $ B.length bs) <> bs+ where+ bs = encode msg+{-# INLINE serialize #-}++-- * Internal++type Mailbox a = (Output (Relay a), Input (Relay a))+type HandShaker a = forall m . (Functor m, MonadIO m, MonadCatch m)+ => NodeConnT a m (Maybe a)+type Handler a = forall m . MonadIO m => a -> m ()+++-- | Coordinates the handlers in the 'Node'.+handle :: forall a m . (MonadIO m, MonadCatch m, Binary a)+ => a -> NodeConnT a m ()+handle msg = do+ NodeConn Node{magic, handlers, broadcaster} (Connection _ sock) <- ask+ let Handlers{onConnect, onDisconnect, msgConsumer} = handlers+ bracket_ (onConnect msg) (onDisconnect msg) $ do+ (ol, il) <- liftIO $ spawn Unbounded+ liftIO $ do+ let (obc, ibc) = broadcaster+ tid <- myThreadId+ void . atomically . Pipes.Concurrent.send obc $ Relay tid msg+ void . fmap link . async $ do+ runEffect $ (socketReader magic sock :: Producer a IO ())+ >-> P.map Right >-> toOutput ol+ performGC+ void . fmap link . async $ do+ runEffect $ fromInput ibc >-> P.map Left >-> toOutput ol+ performGC+ runEffect $ fromInput il >-> msgConsumer msg++-- ** Header++data Header = Header Int Int deriving (Show, Generic)++instance Binary Header++hSize :: Int+hSize = B.length . encode $ Header 0 0+{-# INLINE hSize #-}++-- ** Socket reader++socketReader :: (MonadIO m, Binary a) => Int -> Socket -> Producer a m ()+socketReader magic sock = fromSocketN sock >+> beheader magic >+> decoder $ ()+{-# INLINABLE socketReader #-}++decoder :: (MonadIO m, Binary a) => () -> Pipe ByteString a m ()+decoder () = forever $ do+ pbs <- await+ forM_ (decode pbs) yield+{-# INLINABLE decoder #-}++beheader :: MonadIO m => Int -> () -> Proxy Int ByteString () ByteString m ()+beheader magic () = forever $ do+ hbs <- request hSize+ case decode hbs of+ Nothing -> return ()+ Just (Header magic' nbytes) -> unless (magic /= magic')+ $ request nbytes >>= respond+{-# INLINABLE beheader #-}++-- ** Strict Binary++encode :: Binary a => a -> ByteString+encode = toStrict . Binary.encode+{-# INLINE encode #-}++decode :: Binary a => ByteString -> Maybe a+decode = fmap third . hush . Binary.decodeOrFail . fromStrict+ where+ third (_,_,x) = x+{-# INLINE decode #-}