zre 0.1.0.0 → 0.1.0.1
raw patch · 7 files changed
+185/−78 lines, 7 filesdep +config-inidep +data-defaultdep +repline
Dependencies added: config-ini, data-default, repline, text
Files
- app/Cat.hs +1/−4
- app/Main.hs +37/−11
- src/Network/ZRE.hs +24/−61
- src/Network/ZRE/Config.hs +48/−0
- src/Network/ZRE/Options.hs +57/−0
- src/Network/ZRE/Types.hs +9/−0
- zre.cabal +9/−2
app/Cat.hs view
@@ -18,10 +18,7 @@ args <- getArgs let group = B.pack $ head args :: B.ByteString - -- XXX: FIXME: doesn't uses global option parser,- -- we should probably have 'zre cat' and only keep this in examples- -- not as an app- runZre' defaultConf $ groupCat group+ runZre $ groupCat group groupCat :: Group -> ZRE () groupCat group = do
app/Main.hs view
@@ -3,7 +3,7 @@ {-# OPTIONS_GHC -fno-warn-unused-matches #-} module Main where -import Control.Monad (forever)+import Control.Monad (forever, void) import Control.Monad.IO.Class import Control.Concurrent.Async.Lifted @@ -11,13 +11,35 @@ import Network.ZRE import Network.ZRE.Parse+import Network.ZRE.Types +import Control.Concurrent.STM+import System.Console.Repline+import Data.List (isPrefixOf)++type Repl a = HaskelineT IO a++completer :: Monad m => WordCompleter m+completer n = do+ let names = [ "join"+ , "leave"+ , "shout"+ , "whisper"+ , "debug"+ , "nodebug"+ , "quit" ]++ return $ filter (isPrefixOf n) (map ('/':) names)++ini :: Repl ()+ini = liftIO $ putStrLn "Welcome!"+ main :: IO ()-main = runZre chatApp+main = runZre replApp -chatApp :: ZRE (a, b)-chatApp = do- recv `concurrently` act+replApp :: ZRE ()+replApp = void $ do+ recv `concurrently` repl where recv = forever $ do evt <- readZ@@ -31,12 +53,16 @@ Whisper uuid content _time -> put ["Whisper from", toASCIIBytes uuid, B.concat content] x -> liftIO $ print x - act = forever $ do- liftIO $ B.putStr " >"- msg <- fmap B.pack $ liftIO getLine- case parseAttoApi msg of- (Left err) -> liftIO $ B.putStr $ B.pack err- (Right cmd) -> writeZ cmd+ repl = do+ q <- getApiQueue+ liftIO $ evalRepl ">>> " (cmd q) [] (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 () put = liftIO . B.putStrLn . (B.intercalate " ")
src/Network/ZRE.hs view
@@ -2,7 +2,8 @@ {-# LANGUAGE RecordWildCards #-} module Network.ZRE ( runZre- , runZre'+ , runZreCfg+ , runZreOpts , readZ , writeZ , unReadZ@@ -20,7 +21,9 @@ , znodebug , zquit , pEndpoint- , toASCIIBytes) where+ , toASCIIBytes+ , getApiQueue+ , getEventQueue) where import Prelude hiding (putStrLn, take) import Control.Monad hiding (join)@@ -38,69 +41,25 @@ import qualified Data.ZRE as Z import Network.ZRE.Beacon-import Network.ZRE.Utils+import Network.ZRE.Config+import Network.ZRE.Options import Network.ZRE.Peer-import Network.ZRE.ZMQ import Network.ZRE.Types-import System.ZMQ4.Endpoint+import Network.ZRE.Utils+import Network.ZRE.ZMQ import Network.ZGossip+import System.ZMQ4.Endpoint import Options.Applicative import Data.Semigroup ((<>)) -parseCfg :: Parser ZRECfg-parseCfg = ZRECfg- <$> (B.pack <$> strOption- (long "name"- <> short 'n'- <> value ""- <> help "Node name"))- <*> (isec <$> option auto- (long "quiet-period"- <> short 'q'- <> metavar "N"- <> value (sec (1.0 :: Float))- <> help "Ping peer after N seconds"))- <*> ((*100000) <$> option auto- (long "dead-period"- <> short 'd'- <> metavar "N"- <> value (sec (1.0 :: Float))- <> help "Mark peer dead after N seconds"))- <*> ((*100000) <$> option auto- (long "beacon-period"- <> short 'b'- <> metavar "N"- <> value (sec (0.9 :: Float))- <> help "Send beacon every N seconds"))- <*> ((map B.pack) <$> many (strOption- (long "iface"- <> short 'i'- <> metavar "IFACE"- <> help "Interfaces")))- <*> option (attoReadM parseAttoUDPEndpoint)- (long "mcast"- <> short 'm'- <> metavar "IP:PORT"- <> value defMCastEndpoint- <> help "IP:PORT of the multicast group")- <*> optional (option (attoReadM parseAttoTCPEndpoint)- (long "gossip"- <> short 'g'- <> metavar "IP:PORT"- <> help "IP:PORT of the gossip server"))--attoReadM :: (B.ByteString -> Either String a) -> ReadM a-attoReadM p = eitherReader (p . B.pack)--runZre :: ZRE a -> IO ()-runZre app = do+runZreOpts :: ZRE a -> IO ()+runZreOpts app = do cfg <- execParser opts- print cfg- runZre' cfg app+ runZreCfg cfg app where- opts = info (parseCfg <**> helper)+ opts = info (parseOptions <**> helper) ( fullDesc <> progDesc "ZRE" <> header "zre tools" )@@ -125,13 +84,17 @@ -> (B.ByteString, B.ByteString, a) -> IO () runIface s port (iface, ipv4, ipv6) = do- print ["Bind to", bshow port, bshow iface, bshow ipv4, bshow ipv6] r <- async $ zreRouter (newTCPEndpoint ipv4 port) (inbox s) atomically $ modifyTVar s $ \x -> x { zreIfaces = M.insert iface [r] (zreIfaces x) } -runZre' :: ZRECfg -> ZRE a -> IO ()-runZre' ZRECfg{..} app = do+runZre :: ZRE a -> IO ()+runZre a = do+ cfg <- envZRECfg+ runZreCfg cfg a++runZreCfg :: ZRECfg -> ZRE a -> IO ()+runZreCfg ZRECfg{..} app = do ifcs <- getIfaces zreInterfaces u <- maybeM (exitFail "Unable to get UUID") return nextUUID@@ -144,11 +107,12 @@ let zreEndpoint = newTCPEndpoint ipv4 zrePort print zreEndpoint+ B.putStrLn $ "Starting with " <> (bshow zreEndpoint) zreName <- getName zreNamed - inQ <- atomically $ newTBQueue 1000- outQ <- atomically $ newTBQueue 1000+ inQ <- atomically $ newTBQueue 10000+ outQ <- atomically $ newTBQueue 10000 s <- newZREState zreName zreEndpoint u inQ outQ @@ -166,7 +130,6 @@ mapM_ (runIface s zrePort) ifaces wait apiAsync- --wait userAppAsync return () api :: TVar ZREState -> IO ()
+ src/Network/ZRE/Config.hs view
@@ -0,0 +1,48 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+module Network.ZRE.Config where++import System.Environment+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++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)+ return $ ZRECfg {..}++parseZRECfg :: FilePath -> IO (Either String ZRECfg)+parseZRECfg fpath = do+ rs <- TIO.readFile fpath+ return $ parseIniFile rs iniParser++-- If ZRECFG env var is set, try parsing config file it is pointing to,+-- return default config otherwise.+--+-- 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"+ mname <- lookupEnv "ZRENAME"+ case menv of+ Nothing -> return $ maybe (def) (\x -> def { zreNamed = B.pack x }) mname+ Just env -> do+ res <- parseZRECfg env+ case res of+ Left err -> putStrLn ("Unable to parse config: " ++ err) >> exitFailure+ Right cfg -> return cfg
+ src/Network/ZRE/Options.hs view
@@ -0,0 +1,57 @@+{-# LANGUAGE OverloadedStrings #-}++module Network.ZRE.Options (+ parseOptions+ ) where++import Options.Applicative+import Data.Semigroup ((<>))+import qualified Data.ByteString.Char8 as B++import Network.ZRE.Types+import System.ZMQ4.Endpoint++parseOptions :: Parser ZRECfg+parseOptions = ZRECfg+ <$> (B.pack <$> strOption+ (long "name"+ <> short 'n'+ <> value ""+ <> help "Node name"))+ <*> (isec <$> option auto+ (long "quiet-period"+ <> short 'q'+ <> metavar "N"+ <> value (sec (1.0 :: Float))+ <> help "Ping peer after N seconds"))+ <*> ((*100000) <$> option auto+ (long "dead-period"+ <> short 'd'+ <> metavar "N"+ <> value (sec (1.0 :: Float))+ <> help "Mark peer dead after N seconds"))+ <*> ((*100000) <$> option auto+ (long "beacon-period"+ <> short 'b'+ <> metavar "N"+ <> value (sec (0.9 :: Float))+ <> help "Send beacon every N seconds"))+ <*> ((map B.pack) <$> many (strOption+ (long "iface"+ <> short 'i'+ <> metavar "IFACE"+ <> help "Interfaces")))+ <*> option (attoReadM parseAttoUDPEndpoint)+ (long "mcast"+ <> short 'm'+ <> metavar "IP:PORT"+ <> value defMCastEndpoint+ <> help "IP:PORT of the multicast group")+ <*> optional (option (attoReadM parseAttoTCPEndpoint)+ (long "gossip"+ <> short 'g'+ <> metavar "IP:PORT"+ <> help "IP:PORT of the gossip server"))++attoReadM :: (B.ByteString -> Either String a) -> ReadM a+attoReadM p = eitherReader (p . B.pack)
src/Network/ZRE/Types.hs view
@@ -16,6 +16,7 @@ import qualified Data.Set as S import qualified Data.ByteString.Char8 as B import Data.Time.Clock+import Data.Default import Data.ZRE hiding (Shout, Whisper) -- (Name, Seq, Group, Groups, GroupSeq, Headers, Content, ZRECmd, ZREMsg) import System.ZMQ4.Endpoint@@ -76,6 +77,9 @@ , zreMCast = defMCastEndpoint } +instance Default ZRECfg where+ def = defaultConf+ data Event = New UUID (Maybe Name) Groups Headers Endpoint | Ready UUID Name Groups Headers Endpoint@@ -172,6 +176,11 @@ writeZ x = do (_, a) <- ask liftIO $ atomically $ writeTBQueue a x++getEventQueue :: ZRE (EventQueue)+getEventQueue = ask >>= return . fst+getApiQueue :: ZRE (APIQueue)+getApiQueue = ask >>= return . snd zjoin :: Group -> ZRE () zjoin = writeZ . DoJoin
zre.cabal view
@@ -1,8 +1,8 @@ name: zre-version: 0.1.0.0+version: 0.1.0.1 synopsis: ZRE protocol implementation description: See README.rst---homepage: https://+homepage: https://github.com/vpsfreecz/haskell-zre/ license: BSD3 license-file: LICENSE author: Richard Marko@@ -17,6 +17,8 @@ hs-source-dirs: src exposed-modules: Network.ZRE , Network.ZRE.Beacon+ , Network.ZRE.Config+ , Network.ZRE.Options , Network.ZRE.Parse , Network.ZRE.Peer , Network.ZRE.Utils@@ -32,6 +34,7 @@ build-depends: base >= 4.7 && < 5 , async , attoparsec+ , data-default , network , network-info , network-multicast@@ -46,6 +49,8 @@ , sockaddr , process , random+ , text+ , config-ini , stm , time , uuid@@ -61,6 +66,8 @@ , bytestring , async , lifted-async+ , repline+ , stm , zre default-language: Haskell2010