zre 0.1.0.2 → 0.1.1.0
raw patch · 28 files changed
+777/−321 lines, 28 filesdep +cerealdep −config-ininew-component:exe:zgossip-server
Dependencies added: cereal
Dependencies removed: config-ini
Files
- CHANGELOG.md +28/−0
- README.rst +102/−0
- app/Cat.hs +21/−8
- app/Main.hs +31/−24
- app/Match.hs +21/−14
- app/Monadic.hs +1/−3
- app/Time.hs +6/−3
- app/Worker.hs +3/−3
- src/Data/ZGossip.hs +14/−14
- src/Data/ZMQParse.hs +13/−12
- src/Data/ZRE.hs +68/−23
- src/Network/ZGossip.hs +3/−2
- src/Network/ZGossip/ZMQ.hs +8/−6
- src/Network/ZRE.hs +27/−23
- src/Network/ZRE/Beacon.hs +13/−8
- src/Network/ZRE/Chan.hs +124/−0
- src/Network/ZRE/Config.hs +119/−45
- src/Network/ZRE/Lib.hs +22/−23
- src/Network/ZRE/Options.hs +11/−5
- src/Network/ZRE/Parse.hs +9/−8
- src/Network/ZRE/Peer.hs +13/−10
- src/Network/ZRE/Types.hs +44/−55
- src/Network/ZRE/Utils.hs +14/−11
- src/Network/ZRE/ZMQ.hs +3/−2
- src/System/ZMQ4/Endpoint.hs +25/−15
- test/Arbitrary.hs +3/−0
- zre.cabal +22/−4
- zre.conf +9/−0
+ CHANGELOG.md view
@@ -0,0 +1,28 @@+# Version [0.1.1.0](https://github.com/sorki/haskell-zre/compare/0.1.0.2...0.1.1.0) (2020-06-17)++* Changelog started. Previous release was `0.1.0.2`.++* Breaking changes:+ * `zgossip_server` executable is renamed to `zgossip-server`+ * Removed `runZreOpts` in favor of `runZreParse` which now accepts another `optparse-applicative` parser+ * Groups now use `newtype Group` instead of plain String(s)+ and need to be constructed with `mkGroup`+ * `zrecvShouts` now requires `Group` and filters the message for this group only+ * Command line options changed:+ * `iface` renamed to `interface`+ * `mcast` renamed to `multicast-group`+ * Added `quiet-ping-rate`+ * No longer ships with `stack.yaml`++* Additions+ * Experimental `Network.ZRE.Chan` for working with typed channels++* Minor:+ * Spots new `examples` flag for building with examples (disabled by default)++---++`zre` uses [PVP Versioning][1].++[1]: https://pvp.haskell.org+
+ README.rst view
@@ -0,0 +1,102 @@+zre+===++ZRE protocol implementation https://rfc.zeromq.org/spec:36/ZRE/++Peer-to-peer local area networking with reliable group messaging+and automatic peer discovery.++Usage+-----++Dependencies::++ zeromq4++Clone and test::++ git clone https://github.com/sorki/haskell-zre/+ cd haskell-zre+ nix-build+ ./result/bin/zre+ # in another terminal or networked computer+ ./result/bin/zre++Two zre peers should find each other and be able to send message between each other.+Firewall needs to allow traffic on UDP port 5670 and TCP port range 41000-41100.+Application picks random port from this range and advertises it to network.++Applications+------------++Few applications are provided to get you started:++ - zre - interact and dump events+ - zrecat <group> - cat messages for group++These can be installed locally using `pkgs.haskellPackage.zre`.++Try running multiple copies of `zre` and `zrecat` on+the same computer or computers on the local network::++ zre+ # another terminal+ zrecat test+ # now in original terminal you can join testgroup with+ > /join test+ # or send messages to it+ > /shout test msg++Send uptime periodically to uptime group::++ ( while true; do uptime; sleep 1; done ) | zrecat uptime+++Cat file to group::++ cat /etc/os-release | zrecat test++Interact manually::++ zre+ # in zre shell following commands are supported:+ > /join time+ > /shout time test!+ > /leave time+ > /join uptime+ > /whisper <uuid> message++ZGossip+-------++Implementation of gossip protocol is included in form of key value TTL server.+This allows connecting peers from different networks (or subnets) not reachable via multicast+beacon. This service requires TCP port 31337 and can be started with `zgossip_server` binary.++Run server::++ zgossip_server++Pass gossip endpoint to apps with::++ zre -g <gossip_ip>:31337++Configuration+-------------++ZRE applications using `runZre` will automatically try to load configuration+file if `ZRECFG` environment variable points to it. See `zre.conf` for configuration+example::++ ZRECFG=./zre.conf zrecat test++To be able to use one config for multiple apps and still be able to distinguish between+them you can also set `ZRENAME` environment variable which overrides name+from config or default config if `ZRECFG` is not used::++ ZRENAME=zrenode1 zrecat test++Demos+-----++* https://asciinema.org/a/106340
app/Cat.hs view
@@ -1,4 +1,5 @@ {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-} module Main where import Control.Monad@@ -12,19 +13,31 @@ import System.Environment import Network.ZRE+import Options.Applicative +data CatOpts = CatOpts {+ shutdown :: Bool -- ^ Close connection after EOF on the input+ , lineBuffered :: Bool+ , bufferSize :: Int+ , catGroup :: Group+ } deriving (Eq, Show, Ord)++parseCatOptions = CatOpts+ <$> switch (long "shutdown" <> short 'N')+ <*> switch (long "bufbyline" <> short 'l')+ <*> option auto (long "bufsize" <> short 'O' <> value (1024*128))+ <*> (mkGroup <$> strArgument (metavar "GROUP"))+ main :: IO () main = do- args <- getArgs- let group = B.pack $ head args :: B.ByteString-- runZre $ groupCat group+ runZreParse parseCatOptions $ groupCat -groupCat :: Group -> ZRE ()-groupCat group = do- zjoin group+groupCat :: CatOpts -> ZRE ()+groupCat CatOpts{..} = do+ zjoin catGroup+ void $ async $ catln -- wait a sec so join is received by peers before sending stuff- void $ async $ (liftIO $ threadDelay 1000000) >> (stdin' group)+ void $ async $ (liftIO $ threadDelay 1000000) >> (stdin' catGroup) cat catln :: ZRE ()
app/Main.hs view
@@ -39,30 +39,37 @@ replApp :: ZRE () replApp = void $ do- recv `concurrently` repl- where- recv = forever $ do- evt <- readZ- case evt of- New uuid mname groups headers endpoint -> put ["New peer", toASCIIBytes uuid, pEndpoint endpoint]- Ready uuid name groups headers endpoint -> put ["Ready peer", name, toASCIIBytes uuid]- Quit uuid mname -> put ["Peer quit", toASCIIBytes uuid]- GroupJoin uuid group -> put ["Join group", group, toASCIIBytes uuid]- GroupLeave uuid group -> put ["Leave group", group, toASCIIBytes uuid]- Shout _uuid group content _time -> put ["Shout for group", group, ">", B.concat content]- Whisper uuid content _time -> put ["Whisper from", toASCIIBytes uuid, B.concat content]- x -> liftIO $ print x+ recv `concurrently` repl+ where+ recv = forever $ do+ evt <- readZ+ case evt of+ New uuid mname groups headers endpoint ->+ put ["New peer", toASCIIBytes uuid, pEndpoint endpoint]+ Ready uuid name groups headers endpoint ->+ put ["Ready peer", name, toASCIIBytes uuid]+ Quit uuid mname ->+ put ["Peer quit", toASCIIBytes uuid]+ GroupJoin uuid group ->+ put ["Join group", unGroup group, toASCIIBytes uuid]+ GroupLeave uuid group ->+ put ["Leave group", unGroup group, toASCIIBytes uuid]+ Shout _uuid group content _time ->+ put ["Shout for group", unGroup group, ">", B.concat content]+ Whisper uuid content _time ->+ put ["Whisper from", toASCIIBytes uuid, B.concat content]+ x -> liftIO $ print x - repl = do- q <- getApiQueue- liftIO $ evalRepl (pure ">>> ") (cmd q) [] Nothing (Word completer) ini- liftIO $ atomically $ writeTBQueue q DoQuit+ repl = do+ q <- getApiQueue+ liftIO $ evalRepl (pure ">>> ") (cmd q) [] Nothing (Word completer) ini+ liftIO $ atomically $ writeTBQueue q DoQuit - cmd :: APIQueue -> String -> Repl ()- cmd q x = do- case parseAttoApi $ B.pack x of- (Left err) -> liftIO $ B.putStrLn $ B.pack $ "Unable to parse command: " ++ err- (Right cmd) -> liftIO $ atomically $ writeTBQueue q cmd- return ()+ cmd :: APIQueue -> String -> Repl ()+ cmd q x = do+ case parseAttoApi $ B.pack x of+ (Left err) -> liftIO $ B.putStrLn $ B.pack $ "Unable to parse command: " ++ err+ (Right cmd) -> liftIO $ atomically $ writeTBQueue q cmd+ return () - put = liftIO . B.putStrLn . (B.intercalate " ")+ put = liftIO . B.putStrLn . (B.intercalate " ")
app/Match.hs view
@@ -3,37 +3,44 @@ {-# LANGUAGE FlexibleContexts #-} module Main where -import Control.Monad+import Data.ByteString (ByteString) -import qualified Data.ByteString.Char8 as B+import qualified Control.Monad+import qualified Data.ByteString.Char8 import Network.ZRE main :: IO () main = runZre app -replyGroup :: (B.ByteString -> B.ByteString) -> ZRE ()+replyGroup :: (ByteString -> ByteString) -> ZRE () replyGroup f = do m <- readZ case m of- Shout _uuid g content _time -> zshout g $ f $ B.concat content+ Shout _uuid g content _time -> zshout g+ $ f+ $ Data.ByteString.Char8.concat content+ _ -> return () echo :: ZRE () echo = replyGroup id rev :: ZRE ()-rev = replyGroup B.reverse+rev = replyGroup Data.ByteString.Char8.reverse app :: ZRE () app = do- zjoin "a"- zjoin "b"- zjoin "c"- zjoin "d"- forever $ match [- isGroupMsg "a" ==> echo- , isGroupMsg "b" ==> rev- , iff (isGroupMsg "c") echo- , decodeShouts (\x -> Right $ B.pack $ show x) (\x -> zshout "d" x)+ let gs@[a, b, c, d] = map mkGroup ["a", "b", "c", "d"]+ mapM_ zjoin gs+ Control.Monad.forever $ match [+ isGroupMsg a ==> echo+ , isGroupMsg b ==> rev+ , iff (isGroupMsg c) echo+ , decodeShouts+ (\x -> Right $ Data.ByteString.Char8.pack $ show x) -- fake decode fn+ (\x -> either+ zfail+ (zshout d) x+ ) ]
app/Monadic.hs view
@@ -20,7 +20,7 @@ -- shouts ohai vololo for group, then one vololo per second, forever ohaivololo :: ZRE () ohaivololo = do- let group = "chat"+ let group = mkGroup "chat" zjoin group zshout group "ohai" forever $ do@@ -31,6 +31,4 @@ dump = forever $ do e <- readZ case e of--- (Message ZREMsg{ msgCmd=(Shout _ content) }) -> liftIO $ B.putStrLn $ B.concat content x -> liftIO $ print x- --_ -> return ()
app/Time.hs view
@@ -9,15 +9,18 @@ import Control.Concurrent.Async.Lifted import Data.Time -import qualified Data.ByteString.Char8 as B+import qualified Data.ByteString.Char8 import Network.ZRE +timeGroup :: Group+timeGroup = mkGroup "time"+ main :: IO () main = runZre $ do void $ async $ forever $ readZ >>= liftIO . print- zjoin "time"+ zjoin timeGroup forever $ do ct <- liftIO $ getCurrentTime- zshout "time" (B.pack $ show ct)+ zshout timeGroup (Data.ByteString.Char8.pack $ show ct) liftIO $ threadDelay 1000000
app/Worker.hs view
@@ -21,7 +21,7 @@ raw = forever $ readZ >>= liftIO .print workersGroup :: Group-workersGroup = "work"+workersGroup = mkGroup "work" worker :: ZRE () worker = forever $ do@@ -29,7 +29,7 @@ case e of Whisper _uuid content _time -> do liftIO $ B.putStrLn $ B.concat content- let grp = B.concat content+ let grp = mkGroup $ B.concat content zjoin grp void $ async $ forever $ do zshout workersGroup msg@@ -44,6 +44,6 @@ e <- readZ case e of (Ready uuid _name _groups _headers _endp) -> do- zjoin "gimme"+ zjoin $ mkGroup "gimme" zwhisper uuid "gimme" x -> liftIO $ print x
src/Data/ZGossip.hs view
@@ -17,27 +17,27 @@ ) where import Prelude hiding (putStrLn, take)-import qualified Data.ByteString.Char8 as B-import qualified Data.ByteString.Lazy as BL+import Data.ByteString (ByteString) import GHC.Word import Data.ZMQParse -import Network.ZRE.Utils (bshow)-+-- | Version of the ZGossip protocol zgsVer :: Int zgsVer = 1++-- | Signature of the ZGossip protocol zgsSig :: Word16 zgsSig = 0xAAA0 -type Peer = B.ByteString-type Key = B.ByteString-type Value = B.ByteString-type TTL = Int+type Peer = ByteString+type Key = ByteString+type Value = ByteString+type TTL = Int data ZGSMsg = ZGSMsg {- zgsFrom :: Maybe B.ByteString+ zgsFrom :: Maybe ByteString , zgsCmd :: ZGSCmd } deriving (Show, Eq, Ord) @@ -59,7 +59,7 @@ newZGS :: ZGSCmd -> ZGSMsg newZGS cmd = ZGSMsg Nothing cmd -encodeZGS :: ZGSMsg -> B.ByteString+encodeZGS :: ZGSMsg -> ByteString encodeZGS ZGSMsg{..} = msg where msg = runPut $ do@@ -81,7 +81,7 @@ <*> parseLongString <*> getInt32 -parseCmd :: B.ByteString -> Get ZGSMsg+parseCmd :: ByteString -> Get ZGSMsg parseCmd from = do cmd <- (getInt8 :: Get Int) ver <- getInt8@@ -100,11 +100,11 @@ return $ ZGSMsg (Just from) zcmd -parseZGS :: [B.ByteString] -> Either String ZGSMsg+parseZGS :: [ByteString] -> Either String ZGSMsg parseZGS [from, msg] = parseZgs from msg-parseZGS x = Left "empty message"+parseZGS _ = Left "empty message" -parseZgs :: B.ByteString -> B.ByteString -> Either String ZGSMsg+parseZgs :: ByteString -> ByteString -> Either String ZGSMsg parseZgs from msg = flip runGet msg $ do sig <- getInt16 if sig /= zgsSig
src/Data/ZMQParse.hs view
@@ -31,6 +31,7 @@ where import Prelude hiding (putStrLn, take)+import Data.ByteString (ByteString) import qualified Data.ByteString.Char8 as B import qualified Data.ByteString.Lazy as BL @@ -49,66 +50,66 @@ getInt32 :: (Integral a) => Get a getInt32 = fromIntegral <$> getWord32be -parseString :: Get B.ByteString+parseString :: Get ByteString parseString = do len <- getInt8 st <- getByteString len return st -parseLongString :: Get B.ByteString+parseLongString :: Get ByteString parseLongString = do len <- getInt32 st <- getByteString len return st -parseStrings :: Get [B.ByteString]+parseStrings :: Get [ByteString] parseStrings = do count <- getInt32 res <- sequence $ replicate count parseLongString return res -parseKV :: Get (B.ByteString, B.ByteString)+parseKV :: Get (ByteString, ByteString) parseKV = do key <- parseString value <- parseLongString return (key, value) -parseMap :: Get (M.Map B.ByteString B.ByteString)+parseMap :: Get (M.Map ByteString ByteString) parseMap = do count <- getInt32 res <- sequence $ replicate count parseKV return $ M.fromList res -putByteStringLen :: B.ByteString -> PutM ()+putByteStringLen :: ByteString -> PutM () putByteStringLen x = do putInt8 $ fromIntegral $ B.length x putByteString x -putLongByteStringLen :: B.ByteString -> PutM ()+putLongByteStringLen :: ByteString -> PutM () putLongByteStringLen x = do putInt32be $ fromIntegral $ B.length x putByteString x -putByteStrings :: Foldable t => t B.ByteString -> PutM ()+putByteStrings :: Foldable t => t ByteString -> PutM () putByteStrings x = do putInt32be $ fromIntegral $ length x mapM_ putLongByteStringLen x -putKV :: (B.ByteString, B.ByteString) -> PutM ()+putKV :: (ByteString, ByteString) -> PutM () putKV (k, v) = do putByteStringLen k putLongByteStringLen v -putMap :: M.Map B.ByteString B.ByteString -> PutM ()+putMap :: M.Map ByteString ByteString -> PutM () putMap m = do putInt32be $ fromIntegral $ length ml mapM_ putKV ml where ml = M.toList m -runGet :: Get a -> B.ByteString -> Either String a+runGet :: Get a -> ByteString -> Either String a runGet g b = case Get.runGetOrFail g (BL.fromStrict b) of (Left (_unconsumed, _offset, err)) -> Left err (Right (_unconsumed, _offset, res)) -> Right res -runPut :: Put -> B.ByteString+runPut :: Put -> ByteString runPut = BL.toStrict . Put.runPut
src/Data/ZRE.hs view
@@ -1,5 +1,11 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE Rank2Types #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE AllowAmbiguousTypes #-}+ module Data.ZRE ( zreVer , newZRE@@ -11,25 +17,40 @@ , Headers , Content , Group+ , mkGroup+ , unGroup , Groups , Seq , GroupSeq , ZREMsg(..)- , ZRECmd(..)) where+ , ZRECmd(..)+ , SymbolicGroup+ , KnownGroup+ , knownToGroup+ ) where+ import Prelude hiding (putStrLn, take)+import Data.ByteString (ByteString)+ import qualified Data.ByteString.Char8 as B import qualified Data.ByteString.Lazy as BL import GHC.Word -import qualified Data.Map as M-import qualified Data.Set as S+import Data.Map (Map)+import Data.Set (Set)++import qualified Data.Set+ import Data.UUID import Data.Time.Clock import System.ZMQ4.Endpoint import Data.ZMQParse +import GHC.TypeLits+import Data.Proxy+ zreVer :: Int zreVer = 2 zreSig :: Word16@@ -37,12 +58,30 @@ type Seq = Int type GroupSeq = Int-type Group = B.ByteString-type Groups = S.Set Group-type Name = B.ByteString-type Headers = M.Map B.ByteString B.ByteString-type Content = [B.ByteString] +type SymbolicGroup = Symbol+type KnownGroup = KnownSymbol++-- | Convert from symbolic "KnownGroup" to "Group".+knownToGroup :: forall n. KnownGroup n => Group+knownToGroup = Group $ B.pack $ symbolVal @n Proxy++newtype Group = Group ByteString+ deriving (Show, Eq, Ord)++-- | Constructor for "Group"+mkGroup :: ByteString -> Group+mkGroup = Group++unGroup :: Group -> ByteString+unGroup (Group a) = a++type Groups = Set Group++type Name = ByteString+type Headers = Map ByteString ByteString+type Content = [ByteString]+ data ZREMsg = ZREMsg { msgFrom :: Maybe UUID , msgSeq :: Seq@@ -60,7 +99,7 @@ | PingOk deriving (Show, Eq, Ord) -zreBeacon :: B.ByteString -> Port -> B.ByteString+zreBeacon :: ByteString -> Port -> ByteString zreBeacon uuid port = runPut $ do putByteString "ZRE" -- XXX: for compatibility with zyre implementation@@ -80,8 +119,8 @@ Just uuid -> return uuid Nothing -> fail "Unable to parse UUID" -parseBeacon :: B.ByteString- -> (Either String (B.ByteString, Integer, UUID, Integer))+parseBeacon :: ByteString+ -> (Either String (ByteString, Integer, UUID, Integer)) parseBeacon = runGet $ do lead <- getByteString 3 ver <- getInt8@@ -106,7 +145,7 @@ newZRE :: Seq -> ZRECmd -> ZREMsg newZRE seqNum cmd = ZREMsg Nothing seqNum Nothing cmd -encodeZRE :: ZREMsg -> [B.ByteString]+encodeZRE :: ZREMsg -> [ByteString] encodeZRE ZREMsg{..} = msg:(getContent msgCmd) where msg = runPut $ do@@ -119,23 +158,26 @@ encodeCmd :: ZRECmd -> PutM () encodeCmd (Hello endpoint groups statusSeq name headers) = do putByteStringLen (pEndpoint endpoint)- putByteStrings groups+ putByteStrings $ (Data.Set.map (\(Group g) -> g)) groups putInt8 $ fromIntegral statusSeq putByteStringLen name putMap headers-encodeCmd (Shout group _content) = putByteStringLen group+encodeCmd (Shout group _content) = putGroup group encodeCmd (Join group statusSeq) = do- putByteStringLen group+ putGroup group putInt8 $ fromIntegral statusSeq encodeCmd (Leave group statusSeq) = do- putByteStringLen group+ putGroup group putInt8 $ fromIntegral statusSeq encodeCmd _ = return () +putGroup :: Group -> PutM ()+putGroup (Group g) = putByteStringLen g+ parseHello :: Get ZRECmd parseHello = Hello <$> parseEndpoint'- <*> fmap S.fromList parseStrings+ <*> (Data.Set.fromList . map Group <$> parseStrings) <*> getInt8 <*> parseString <*> parseMap@@ -146,16 +188,19 @@ (Left err) -> fail $ "Unable to parse endpoint: " ++ err (Right endpoint) -> return endpoint +parseGroup :: Get Group+parseGroup = Group <$> parseString+ parseShout :: Content -> Get ZRECmd-parseShout frames = Shout <$> parseString <*> pure frames+parseShout frames = Shout <$> parseGroup <*> pure frames parseJoin :: Get ZRECmd-parseJoin = Join <$> parseString <*> getInt8+parseJoin = Join <$> parseGroup <*> getInt8 parseLeave :: Get ZRECmd-parseLeave = Leave <$> parseString <*> getInt8+parseLeave = Leave <$> parseGroup <*> getInt8 -parseCmd :: B.ByteString -> Content -> Get ZREMsg+parseCmd :: ByteString -> Content -> Get ZREMsg parseCmd from frames = do cmd <- (getInt8 :: Get Int) ver <- getInt8@@ -180,11 +225,11 @@ return $ ZREMsg (Just uuid) sqn Nothing zcmd -parseZRE :: [B.ByteString] -> Either String ZREMsg+parseZRE :: [ByteString] -> Either String ZREMsg parseZRE (from:msg:rest) = parseZre from msg rest parseZRE _ = Left "empty message" -parseZre :: B.ByteString -> B.ByteString -> Content -> Either String ZREMsg+parseZre :: ByteString -> ByteString -> Content -> Either String ZREMsg parseZre from msg frames = flip runGet msg $ do sig <- getInt16 if sig /= zreSig
src/Network/ZGossip.hs view
@@ -13,6 +13,7 @@ import Control.Concurrent.Async import Control.Concurrent.STM +import Data.ByteString (ByteString) import Data.UUID import qualified Data.Map as M import qualified Data.Set as S@@ -76,10 +77,10 @@ serverHandle _ _ PingOk = return [] serverHandle _ _ Invalid = return [] -tryUUID :: B.ByteString -> B.ByteString+tryUUID :: ByteString -> ByteString tryUUID x = maybe x toASCIIBytes (fromByteString $ BL.fromStrict x) -dbg :: [B.ByteString] -> IO ()+dbg :: [ByteString] -> IO () dbg = B.putStrLn . (B.intercalate " ") -- send DoDiscover ZRE API messages on new Publish message
src/Network/ZGossip/ZMQ.hs view
@@ -8,6 +8,8 @@ import Control.Concurrent.Async import Control.Concurrent.STM import Control.Monad.IO.Class++import Data.ByteString (ByteString) import qualified System.ZMQ4.Monadic as ZMQ import qualified Data.ByteString.Char8 as B import qualified Data.List.NonEmpty as NE@@ -19,7 +21,7 @@ zgossipDealer :: Control.Monad.IO.Class.MonadIO m => Endpoint- -> B.ByteString+ -> ByteString -> TBQueue ZGSCmd -> (ZGSMsg -> IO ()) -> m a@@ -36,7 +38,7 @@ let spam = forever $ do cmd <- liftIO $ atomically $ readTBQueue peerQ --liftIO $ print "Spreading gossip" >> (print $ newZGS cmd)- ZMQ.sendMulti d $ (NE.fromList [encodeZGS $ newZGS cmd] :: NE.NonEmpty B.ByteString)+ ZMQ.sendMulti d $ (NE.fromList [encodeZGS $ newZGS cmd] :: NE.NonEmpty ByteString) let recv = forever $ do input <- ZMQ.receiveMulti d@@ -44,7 +46,7 @@ Left err -> do liftIO $ print $ "Malformed gossip message received: " ++ err liftIO $ print input- Right msg@ZGSMsg{..} -> do+ Right msg -> do liftIO $ handler msg sa <- ZMQ.async spam@@ -53,9 +55,9 @@ zgossipRouter :: (Foldable t, Control.Monad.IO.Class.MonadIO m) => Endpoint- -> (B.ByteString+ -> (ByteString -> ZGSCmd- -> IO (t (B.ByteString, ZGSCmd)))+ -> IO (t (ByteString, ZGSCmd))) -> m a zgossipRouter endpoint handler = ZMQ.runZMQ $ do sock <- ZMQ.socket ZMQ.Router@@ -70,5 +72,5 @@ res <- liftIO $ handler (fromJust zgsFrom) zgsCmd flip mapM_ res $ \(to, cmd) -> do --liftIO $ print "FWDing" >> print (to, cmd)- ZMQ.sendMulti sock $ (NE.fromList [to, to, encodeZGS $ newZGS $ cmd ] :: NE.NonEmpty B.ByteString)+ ZMQ.sendMulti sock $ (NE.fromList [to, to, encodeZGS $ newZGS $ cmd ] :: NE.NonEmpty ByteString)
src/Network/ZRE.hs view
@@ -3,7 +3,7 @@ module Network.ZRE ( runZre , runZreCfg- , runZreOpts+ , runZreParse , readZ , writeZ , unReadZ@@ -12,6 +12,8 @@ , Event(..) , ZRE , Z.Group+ , Z.mkGroup+ , Z.unGroup , zjoin , zleave , zshout@@ -20,6 +22,7 @@ , zdebug , znodebug , zquit+ , zfail , zrecv , pEndpoint , toASCIIBytes@@ -34,6 +37,7 @@ import Control.Concurrent.Async import Control.Concurrent.STM +import Data.ByteString (ByteString) import Data.UUID import Data.UUID.V1 import Data.Maybe@@ -56,20 +60,9 @@ import System.ZMQ4.Endpoint import Options.Applicative-import Data.Semigroup ((<>)) -runZreOpts :: ZRE a -> IO ()-runZreOpts app = do- cfg <- execParser opts- runZreCfg cfg app- where- opts = info (parseOptions <**> helper)- ( fullDesc- <> progDesc "ZRE"- <> header "zre tools" )--getIfaces :: [B.ByteString]- -> IO [(B.ByteString, B.ByteString, B.ByteString)]+getIfaces :: [ByteString]+ -> IO [(ByteString, ByteString, ByteString)] getIfaces ifcs = do case ifcs of [] -> do@@ -85,7 +78,7 @@ runIface :: Show a => TVar ZREState -> Int- -> (B.ByteString, B.ByteString, a)+ -> (ByteString, ByteString, a) -> IO () runIface s port (iface, ipv4, ipv6) = do r <- async $ zreRouter (newTCPEndpoint ipv4 port) (inbox s)@@ -93,12 +86,23 @@ x { zreIfaces = M.insert iface [r] (zreIfaces x) } runZre :: ZRE a -> IO ()-runZre a = do- cfg <- envZRECfg- runZreCfg cfg a+runZre app = runZreParse (pure ()) (\() -> app) +runZreParse :: Parser extra -> (extra -> ZRE a) -> IO ()+runZreParse parseExtra app = do+ -- try to get config from the enviornment variable ENVCFG, /etc/zre.conf+ -- or ~/.zre.conf and override with command line options.+ cfgIni <- envZRECfg "zre"+ (cfgOpts, extras) <- execParser opts+ runZreCfg (overrideNonDefault cfgIni cfgOpts) (app extras)+ where+ opts = info (((,) <$> parseOptions <*> parseExtra) <**> helper)+ ( fullDesc+ <> progDesc "ZRE"+ <> header "zre tools" )+ runZreCfg :: ZRECfg -> ZRE a -> IO ()-runZreCfg ZRECfg{..} app = do+runZreCfg cfg@ZRECfg{..} app = do ifcs <- getIfaces zreInterfaces u <- maybeM (exitFail "Unable to get UUID") return nextUUID@@ -118,7 +122,7 @@ inQ <- atomically $ newTBQueue 1000000 outQ <- atomically $ newTBQueue 1000000 - s <- newZREState zreName zreEndpoint u inQ outQ zreDbg+ s <- newZREState zreName zreEndpoint u inQ outQ zreDbg cfg -- FIXME: support multiple gossip clients case zreZGossip of@@ -239,15 +243,15 @@ Z.Shout group content -> atomically $ do emit s $ Shout from group content time- emitdbg s $ B.intercalate " " ["shout for group", group, ">", B.concat content]+ emitdbg s $ B.intercalate " " ["shout for group", Z.unGroup group, ">", B.concat content] Z.Join group groupSeq -> atomically $ do joinGroup s peer group groupSeq- emitdbg s $ B.intercalate " " ["join", group, bshow groupSeq]+ emitdbg s $ B.intercalate " " ["join", Z.unGroup group, bshow groupSeq] Z.Leave group groupSeq -> atomically $ do leaveGroup s peer group groupSeq- emitdbg s $ B.intercalate " " ["leave", group, bshow groupSeq]+ emitdbg s $ B.intercalate " " ["leave", Z.unGroup group, bshow groupSeq] Z.Ping -> atomically $ do msgPeer peer Z.PingOk
src/Network/ZRE/Beacon.hs view
@@ -1,15 +1,19 @@ {-# LANGUAGE OverloadedStrings #-}-module Network.ZRE.Beacon (beacon, beaconRecv) where+module Network.ZRE.Beacon (+ beacon+ , beaconRecv+ ) where import Control.Monad import Control.Exception import Control.Concurrent import Control.Concurrent.STM-import Network.Socket hiding (accept, send, sendTo, recv, recvFrom)+import Network.Socket import Network.Socket.ByteString import Network.SockAddr import Network.Multicast +import Data.ByteString (ByteString) import Data.Maybe import Data.UUID import Data.Time.Clock@@ -21,6 +25,7 @@ import Network.ZRE.Types import System.ZMQ4.Endpoint +-- | Receive beacons from UDP multicast beaconRecv :: TVar ZREState -> Endpoint -> IO b beaconRecv s e = do sock <- multicastReceiver (B.unpack $ endpointAddr e) (fromIntegral $ fromJust $ endpointPort e)@@ -36,10 +41,10 @@ beaconHandle s (showSockAddrBS x) uuid (fromIntegral port) _ -> return () --- handle messages received on beacon--- creates new peers--- updates peers last heard-beaconHandle :: TVar ZREState -> B.ByteString -> UUID -> Int -> IO ()+-- | Handle messages received on beacon+-- * creates new peers+-- * updates peers last heard+beaconHandle :: TVar ZREState -> ByteString -> UUID -> Int -> IO () beaconHandle s addr uuid port = do st <- atomically $ readTVar s @@ -56,8 +61,8 @@ return () --- sends udp multicast beacons-beacon :: AddrInfo -> B.ByteString -> Port -> IO a+-- | Send UDP multicast beacons periodically+beacon :: AddrInfo -> ByteString -> Port -> IO () beacon addrInfo uuid port = do withSocketsDo $ do bracket (getSocket addrInfo) close (talk (addrAddress addrInfo) (zreBeacon uuid port))
+ src/Network/ZRE/Chan.hs view
@@ -0,0 +1,124 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE AllowAmbiguousTypes #-}+module Network.ZRE.Chan (+ zreChan+ , zreChan'+ , mapToGroup+ , mapToGroup'+ ) where++import Control.Concurrent.STM (TChan)+import Data.Serialize (Serialize)+import Data.ZRE (Group, KnownGroup, knownToGroup)+import Network.ZRE (ZRE)++import qualified Control.Concurrent.Async.Lifted+import qualified Control.Concurrent.STM+import qualified Control.Monad+import qualified Control.Monad.IO.Class+import qualified Data.Serialize++import qualified Network.ZRE++-- | Map function to deserialized data type received from one group+-- and send it encoded to another group. Basically a typed proxy between+-- two groups.+mapToGroup :: forall fromGroup toGroup from to .+ ( Serialize from+ , Show from+ , Serialize to+ , KnownGroup fromGroup+ , KnownGroup toGroup+ )+ => (from -> to) -- ^ Conversion function+ -> ZRE ()+mapToGroup fn = mapToGroup'+ (knownToGroup @fromGroup)+ (knownToGroup @toGroup)+ fn++-- | Like `mapToGroup` but with non-symbolic groups+mapToGroup' :: (Show from, Serialize from, Serialize to)+ => Group -- ^ Group to listen to and decode its messages+ -> Group -- ^ Group to send encoded messages to+ -> (from -> to) -- ^ Conversion function+ -> ZRE ()+mapToGroup' fromGroup toGroup fn = do+ Network.ZRE.zjoin fromGroup+ Network.ZRE.zjoin toGroup++ Network.ZRE.zrecvShoutsDecode fromGroup Data.Serialize.decode+ $ \(mdec :: Either String from) -> do+ case mdec of+ Left e -> do+ Network.ZRE.zfail+ $ "Unable to decode message from "+ ++ show fromGroup ++ " error was: " ++ e+ Right dec -> do+ Network.ZRE.zshout toGroup $ Data.Serialize.encode $ fn dec++-- | Typed ZRE channel using two groups+--+-- * @input -> outputGroup@ for transfering encoded data+-- * @inputGroup -> output@ for receiving decoded data+--+-- Unexpected data on channel will result in error.+zreChan :: forall input output inputGroup outputGroup .+ ( Serialize input+ , Serialize output+ , KnownGroup inputGroup+ , KnownGroup outputGroup+ )+ => IO ( TChan input+ , TChan output)+zreChan = zreChan'+ (knownToGroup @outputGroup)+ (knownToGroup @inputGroup)++-- | Like `zreChan` but with non-symbolic groups+zreChan' :: (Serialize input, Serialize output)+ => Group+ -> Group+ -> IO ( TChan input+ , TChan output)+zreChan' outputGroup inputGroup = do+ chanInput <- Control.Concurrent.STM.newTChanIO+ chanOutput <- Control.Concurrent.STM.newTChanIO++ _ <- Control.Concurrent.Async.Lifted.async $ Network.ZRE.runZre $ do++ -- joining the outputGroup is not strictly needed for+ -- shouts to pass thru, for indication only+ Network.ZRE.zjoin outputGroup++ -- shout input to outputGroup+ Control.Monad.void+ $ Control.Concurrent.Async.Lifted.async+ $ Control.Monad.forever+ $ do+ out <-+ Control.Monad.IO.Class.liftIO+ $ Control.Concurrent.STM.atomically+ $ Control.Concurrent.STM.readTChan chanInput++ Network.ZRE.zshout outputGroup+ $ Data.Serialize.encode out++ -- receive on inputGroup and forward to output+ Network.ZRE.zjoin inputGroup+ Network.ZRE.zrecvShoutsDecode inputGroup Data.Serialize.decode+ $ either+ (\e -> Network.ZRE.zfail+ $ "zreChan: Unable to decode message from input "+ ++ show inputGroup+ ++ " error was: "+ ++ e+ )+ ( Control.Monad.IO.Class.liftIO+ . Control.Concurrent.STM.atomically+ . Control.Concurrent.STM.writeTChan chanOutput+ )++ return (chanInput, chanOutput)
src/Network/ZRE/Config.hs view
@@ -5,59 +5,133 @@ import System.Environment import System.Directory import System.FilePath.Posix-import System.Exit (exitFailure) import qualified Data.ByteString.Char8 as B-import qualified Data.Text as T-import qualified Data.Text.IO as TIO import Network.ZRE.Types-import System.ZMQ4.Endpoint -import Data.Ini.Config-import Data.Default+import Data.Default (def)+import qualified Data.Either+import qualified Data.Foldable -iniParser :: IniParser ZRECfg-iniParser = section "zre" $ do- zreNamed <- B.pack . T.unpack <$> fieldDef "name" (T.pack . B.unpack $ zreNamed def)- zreInterfaces <- fieldDefOf "interfaces" (return . map B.pack . words . T.unpack) []- zreQuietPeriod <- fieldDefOf "quiet-period" (fmap isec . number) (zreQuietPeriod def)- zreDeadPeriod <- fieldDefOf "dead-period" (fmap isec . number) (zreDeadPeriod def)- zreBeaconPeriod <- fieldDefOf "beacon-period" (fmap isec . number) (zreBeaconPeriod def)- zreZGossip <- fieldDefOf "gossip" (fmap Just . parseAttoTCPEndpoint . B.pack . T.unpack) (zreZGossip def)- zreMCast <- fieldDefOf "multicast-group" (parseAttoTCPEndpoint . B.pack . T.unpack) (zreMCast def)- zreDbg <- fieldDefOf "debug" flag (zreDbg def)- return $ ZRECfg {..}+import Options.Applicative+import Network.ZRE.Options -parseZRECfg :: FilePath -> IO (Either String ZRECfg)-parseZRECfg fpath = do- rs <- TIO.readFile fpath- return $ parseIniFile rs iniParser+import qualified Data.Text+import qualified Data.Attoparsec.Text --- If ZRECFG env var is set, try parsing config file it is pointing to,--- return default config otherwise.+trueStr :: Data.Attoparsec.Text.Parser Bool+trueStr = pure True <$> (+ Data.Foldable.asum+ . map Data.Attoparsec.Text.string $ [ "true", "t", "yes", "y" ]+ )++falseStr :: Data.Attoparsec.Text.Parser Bool+falseStr = pure False <$> (+ Data.Foldable.asum+ . map Data.Attoparsec.Text.string $ [ "false" , "f" , "no" , "n" ]+ )++iniFileToArgs :: [String] -> String -> [String]+iniFileToArgs sections file =+ concatMap (\(k, v) -> ["--" ++ k] ++ (if v /= "" then [v] else []))+ . map (Data.Text.unpack <$>)+ . map (\(k, v) -> if Data.Either.isRight $ Data.Attoparsec.Text.parseOnly (trueStr) v then (k, "") else (k, v)) -- fix --flag true -> --flag+ . filter (\(_, v) -> case Data.Attoparsec.Text.parseOnly (trueStr <|> falseStr) v of+ Left _e -> True+ Right b -> b)+ . map (Data.Text.pack <$>)+ . map (\x ->+ let t = takeWhile (flip elem $ '-':['a'..'z'])+ in (t x, dropWhile (== ' ') $ drop 1 $ dropWhile (/= '=') x)+ )+ . concatMap (\(_section, fields) -> fields)+ . filter (\(section, _fields) -> section `elem` sections)+ . groupBySections+ . filter (\(x:_xs) -> x /= '#') -- comments+ . filter (/="") -- empty+ $ lines file++-- transform [ "[zre]", "debug = false" "gossip=localhost:31337" "[zrecat]" "bufsize = 300"+-- to+-- [("zre", ["debug=false", "gossip=localhost:31337"]), ("zrecat", ["bufsize=300"])]+groupBySections :: [String] -> [(String, [String])]+groupBySections lines' = go lines'+ where+ go [] = []+ go ((x:xs):ls) | x == '[' = (takeWhile (flip elem $ '-':['a'..'z']) xs, keyVals ls):go ls+ go (_l:ls) | otherwise = go ls+ keyVals [] = []+ keyVals ls = takeWhile (\(x:_) -> '[' /= x) ls++-- | Override config value from new iff it differs to default value ----- if ZRENAME env var is set, it overrides name field in ZRECFG config--- or default config respectively.-envZRECfg :: IO (ZRECfg)-envZRECfg = do- menv <- lookupEnv "ZRECFG"+-- This could be done with `gzipWithT` and Generics+overrideNonDefault :: ZRECfg -> ZRECfg -> ZRECfg+overrideNonDefault orig new = ZRECfg {+ zreNamed = ovr (zreNamed orig) (zreNamed new) (zreNamed def)+ , zreQuietPeriod = ovr (zreQuietPeriod orig) (zreQuietPeriod new) (zreQuietPeriod def)+ , zreQuietPingRate = ovr (zreQuietPingRate orig) (zreQuietPingRate new) (zreQuietPingRate def)+ , zreDeadPeriod = ovr (zreDeadPeriod orig) (zreDeadPeriod new) (zreDeadPeriod def)+ , zreBeaconPeriod = ovr (zreBeaconPeriod orig) (zreBeaconPeriod new) (zreBeaconPeriod def)+ , zreInterfaces = ovr (zreInterfaces orig) (zreInterfaces new) (zreInterfaces def)+ , zreMCast = ovr (zreMCast orig) (zreMCast new) (zreMCast def)+ , zreZGossip = ovr (zreZGossip orig) (zreZGossip new) (zreZGossip def)+ , zreDbg = ovr (zreDbg orig) (zreDbg new) (zreDbg def)+ }+ where+ ovr :: (Eq a) => a -> a -> a -> a+ ovr _o n d | n /= d = n+ ovr o _n _d | otherwise = o++parseZRECfg :: String -> FilePath -> IO (Either String ZRECfg)+parseZRECfg exeName fpath = do+ isFile <- doesFileExist fpath+ case isFile of+ False -> pure $ Left "No such file"+ True -> do+ f <- readFile fpath+ let cfg = execParserPure defaultPrefs opts (iniFileToArgs ["zre", exeName] f)+ case cfg of+ -- we always fail when one of the configs fails to parse+ Failure e -> error $ fst $ renderFailure e ""+ Success cfg' -> return $ Right $ cfg'+ CompletionInvoked _ -> error "No completion"+ where+ opts = info (parseOptions <**> helper)+ ( fullDesc+ <> progDesc "ZRE"+ <> header "zre tools" )++-- The order is+-- * path of @ZRECFG@ env iff set+-- * @/etc/zre.conf@+-- * @~/.zre.conf@+-- * @default@+--+-- This also tries to parse subsection for zre programs according to their+-- name and construct correct command line for these so we can do+--+-- @[zrecat]@+-- @bufsize = 1024@+--+-- If @ZRENAME@ env var is set, it overrides name field in the result config.+envZRECfg :: String -> IO (ZRECfg)+envZRECfg exeName = do+ menv <- lookupEnv "ZRECFG" mname <- lookupEnv "ZRENAME"- case menv of- Nothing -> do- hom <- getHomeDirectory- let homPth = hom </> ".zre.conf"- tst <- doesFileExist homPth- case tst of- False -> return $ maybeUpdateName def mname- True -> do- res <- parseZRECfg homPth- case res of- Left err -> putStrLn ("Unable to parse config: " ++ err) >> exitFailure- Right cfg -> return $ maybeUpdateName cfg mname- Just env -> do- res <- parseZRECfg env- case res of- Left err -> putStrLn ("Unable to parse config: " ++ err) >> exitFailure- Right cfg -> return $ maybeUpdateName cfg mname++ hom <- getHomeDirectory++ cfg <- asumOneConfig [+ maybe (pure $ Left "No ZRECFG env") (parseZRECfg exeName) menv+ , parseZRECfg exeName "/etc/zre.conf"+ , parseZRECfg exeName $ hom </> ".zre.conf"+ , return $ Right def+ ]+ return $ maybeUpdateName cfg mname where maybeUpdateName cfg mname = maybe cfg (\x -> cfg { zreNamed = B.pack x}) mname+ asumOneConfig [] = error "Can't happen"+ asumOneConfig (x:xs) = x >>= \y -> case y of+ Left _e -> asumOneConfig xs+ Right cfg -> return $ cfg
src/Network/ZRE/Lib.hs view
@@ -2,44 +2,43 @@ import Control.Applicative import Control.Monad+import Data.ByteString (ByteString) -import Data.Either-import qualified Data.ByteString.Char8 as B+import qualified Data.ByteString.Char8 import Data.ZRE (Group) import Network.ZRE.Types -zrecvWithShout:: (B.ByteString -> ZRE ()) -> ZRE ()-zrecvWithShout f = do+zrecvWithShout :: Group+ -> (ByteString -> ZRE ())+ -> ZRE ()+zrecvWithShout group f = do e <- zrecv case e of- Shout _ _ content _time -> f (B.concat content)- _ -> return ()+ Shout _ forGroup content _time | forGroup == group ->+ f (Data.ByteString.Char8.concat content) -zrecvShouts :: (B.ByteString -> ZRE ()) -> ZRE b-zrecvShouts fn = forever $ zrecvWithShout fn+ _ | otherwise -> return () -whenDecodes :: Monad m- => (msg -> Either a decoded)- -> (decoded -> m ())- -> msg- -> m ()-whenDecodes decoder action content = case decoder content of- Left _ -> return ()- Right x -> action x+zrecvShouts :: Group+ -> (ByteString -> ZRE ())+ -> ZRE ()+zrecvShouts group fn = forever $ zrecvWithShout group fn +zrecvShoutsDecode :: Group+ -> (ByteString -> Either String decoded)+ -> (Either String decoded -> ZRE ())+ -> ZRE ()+zrecvShoutsDecode group decFn handler = zrecvShouts group $ handler . decFn+ decodeShouts :: (Monad m, Alternative m)- => (Event -> Either a decoded)- -> (decoded -> ZRE ())+ => (Event -> Either String decoded)+ -> (Either String decoded -> ZRE ()) -> Event -> m (ZRE ()) decodeShouts fn action msg = do guard $ isShout msg- let res = fn msg- guard $ isRight res- case res of- Left _ -> return $ pure ()- Right x -> return $ readZ >> action x+ return $ readZ >> (action . fn $ msg) isShout :: Event -> Bool isShout (Shout _uuid _group _content _time) = True
src/Network/ZRE/Options.hs view
@@ -5,7 +5,7 @@ ) where import Options.Applicative-import Data.Semigroup ((<>))+import Data.ByteString (ByteString) import qualified Data.ByteString.Char8 as B import Network.ZRE.Types@@ -24,6 +24,12 @@ <> metavar "N" <> value (sec (1.0 :: Float)) <> help "Ping peer after N seconds"))+ <*> (isec <$> option auto+ (long "quiet-ping-rate"+ <> short 'p'+ <> metavar "N"+ <> value (sec (1.0 :: Float))+ <> help "Peer ping rate after quiet period passed")) <*> ((*100000) <$> option auto (long "dead-period" <> short 'd'@@ -37,12 +43,12 @@ <> value (sec (0.9 :: Float)) <> help "Send beacon every N seconds")) <*> ((map B.pack) <$> many (strOption- (long "iface"+ (long "interface" <> short 'i' <> metavar "IFACE" <> help "Interfaces"))) <*> option (attoReadM parseAttoUDPEndpoint)- (long "mcast"+ (long "multicast-group" <> short 'm' <> metavar "IP:PORT" <> value defMCastEndpoint@@ -52,7 +58,7 @@ <> short 'g' <> metavar "IP:PORT" <> help "IP:PORT of the gossip server"))- <*> (flag' False (long "debug" <> short 'd'))+ <*> switch (long "debug" <> short 'd') -attoReadM :: (B.ByteString -> Either String a) -> ReadM a+attoReadM :: (ByteString -> Either String a) -> ReadM a attoReadM p = eitherReader (p . B.pack)
src/Network/ZRE/Parse.hs view
@@ -4,13 +4,14 @@ import Control.Applicative +import Data.ByteString (ByteString) import Data.UUID import Data.Attoparsec.ByteString.Char8 as A-import qualified Data.ByteString.Char8 as B +import Data.ZRE (mkGroup) import Network.ZRE.Types -parseAttoApi :: B.ByteString -> Either String API+parseAttoApi :: ByteString -> Either String API parseAttoApi = A.parseOnly parseApi parseApi :: Parser API@@ -22,24 +23,24 @@ parseCmd :: Parser API parseCmd =- DoJoin <$> (string "join" *> lskip *> word)- <|> DoLeave <$> (string "leave" *> lskip *> word)- <|> DoShout <$> (string "shout" *> lskip *> word) <*> (lskip *> word)+ DoJoin . mkGroup <$> (string "join" *> lskip *> word)+ <|> DoLeave . mkGroup <$> (string "leave" *> lskip *> word)+ <|> DoShout <$> (string "shout" *> lskip *> (mkGroup <$> word)) <*> (lskip *> word) <|> DoWhisper <$> (string "whisper" *> uuid) <*> lw <|> DoDebug <$> (string "debug" *> pure True) <|> DoDebug <$> (string "nodebug" *> pure False) <|> (string "quit" >> pure DoQuit) -lw :: Parser B.ByteString+lw :: Parser ByteString lw = lskip *> word lskip :: Parser () lskip = skipWhile (==' ') -word :: Parser B.ByteString+word :: Parser ByteString word = A.takeWhile (/=' ') ---uEOL :: Parser B.ByteString+--uEOL :: Parser ByteString --uEOL = A.takeTill (pure False) uuid :: Parser UUID
src/Network/ZRE/Peer.hs view
@@ -34,6 +34,7 @@ import qualified Data.Map as M import qualified Data.Set as Set import qualified Data.ByteString.Char8 as B+import Data.ByteString (ByteString) import Data.Time.Clock import Data.UUID import Data.ZRE()@@ -43,7 +44,7 @@ import Network.ZRE.Utils import Network.ZRE.ZMQ (zreDealer) -printPeer :: Peer -> B.ByteString+printPeer :: Peer -> ByteString printPeer Peer{..} = B.intercalate " " ["Peer", bshow peerName,@@ -177,20 +178,22 @@ pinger :: TVar ZREState -> TVar Peer -> IO b pinger s peer = forever $ do p@Peer{..} <- atomically $ readTVar peer+ cfg <- atomically $ zreCfg <$> readTVar s+ now <- getCurrentTime- if diffUTCTime now peerLastHeard > deadPeriod+ if diffUTCTime now peerLastHeard > (fromIntegral $ zreDeadPeriod cfg) then do atomically $ emitdbg s $ B.unwords ["Peer over deadPeriod, destroying", bshow p] destroyPeer s peerUUID else do let tdiff = diffUTCTime now peerLastHeard- if tdiff > quietPeriod+ if tdiff > (fromIntegral $ zreQuietPeriod cfg) then do atomically $ emitdbg s $ B.unwords ["Peer over quietPeriod, sending hugz", bshow p] atomically $ writeTBQueue peerQueue $ Ping- threadDelay quietPingRate+ threadDelay (zreQuietPingRate cfg) else do- threadDelay $ sec (quietPeriod - tdiff)+ threadDelay $ sec $ (fromIntegral $ zreQuietPeriod cfg) - tdiff lookupPeer :: TVar ZREState -> UUID -> STM (Maybe (TVar Peer)) lookupPeer s uuid = do@@ -270,7 +273,7 @@ (Just group) -> do mapM_ (flip msgPeer msg) group -shoutGroup :: TVar ZREState -> Group -> B.ByteString -> STM ()+shoutGroup :: TVar ZREState -> Group -> ByteString -> STM () shoutGroup s group msg = msgGroup s group $ Shout group [msg] shoutGroupMulti :: TVar ZREState -> Group -> Content -> STM ()@@ -282,7 +285,7 @@ msgAllLeave :: TVar ZREState -> Group -> GroupSeq -> STM () msgAllLeave s group sq = msgAll s $ Leave group sq -whisperPeerUUID :: TVar ZREState -> UUID -> B.ByteString -> STM ()+whisperPeerUUID :: TVar ZREState -> UUID -> ByteString -> STM () whisperPeerUUID s uuid msg = msgPeerUUID s uuid $ Whisper [msg] printPeers :: M.Map k (TVar Peer) -> IO ()@@ -293,9 +296,9 @@ p <- atomically $ readTVar pt B.putStrLn $ printPeer p -printGroup :: (B.ByteString, M.Map k (TVar Peer)) -> IO ()-printGroup (k,v) = do- B.putStrLn $ B.intercalate " " ["group", k, "->"]+printGroup :: (Group, M.Map k (TVar Peer)) -> IO ()+printGroup (g, v) = do+ B.putStrLn $ B.intercalate " " ["group", unGroup g, "->"] printPeers v printAll :: TVar ZREState -> IO ()
src/Network/ZRE/Types.hs view
@@ -11,10 +11,9 @@ import Control.Monad.Trans.Control import Control.Concurrent.Async import Control.Concurrent.STM+import Data.ByteString (ByteString)+import Data.Map (Map) import Data.UUID-import qualified Data.Map as M-import qualified Data.Set as S-import qualified Data.ByteString.Char8 as B import Data.Time.Clock import Data.Default @@ -33,35 +32,16 @@ zreBeaconMs :: Int zreBeaconMs = 900000 --- send hugz after x mseconds--- agressive---quietPeriod = (msec 200) / 1000000.0 :: NominalDiffTime---deadPeriod = (msec 600) / 1000000.0 :: NominalDiffTime---- lazy-quietPeriod :: NominalDiffTime-quietPeriod = (fromIntegral $ sec (1.0 :: Float)) / 1000000.0--deadPeriod :: NominalDiffTime-deadPeriod = (fromIntegral $ sec (5.0 :: Float)) / 1000000.0--quietPingRate :: Int-quietPingRate = sec (1.0 :: Float)---- send beacon every 1 ms (much aggressive, will kill networkz)---zreBeaconMs = 1000 :: Int---quietPeriod = 2000 / 100000.0 :: NominalDiffTime---deadPeriod = 6000 / 100000.0 :: NominalDiffTime- data ZRECfg = ZRECfg {- zreNamed :: B.ByteString- , zreQuietPeriod :: Int- , zreDeadPeriod :: Int- , zreBeaconPeriod :: Int- , zreInterfaces :: [B.ByteString]- , zreMCast :: Endpoint- , zreZGossip :: Maybe Endpoint- , zreDbg :: Bool+ zreNamed :: ByteString+ , zreQuietPeriod :: Int+ , zreQuietPingRate :: Int+ , zreDeadPeriod :: Int+ , zreBeaconPeriod :: Int+ , zreInterfaces :: [ByteString]+ , zreMCast :: Endpoint+ , zreZGossip :: Maybe Endpoint+ , zreDbg :: Bool } deriving (Show) defMCastEndpoint :: Endpoint@@ -69,14 +49,15 @@ defaultConf :: ZRECfg defaultConf = ZRECfg {- zreNamed = "zre"- , zreQuietPeriod = sec (1.0 :: Float)- , zreDeadPeriod = sec (5.0 :: Float)- , zreBeaconPeriod = sec (0.9 :: Float)- , zreInterfaces = []- , zreZGossip = Nothing- , zreMCast = defMCastEndpoint- , zreDbg = False+ zreNamed = "zre"+ , zreQuietPeriod = sec (1.0 :: Float)+ , zreQuietPingRate = sec (1.0 :: Float)+ , zreDeadPeriod = sec (5.0 :: Float)+ , zreBeaconPeriod = sec (0.9 :: Float)+ , zreInterfaces = []+ , zreZGossip = Nothing+ , zreMCast = defMCastEndpoint+ , zreDbg = False } instance Default ZRECfg where@@ -91,22 +72,22 @@ | Message ZREMsg | Shout UUID Group Content UTCTime | Whisper UUID Content UTCTime- | Debug B.ByteString+ | Debug ByteString deriving (Show) data API = DoJoin Group | DoLeave Group- | DoShout Group B.ByteString- | DoShoutMulti Group [B.ByteString]- | DoWhisper UUID B.ByteString+ | DoShout Group ByteString+ | DoShoutMulti Group [ByteString]+ | DoWhisper UUID ByteString | DoDiscover UUID Endpoint | DoDebug Bool | DoQuit deriving (Show) -type Peers = M.Map UUID (TVar Peer)-type PeerGroups = M.Map Group Peers+type Peers = Map UUID (TVar Peer)+type PeerGroups = Map Group Peers type EventQueue = TBQueue Event type APIQueue = TBQueue API@@ -124,7 +105,8 @@ , zreDebug :: Bool , zreIn :: EventQueue , zreOut :: APIQueue- , zreIfaces :: M.Map B.ByteString [Async ()]+ , zreIfaces :: Map ByteString [Async ()]+ , zreCfg :: ZRECfg } data Peer = Peer {@@ -193,13 +175,13 @@ zleave :: Group -> ZRE () zleave = writeZ . DoLeave -zshout :: Group -> B.ByteString -> ZRE ()+zshout :: Group -> ByteString -> ZRE () zshout group msg = writeZ $ DoShout group msg -zshout' :: Group -> [B.ByteString] -> ZRE ()+zshout' :: Group -> [ByteString] -> ZRE () zshout' group msgs = writeZ $ DoShoutMulti group msgs -zwhisper :: UUID -> B.ByteString -> ZRE ()+zwhisper :: UUID -> ByteString -> ZRE () zwhisper uuid msg = writeZ $ DoWhisper uuid msg zdebug :: ZRE ()@@ -211,6 +193,11 @@ zquit :: ZRE () zquit = writeZ $ DoQuit +zfail :: String -> ZRE ()+zfail errorMsg = do+ liftIO $ putStrLn errorMsg+ writeZ $ DoQuit+ zrecv :: ZRE (Event) zrecv = readZ @@ -223,19 +210,21 @@ -> EventQueue -> APIQueue -> Bool+ -> ZRECfg -> IO (TVar ZREState)-newZREState name endpoint u inQ outQ dbg = atomically $ newTVar $+newZREState name endpoint u inQ outQ dbg cfg = atomically $ newTVar $ ZREState { zreUUID = u- , zrePeers = M.empty- , zrePeerGroups = M.empty+ , zrePeers = mempty+ , zrePeerGroups = mempty , zreEndpoint = endpoint- , zreGroups = S.empty+ , zreGroups = mempty , zreGroupSeq = 0 , zreName = name- , zreHeaders = M.empty+ , zreHeaders = mempty , zreDebug = dbg , zreIn = inQ , zreOut = outQ- , zreIfaces = M.empty+ , zreIfaces = mempty+ , zreCfg = cfg }
src/Network/ZRE/Utils.hs view
@@ -10,9 +10,12 @@ , getName , randPort , emit- , emitdbg) where+ , emitdbg+ ) where +import Data.ByteString (ByteString)+ import System.Exit import System.Process import System.Random@@ -29,18 +32,18 @@ import qualified Data.ByteString.Char8 as B import qualified Data.ByteString.Lazy as BL -uuidByteString :: UUID -> B.ByteString+uuidByteString :: UUID -> ByteString uuidByteString = BL.toStrict . toByteString -exitFail :: B.ByteString -> IO b+exitFail :: ByteString -> IO b exitFail msg = do B.putStrLn msg exitFailure -bshow :: (Show a) => a -> B.ByteString+bshow :: (Show a) => a -> ByteString bshow = B.pack . show -getDefRoute :: IO (Maybe (B.ByteString, B.ByteString))+getDefRoute :: IO (Maybe (ByteString, ByteString)) getDefRoute = do ipr <- fmap lines $ readProcess "ip" ["route"] [] return $ listToMaybe $ catMaybes $ map getDef (map words ipr)@@ -48,24 +51,24 @@ getDef ("default":"via":gw:"dev":dev:_) = Just (B.pack gw, B.pack dev) getDef _ = Nothing -getIface :: B.ByteString -> IO (Maybe NetworkInterface)+getIface :: ByteString -> IO (Maybe NetworkInterface) getIface iname = do ns <- getNetworkInterfaces return $ listToMaybe $ filter (\x -> name x == B.unpack iname) ns -getIfaceReport :: B.ByteString- -> IO (B.ByteString, B.ByteString, B.ByteString)+getIfaceReport :: ByteString+ -> IO (ByteString, ByteString, ByteString) getIfaceReport iname = do i <- getIface iname case i of Nothing -> exitFail $ "Unable to get info for interace " `B.append` iname (Just NetworkInterface{..}) -> return (iname, B.pack $ show ipv4, B.pack $ show ipv6) -getName :: B.ByteString -> IO B.ByteString+getName :: ByteString -> IO ByteString getName "" = fmap B.pack getHostName getName x = return x -randPort :: B.ByteString -> IO Port+randPort :: ByteString -> IO Port randPort ip = loop (100 :: Int) where loop cnt = do@@ -97,7 +100,7 @@ st <- readTVar s writeTBQueue (zreIn st) x -emitdbg :: TVar ZREState -> B.ByteString -> STM ()+emitdbg :: TVar ZREState -> ByteString -> STM () emitdbg s x = do st <- readTVar s case zreDebug st of
src/Network/ZRE/ZMQ.hs view
@@ -4,6 +4,7 @@ import Control.Monad import Control.Concurrent.STM import Control.Monad.IO.Class+import Data.ByteString (ByteString) import qualified System.ZMQ4.Monadic as ZMQ import qualified Data.ByteString.Char8 as B import qualified Data.List.NonEmpty as NE@@ -14,7 +15,7 @@ zreDealer :: Control.Monad.IO.Class.MonadIO m => Endpoint- -> B.ByteString+ -> ByteString -> TBQueue ZRECmd -> m a zreDealer endpoint ourUUID peerQ = ZMQ.runZMQ $ do@@ -30,7 +31,7 @@ where loop d x = do cmd <- liftIO $ atomically $ readTBQueue peerQ -- liftIO $ print "Sending" >> (print $ newZRE x cmd)- ZMQ.sendMulti d $ (NE.fromList $ encodeZRE $ newZRE x cmd :: NE.NonEmpty B.ByteString)+ ZMQ.sendMulti d $ (NE.fromList $ encodeZRE $ newZRE x cmd :: NE.NonEmpty ByteString) loop d (x+1) zreRouter :: Control.Monad.IO.Class.MonadIO m
src/System/ZMQ4/Endpoint.hs view
@@ -19,28 +19,32 @@ , Port , Address , Transport(..)- , Endpoint(..)) where+ , Endpoint(..)+ ) where import Control.Applicative import Data.Attoparsec.ByteString.Char8 as A+import Data.ByteString (ByteString)+import Network.Socket (AddrInfo)+ import qualified Data.ByteString.Char8 as B-import Data.Char (toLower)+import qualified Data.Char -import Network.Socket -- only need addrAddress-import Network.SockAddr (showSockAddrBS)+import qualified Network.Socket+import qualified Network.SockAddr type Port = Int-type Address = B.ByteString+type Address = ByteString data Transport = TCP | UDP | IPC | InProc | PGM | EPGM deriving (Show, Eq, Ord) data Endpoint = Endpoint Transport Address (Maybe Port) deriving (Show, Eq, Ord) -pTransport :: Show a => a -> B.ByteString-pTransport x = B.pack $ map toLower $ show x+pTransport :: Show a => a -> ByteString+pTransport x = B.pack $ map Data.Char.toLower $ show x -pEndpoint :: Endpoint -> B.ByteString+pEndpoint :: Endpoint -> ByteString pEndpoint (Endpoint t a Nothing) = B.concat [pTransport t, "://" , a] pEndpoint (Endpoint t a (Just p)) = B.concat [pTransport t, "://" , a, ":", B.pack $ show p] @@ -54,13 +58,19 @@ newEndpointPort' transport addr port = Endpoint transport addr port newEndpointPortAddrInfo' :: Transport -> AddrInfo -> Maybe Port -> Endpoint-newEndpointPortAddrInfo' transport addr port = newEndpointPort' transport (showSockAddrBS $ addrAddress addr) port+newEndpointPortAddrInfo' transport addr port =+ newEndpointPort'+ transport+ (Network.SockAddr.showSockAddrBS $ Network.Socket.addrAddress addr)+ port newEndpointPort :: Transport -> Address -> Port -> Endpoint newEndpointPort transport addr port = newEndpointPort' transport addr (Just port) newEndpointPortAddrInfo :: Transport -> AddrInfo -> Port -> Endpoint-newEndpointPortAddrInfo transport addr port = newEndpointPortAddrInfo' transport addr (Just port)+newEndpointPortAddrInfo transport addr port =+ newEndpointPortAddrInfo'+ transport addr (Just port) newTCPEndpoint :: Address -> Port -> Endpoint newTCPEndpoint addr port = newEndpointPort TCP addr port@@ -72,8 +82,8 @@ newTCPEndpointAddrInfo addr port = newEndpointPortAddrInfo TCP addr port toAddrInfo :: Endpoint -> IO [AddrInfo]-toAddrInfo (Endpoint _ a (Just p)) = getAddrInfo Nothing (Just $ B.unpack a) (Just $ show p)-toAddrInfo (Endpoint _ a _) = getAddrInfo Nothing (Just $ B.unpack a) Nothing+toAddrInfo (Endpoint _ a (Just p)) = Network.Socket.getAddrInfo Nothing (Just $ B.unpack a) (Just $ show p)+toAddrInfo (Endpoint _ a _) = Network.Socket.getAddrInfo Nothing (Just $ B.unpack a) Nothing parseTransport :: Parser Transport parseTransport = do@@ -108,13 +118,13 @@ parseUDPEndpoint :: Parser Endpoint parseUDPEndpoint = Endpoint <$> pure UDP <*> parseAddress <*> optional parsePort -parseAttoEndpoint :: B.ByteString -> Either String Endpoint+parseAttoEndpoint :: ByteString -> Either String Endpoint parseAttoEndpoint = A.parseOnly parseEndpoint -parseAttoTCPEndpoint :: B.ByteString -> Either String Endpoint+parseAttoTCPEndpoint :: ByteString -> Either String Endpoint parseAttoTCPEndpoint = A.parseOnly parseTCPEndpoint -parseAttoUDPEndpoint :: B.ByteString -> Either String Endpoint+parseAttoUDPEndpoint :: ByteString -> Either String Endpoint parseAttoUDPEndpoint = A.parseOnly parseUDPEndpoint endpointAddr :: Endpoint -> Address
test/Arbitrary.hs view
@@ -38,6 +38,9 @@ <*> (fmap abs <$> arbitrary) +instance Arbitrary Group where+ arbitrary = mkGroup <$> arbitrary+ instance Arbitrary ZRECmd where arbitrary = oneof [ Hello <$> arbitrary
zre.cabal view
@@ -1,5 +1,5 @@ name: zre-version: 0.1.0.2+version: 0.1.1.0 synopsis: ZRE protocol implementation description: Peer-to-peer local area networking with reliable group messaging and automatic peer discovery.@@ -14,13 +14,23 @@ copyright: 2020 Richard Marko category: Networking build-type: Simple--- extra-source-files: cabal-version: >=1.10+extra-source-files:+ CHANGELOG.md+ README.rst+ zre.conf +flag examples+ default:+ False+ description:+ Builds example applications+ library hs-source-dirs: src exposed-modules: Network.ZRE , Network.ZRE.Beacon+ , Network.ZRE.Chan , Network.ZRE.Config , Network.ZRE.Lib , Network.ZRE.Options@@ -38,7 +48,9 @@ , System.ZMQ4.Endpoint build-depends: base >= 4.7 && < 5 , async+ , lifted-async , attoparsec+ , cereal , data-default , network , network-bsd@@ -57,7 +69,6 @@ , process , random , text- , config-ini , stm , time , uuid@@ -79,6 +90,8 @@ default-language: Haskell2010 executable mzre+ if !flag(examples)+ buildable: False hs-source-dirs: app main-is: Monadic.hs ghc-options: -threaded -rtsopts -with-rtsopts=-N@@ -91,6 +104,8 @@ default-language: Haskell2010 executable zreworker+ if !flag(examples)+ buildable: False hs-source-dirs: app main-is: Worker.hs ghc-options: -threaded -rtsopts -with-rtsopts=-N@@ -103,7 +118,7 @@ , zre default-language: Haskell2010 -executable zgossip_server+executable zgossip-server hs-source-dirs: app main-is: ZGossipSrv.hs ghc-options: -threaded -rtsopts -with-rtsopts=-N@@ -117,6 +132,8 @@ default-language: Haskell2010 executable zrematch+ if !flag(examples)+ buildable: False hs-source-dirs: app main-is: Match.hs ghc-options: -threaded -rtsopts -with-rtsopts=-N@@ -146,6 +163,7 @@ ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends: base , bytestring+ , optparse-applicative , time , lifted-async , zre
+ zre.conf view
@@ -0,0 +1,9 @@+[zre]+name = exampleZREApp+gossip = kvm-nixos:31337+#interface =+#multicast-group = IP:PORT++#quiet-period = 5 # in seconds+#dead-period = 10 # in seconds+#beacon-period = 1 # in seconds