diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/examples/addr/Main.hs b/examples/addr/Main.hs new file mode 100644 --- /dev/null +++ b/examples/addr/Main.hs @@ -0,0 +1,173 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE LambdaCase #-} + +module Main (main) where + +import Prelude hiding (log) +import Control.Applicative ((<$>), (*>), (<*>)) +import Control.Monad (void, unless, forever) +import Data.Monoid ((<>)) +import Control.Concurrent + ( forkIO + , myThreadId + , threadDelay + , killThread + , MVar + , newMVar + , readMVar + , modifyMVar_ + ) +import Data.Foldable (traverse_) +import Control.Exception (finally) +import GHC.Generics (Generic) +import Data.ByteString (ByteString) +import qualified Data.ByteString.Char8 as B8 +import Control.Monad.Reader (ask) +import Data.Binary (Binary(..), putWord8, getWord8) +import Data.Set (Set, union) +import qualified Data.Set as Set +import Network.Socket (SockAddr(..), iNADDR_ANY, PortNumber(..)) +import Control.Error (runMaybeT, hoistMaybe) +import Pipes (Consumer, (>->), runEffect, await, each) +import Pipes.Network.TCP (toSocket, send) +import qualified Pipes.Prelude as P +import Network.Simple.SockAddr (connectFork) +import Pipes.Network.P2P + +path1,path2 :: String +path1 = "/tmp/n1.socket" +path2 = "/tmp/n2.socket" + +addr1,addr2,addr3,addr4 :: SockAddr +addr1 = SockAddrUnix path1 +addr2 = SockAddrUnix path2 +addr3 = SockAddrInet 1234 iNADDR_ANY +addr4 = SockAddrInet 1235 iNADDR_ANY + +main :: IO () +main = do + n1 <- addrExchanger addr1 + n2 <- addrExchanger addr2 + n3 <- addrExchanger addr3 + n4 <- addrExchanger addr4 + t1 <- forkIO $ launch n1 [] + threadDelay 10000 + t2 <- forkIO $ launch n2 [addr1] + threadDelay 10000 + t3 <- forkIO $ launch n3 [addr1, addr2] + threadDelay 10000 + launch n4 [addr1] `finally` traverse_ killThread [t1,t2,t3] + where + addrExchanger :: SockAddr -> IO (Node AddrMsg) + addrExchanger addr = do + ps <- newMVar Set.empty + node 3741 addr $ Handlers outgoing + (incoming ps) + (register ps addr) + (unregister ps addr) + (handler ps) + +outgoing :: (Functor m, MonadIO m) => NodeConnT AddrMsg m (Maybe AddrMsg) +outgoing = runMaybeT $ do + NodeConn n (Connection addr _) <- ask + deliver . ME . Addr $ address n + expect . ME $ Addr addr + deliver ACK + expect ACK + deliver GETADDR + return . ADDR $ Addr addr + +incoming :: (Functor m, MonadIO m) + => MVar (Set Address) + -> NodeConnT AddrMsg m (Maybe AddrMsg) +incoming peers = runMaybeT $ do + NodeConn n _ <- ask + fetch >>= \case + ME addr@(Addr sockaddr) -> do + ps <- liftIO $ readMVar peers + if Set.notMember addr ps + then do deliver . ME . Addr $ address n + deliver ACK + expect ACK + return . ADDR $ Addr sockaddr + else hoistMaybe Nothing + _ -> hoistMaybe Nothing + +register :: MonadIO m + => MVar (Set Address) + -> SockAddr + -> AddrMsg + -> m () +register peers me (ADDR addr@(Addr other)) = liftIO $ do + modifyMVar_ peers $ return . Set.insert addr + log "added: " me other +register _ _ _ = error "register: `AddrMsg` needs to be `ADDR addr`" + +unregister :: MonadIO m + => MVar (Set Address) + -> SockAddr + -> AddrMsg -> m () +unregister peers me (ADDR addr@(Addr other)) = liftIO $ do + modifyMVar_ peers $ return . Set.delete addr + log "deleted: " me other +unregister _ _ _ = error "unregister: `AddrMsg` needs to be `ADDR addr`" + +log :: (Show a, Show b) => ByteString -> a -> b -> IO () +log what addr addr' = + B8.putStrLn $ "Node " <> B8.pack (show addr) + <> ": " + <> "Address " + <> what + <> B8.pack (show addr') + +handler :: (MonadIO m, MonadCatch m) + => MVar (Set Address) + -> AddrMsg + -> Consumer (Either (Relay AddrMsg) AddrMsg) (NodeConnT AddrMsg m) r +handler peers (ADDR addr) = do + NodeConn n@Node{magic} (Connection _ sock) <- ask + forever $ await >>= \case + Right GETADDR -> do + ps <- liftIO $ Set.delete addr <$> readMVar peers + runEffect $ each (Set.elems ps) + >-> P.map (serialize magic . ADDR) + >-> toSocket sock + Right (ADDR a@(Addr a')) -> do + ps <- liftIO $ readMVar peers + unless (Set.null $ Set.fromList [a, addr] `union` ps) + (liftIO . void $ connectFork a' + $ runNodeConn n True a') + Left (Relay tid' msg) -> do + tid <- liftIO myThreadId + unless (tid' == tid) + (liftIO . send sock . serialize magic $ msg) + _ -> return () +handler _ _ = error "handler: `AddrMsg` needs to be `ADDR addr`" + +data AddrMsg = GETADDR + | ADDR Address + | ME Address + | ACK + deriving (Show, Eq, Generic) + +instance Binary AddrMsg + +newtype Address = Addr SockAddr deriving (Show, Eq, Ord) + +instance Binary Address where + put (Addr (SockAddrInet (PortNum port) host)) = + putWord8 0 *> put (port, host) + put (Addr (SockAddrInet6 (PortNum port) flow host scope)) = + putWord8 1 *> put (port, flow, host, scope) + put (Addr (SockAddrUnix str)) = putWord8 2 *> put str + + get = getWord8 >>= \case + 0 -> Addr <$> (SockAddrInet <$> PortNum <$> get <*> get) + 1 -> Addr <$> (SockAddrInet6 <$> PortNum + <$> get + <*> get + <*> get + <*> get) + _ -> Addr <$> SockAddrUnix <$> get diff --git a/pipes-p2p-examples.cabal b/pipes-p2p-examples.cabal new file mode 100644 --- /dev/null +++ b/pipes-p2p-examples.cabal @@ -0,0 +1,38 @@ +name: pipes-p2p-examples +version: 0.1 +cabal-version: >=1.10 +tested-with: GHC == 7.6.3 +build-type: Simple +homepage: https://github.com/jdnavarro/pipes-p2p-examples +bug-reports: https://github.com/jdnavarro/pipes-p2p-examples/issues +license: BSD3 +license-file: LICENSE +author: Danny Navarro +maintainer: j@dannynavarro.net +category: Network, Pipes +synopsis: Examples using pipes-p2p +description: + For now this package includes a dummy P2P network where nodes just exchange + their network addresses. This can be used as boilerplate code for other P2P + networks. Future releases may contain more examples. + +source-repository head + type: git + location: git@github.com:jdnavarro/pipes-p2p-examples.git + +executable address-exchanger + main-is: Main.hs + build-depends: base >=4.6 && <4.8, + mtl, + bytestring, + binary >=0.7, + containers, + network, + errors, + network-simple-sockaddr, + pipes, + pipes-network, + pipes-p2p + hs-source-dirs: examples/addr + default-language: Haskell2010 + ghc-options: -Wall