packages feed

zre (empty) → 0.1.0.0

raw patch · 25 files changed

+2270/−0 lines, 25 filesdep +asyncdep +attoparsecdep +basesetup-changed

Dependencies added: async, attoparsec, base, binary, binary-strict, bytestring, containers, lifted-async, monad-control, mtl, network, network-info, network-multicast, optparse-applicative, process, random, sockaddr, stm, time, transformers-base, uuid, zeromq4-haskell, zre

Files

+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Author name here (c) 2016++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 Author name here 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
+ app/Cat.hs view
@@ -0,0 +1,66 @@+{-# LANGUAGE OverloadedStrings #-}+module Main where++import Control.Monad+import Control.Monad.IO.Class+import Control.Concurrent+import Control.Concurrent.Async.Lifted++import qualified Data.ByteString.Char8 as B++import System.IO+import System.Environment++import Network.ZRE++main :: IO ()+main = do+  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++groupCat :: Group -> ZRE ()+groupCat group = do+  zjoin group+  -- wait a sec so join is received by peers before sending stuff+  void $ async $ (liftIO $ threadDelay 1000000) >> (stdin' group)+  cat++catln :: ZRE ()+catln = forever $ do+  e <- readZ+  case e of+    Shout _uuid _group content _time  -> liftIO $ B.putStrLn $ B.concat content+    Whisper _uuid content _time -> liftIO $ B.putStrLn $ B.concat content+    _ -> return ()++cat :: ZRE ()+cat = forever $ do+  e <- readZ+  case e of+    Shout _uuid _group content _time  -> liftIO $ B.putStr $ B.concat content+    Whisper _uuid content _time -> liftIO $ B.putStr $ B.concat content+    _ -> return ()++stdinln :: Group -> ZRE ()+stdinln group = forever $ do+  l <- fmap B.pack $ liftIO getLine+  zshout group l++bufsize :: Int+bufsize = 1024*128++stdin' :: Group -> ZRE ()+stdin' group = do+  liftIO $ hSetBuffering stdin $ BlockBuffering (Just bufsize)+  forever $ do+    eof <- liftIO $ hIsEOF stdin+    case eof of+      True -> zquit+      False -> do+        buf <- liftIO $ B.hGetSome stdin bufsize+        zshout group buf
+ app/Main.hs view
@@ -0,0 +1,42 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# OPTIONS_GHC -fno-warn-unused-matches #-}+module Main where++import Control.Monad (forever)+import Control.Monad.IO.Class+import Control.Concurrent.Async.Lifted++import qualified Data.ByteString.Char8 as B++import Network.ZRE+import Network.ZRE.Parse++main :: IO ()+main = runZre chatApp++chatApp :: ZRE (a, b)+chatApp = do+      recv `concurrently` act+      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++        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+          return ()++        put = liftIO . B.putStrLn . (B.intercalate " ")
+ app/Match.hs view
@@ -0,0 +1,56 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE FlexibleContexts #-}+module Main where++import Control.Monad++import qualified Data.ByteString.Char8 as B++import Network.ZRE++main :: IO ()+main = runZre app++replyGroup :: (B.ByteString -> B.ByteString) -> ZRE ()+replyGroup f = do+    m <- readZ+    case m of+      Shout _uuid g content _time -> zshout g $ f $ B.concat content+      _ -> return ()++echo :: ZRE ()+echo = replyGroup id++rev :: ZRE ()+rev =  replyGroup B.reverse++app :: ZRE ()+app = do+  zjoin "a"+  zjoin "b"+  forever $ match [+      handleGroup "a" echo+    , handleGroup "b" rev+    ]++isGroupMsg :: Group -> Event -> Bool+isGroupMsg group (Shout _uuid g _content _time) = g == group+isGroupMsg _ _ = False++handleGroup :: MonadPlus m => Group -> b -> Event -> m b+handleGroup group action msg = do+  guard $ isGroupMsg group msg+  return $ action++match :: [Event -> Maybe (ZRE ())] -> ZRE ()+match acts = do+  msg <- readZ+  go acts msg+  where+    go (act:rest) m = do+      case act m of+        Nothing -> go rest m+        Just a -> unReadZ m >> a++    go [] _ = return ()
+ app/Monadic.hs view
@@ -0,0 +1,36 @@+{-# LANGUAGE OverloadedStrings #-}+module Main where++import Control.Monad (forever)+import Control.Monad.IO.Class+import Control.Concurrent+import Control.Concurrent.Async.Lifted++import Network.ZRE++main :: IO ()+main = runZre app++app :: ZRE ((), ())+app = (recv `concurrently` act)+  where+    recv = dump+    act = ohaivololo++-- shouts ohai vololo for group, then one vololo per second, forever+ohaivololo :: ZRE ()+ohaivololo = do+  let group = "chat"+  zjoin group+  zshout group "ohai"+  forever $ do+    zshout group "vololo"+    liftIO $ threadDelay 1000000++dump :: ZRE ()+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
@@ -0,0 +1,23 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE FlexibleContexts #-}+module Main where++import Control.Monad+import Control.Monad.IO.Class+import Control.Concurrent+import Control.Concurrent.Async.Lifted+import Data.Time++import qualified Data.ByteString.Char8 as B++import Network.ZRE++main :: IO ()+main = runZre $ do+  void $ async $ forever $ readZ >>= liftIO . print+  zjoin "time"+  forever $ do+    ct <- liftIO $ getCurrentTime+    zshout "time" (B.pack $ show ct)+    liftIO $ threadDelay 1000000
+ app/Worker.hs view
@@ -0,0 +1,49 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+module Main where++import Control.Monad+import Control.Monad.IO.Class+import Control.Concurrent+import Control.Concurrent.Async.Lifted++import qualified Data.ByteString.Char8 as B++import Network.ZRE++main :: IO ()+main = runZre app++app :: ZRE ()+app = worker++raw :: ZRE ()+raw = forever $ readZ >>= liftIO .print++workersGroup :: Group+workersGroup = "work"++worker :: ZRE ()+worker = forever $ do+  e <- readZ+  case e of+    Whisper _uuid content _time -> do+      liftIO $ B.putStrLn $ B.concat content+      let grp = B.concat content+      zjoin grp+      void $ async $ forever $ do+        zshout workersGroup msg+        liftIO $ threadDelay (1000000)++    x -> liftIO $ print x++  where msg = B.concat $ replicate 10 "woooork"++dealer :: ZRE ()+dealer = forever $ do+  e <- readZ+  case e of+    (Ready uuid _name _groups _headers _endp) -> do+      zjoin "gimme"+      zwhisper uuid "gimme"+    x -> liftIO $ print x
+ app/ZGossipSrv.hs view
@@ -0,0 +1,13 @@+{-# LANGUAGE OverloadedStrings #-}+module Main where++import Network.ZGossip+import System.ZMQ4.Endpoint++gossipPort :: Int+gossipPort = 31337++main :: IO ()+main = do+  let gossipServerEndpoint = newTCPEndpoint "*" gossipPort+  zgossipServer gossipServerEndpoint
+ src/Data/ZGossip.hs view
@@ -0,0 +1,117 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}++-- zgossip protocol https://github.com/zeromq/czmq/blob/master/src/zgossip_msg.bnf+-- client sends HELLO, recieves all stored tuples, forwards to other clients++module Data.ZGossip (+    newZGS+  , parseZGS+  , encodeZGS+  , Key+  , Value+  , TTL+  , Peer+  , ZGSCmd(..)+  , ZGSMsg(..)+  ) where++import Prelude hiding (putStrLn, take)+import qualified Data.ByteString.Char8 as B+import qualified Data.ByteString.Lazy as BL++import GHC.Word++import Data.Binary.Strict.Get+import Data.Binary.Put++import Data.ZMQParse++import Network.ZRE.Utils (bshow)++zgsVer :: Int+zgsVer = 1+zgsSig :: Word16+zgsSig = 0xAAA0++type Peer = B.ByteString+type Key = B.ByteString+type Value = B.ByteString+type TTL = Int++data ZGSMsg = ZGSMsg {+    zgsFrom :: Maybe B.ByteString+  , zgsCmd :: ZGSCmd+  } deriving (Show, Eq, Ord)++data ZGSCmd =+    Hello+  | Publish Key Value TTL+  | Ping+  | PingOk+  | Invalid+  deriving (Show, Eq, Ord)++cmdCode :: ZGSCmd -> Word8+cmdCode Hello           = 0x01+cmdCode (Publish _ _ _) = 0x02+cmdCode Ping            = 0x03+cmdCode PingOk          = 0x04+cmdCode Invalid         = 0x05++newZGS :: ZGSCmd -> ZGSMsg+newZGS cmd = ZGSMsg Nothing cmd++encodeZGS :: ZGSMsg -> B.ByteString+encodeZGS ZGSMsg{..} = msg+  where+    msg = BL.toStrict $ runPut $ do+      putWord16be zgsSig+      putWord8 $ cmdCode zgsCmd+      putInt8 $ fromIntegral zgsVer+      encodeCmd zgsCmd++encodeCmd :: ZGSCmd -> PutM ()+encodeCmd (Publish k v ttl) = do+  putByteStringLen k+  putLongByteStringLen v+  putInt32be $ fromIntegral ttl+encodeCmd _ = return ()++parsePublish :: Get ZGSCmd+parsePublish = Publish+  <$> parseString+  <*> parseLongString+  <*> getInt32++parseCmd :: B.ByteString -> Get ZGSMsg+parseCmd from = do+    cmd <- (getInt8 :: Get Int)+    ver <- getInt8++    if ver /= zgsVer+      then fail "Protocol version mismatch"+      else do++        zcmd <- case cmd of+          0x01 -> pure Hello+          0x02 -> parsePublish+          0x03 -> pure Ping+          0x04 -> pure PingOk+          0x05 -> pure Invalid+          _    -> fail "Unknown command"++        return $ ZGSMsg (Just from) zcmd++parseZGS :: [B.ByteString] -> (Either String ZGSMsg, B.ByteString)+parseZGS [from, msg] = parseZgs from msg+parseZGS x = (Left "empty message", bshow x)++parseZgs :: B.ByteString -> B.ByteString -> (Either String ZGSMsg, B.ByteString)+parseZgs from msg = flip runGet msg $ do+  sig <- getWord16be+  if sig /= zgsSig+    then fail "Signature mismatch"+    else do+      res <- parseCmd from+      return res
+ src/Data/ZMQParse.hs view
@@ -0,0 +1,74 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+module Data.ZMQParse where++import Prelude hiding (putStrLn, take)+import qualified Data.ByteString.Char8 as B++import Data.Binary.Strict.Get+import Data.Binary.Put++import qualified Data.Map as M++getInt8 :: (Integral a) => Get a+getInt8  = fromIntegral <$> getWord8+getInt16 :: (Integral a) => Get a+getInt16 = fromIntegral <$> getWord16be+getInt32 :: (Integral a) => Get a+getInt32 = fromIntegral <$> getWord32be++parseString :: Get B.ByteString+parseString = do+  len <- getInt8+  st <- getByteString len+  return st++parseLongString :: Get B.ByteString+parseLongString = do+  len <- getInt32+  st <- getByteString len+  return st++parseStrings :: Get [B.ByteString]+parseStrings = do+  count <- getInt32+  res <- sequence $ replicate count parseLongString+  return res++parseKV :: Get (B.ByteString, B.ByteString)+parseKV = do+  key <- parseString+  value <- parseLongString+  return (key, value)++parseMap :: Get (M.Map B.ByteString B.ByteString)+parseMap = do+  count <- getInt32+  res <- sequence $ replicate count parseKV+  return $ M.fromList res++putByteStringLen :: B.ByteString -> PutM ()+putByteStringLen x = do+  putInt8 $ fromIntegral $ B.length x+  putByteString x++putLongByteStringLen :: B.ByteString -> PutM ()+putLongByteStringLen x = do+  putInt32be $ fromIntegral $ B.length x+  putByteString x++putByteStrings :: Foldable t => t B.ByteString -> PutM ()+putByteStrings x = do+  putInt32be $ fromIntegral $ length x+  mapM_ putLongByteStringLen x++putKV :: (B.ByteString, B.ByteString) -> PutM ()+putKV (k, v) = do+  putByteStringLen k+  putLongByteStringLen v++putMap :: M.Map B.ByteString B.ByteString -> PutM ()+putMap m = do+  putInt32be $ fromIntegral $ length ml+  mapM_ putKV ml+  where ml = M.toList m
+ src/Data/ZRE.hs view
@@ -0,0 +1,198 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+module Data.ZRE (+    zreVer+  , newZRE+  , parseZRE+  , encodeZRE+  , zreBeacon+  , parseBeacon+  , Name+  , Headers+  , Content+  , Group+  , Groups+  , Seq+  , GroupSeq+  , ZREMsg(..)+  , ZRECmd(..)) where+import Prelude hiding (putStrLn, take)+import qualified Data.ByteString.Char8 as B+import qualified Data.ByteString.Lazy as BL++import GHC.Word++import Data.Binary.Strict.Get+import Data.Binary.Put++import qualified Data.Map as M+import qualified Data.Set as S+import Data.UUID+import Data.Time.Clock++import System.ZMQ4.Endpoint+import Data.ZMQParse++zreVer :: Int+zreVer = 2+zreSig :: Word16+zreSig = 0xAAA1++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]++data ZREMsg = ZREMsg {+    msgFrom :: Maybe UUID+  , msgSeq :: Seq+  , msgTime :: Maybe UTCTime+  , msgCmd :: ZRECmd+  } deriving (Show, Eq, Ord)++data ZRECmd =+    Hello Endpoint Groups GroupSeq Name Headers+  | Whisper Content+  | Shout Group Content+  | Join Group GroupSeq+  | Leave Group GroupSeq+  | Ping+  | PingOk+  deriving (Show, Eq, Ord)++zreBeacon :: B.ByteString -> Port -> B.ByteString+zreBeacon uuid port = BL.toStrict $ runPut $ do+  putByteString "ZRE"+  -- XXX: for compatibility with zyre implementation+  -- this should use 0x01 instead, but why when+  -- we can stick zre version there and use it for filtering?+  -- for now leave in compat mode as we don't+  -- assert this but zyre does+  putInt8 $ fromIntegral (0x01 :: Int) -- compat+  --putInt8 $ fromIntegral zreVer -- non-compat+  putByteString uuid+  putInt16be $ fromIntegral port++parseUUID :: Get UUID+parseUUID =  do+  muuid <- fromByteString . BL.fromStrict <$> getByteString 16+  case muuid of+    Just uuid -> return uuid+    Nothing -> fail "Unable to parse UUID"++parseBeacon :: B.ByteString+            -> (Either String (B.ByteString, Integer, UUID, Integer), B.ByteString)+parseBeacon = runGet $ do+  lead <- getByteString 3+  ver <- fromIntegral <$> getWord8+  uuid <- parseUUID+  port <- fromIntegral <$> getWord16be+  return (lead, ver, uuid, port)++cmdCode :: ZRECmd -> Word8+cmdCode (Hello _ _ _ _ _) = 0x01+cmdCode (Whisper _)       = 0x02+cmdCode (Shout _ _)       = 0x03+cmdCode (Join _ _)        = 0x04+cmdCode (Leave _ _)       = 0x05+cmdCode Ping              = 0x06+cmdCode PingOk            = 0x07++getContent :: ZRECmd -> Content+getContent (Whisper c) = c+getContent (Shout _ c) = c+getContent _ = []++newZRE :: Seq -> ZRECmd -> ZREMsg+newZRE seqNum cmd = ZREMsg Nothing seqNum Nothing cmd++encodeZRE :: ZREMsg -> [B.ByteString]+encodeZRE ZREMsg{..} = msg:(getContent msgCmd)+  where+    msg = BL.toStrict $ runPut $ do+      putWord16be zreSig+      putWord8 $ cmdCode msgCmd+      putInt8 $ fromIntegral zreVer+      putInt16be $ fromIntegral msgSeq+      encodeCmd msgCmd++encodeCmd :: ZRECmd -> PutM ()+encodeCmd (Hello endpoint groups statusSeq name headers) = do+  putByteStringLen (pEndpoint endpoint)+  putByteStrings groups+  putInt8 $ fromIntegral statusSeq+  putByteStringLen name+  putMap headers+encodeCmd (Shout group _content) = putByteStringLen group+encodeCmd (Join group statusSeq) = do+  putByteStringLen group+  putInt8 $ fromIntegral statusSeq+encodeCmd (Leave group statusSeq) = do+  putByteStringLen group+  putInt8 $ fromIntegral statusSeq+encodeCmd _ = return ()++parseHello :: Get ZRECmd+parseHello = Hello+  <$> parseEndpoint'+  <*> fmap S.fromList parseStrings+  <*> getInt8+  <*> parseString+  <*> parseMap+  where+    parseEndpoint' = do+      s <- parseString+      case parseAttoEndpoint s of+        (Left err) -> fail $ "Unable to parse endpoint: " ++ err+        (Right endpoint) -> return endpoint++parseShout :: Content -> Get ZRECmd+parseShout frames = Shout <$> parseString <*> pure frames++parseJoin :: Get ZRECmd+parseJoin = Join <$> parseString <*> getInt8++parseLeave :: Get ZRECmd+parseLeave = Leave <$> parseString <*> getInt8++parseCmd :: B.ByteString -> Content -> Get ZREMsg+parseCmd from frames = do+    cmd <- (getInt8 :: Get Int)+    ver <- getInt8+    sqn <- getInt16++    case runGet parseUUID from of+      (Left _err, _) -> fail "No UUID"+      (Right uuid, _)-> do+        if ver /= zreVer+          then fail "Protocol version mismatch"+          else do++            zcmd <- case cmd of+              0x01 -> parseHello+              0x02 -> pure $ Whisper frames -- parseWhisper+              0x03 -> parseShout frames+              0x04 -> parseJoin+              0x05 -> parseLeave+              0x06 -> pure Ping+              0x07 -> pure PingOk+              _    -> fail "Unknown command"++            return $ ZREMsg (Just uuid) sqn Nothing zcmd++parseZRE :: [B.ByteString] -> (Either String ZREMsg, B.ByteString)+parseZRE (from:msg:rest) = parseZre from msg rest+parseZRE _ = (Left "empty message", "")++parseZre :: B.ByteString -> B.ByteString -> Content -> (Either String ZREMsg, B.ByteString)+parseZre from msg frames = flip runGet msg $ do+  sig <- getWord16be+  if sig /= zreSig+    then fail "Signature mismatch"+    else do+      -- we need to drop 1st byte of from string which is '1':UUID (17 bytes)+      res <- parseCmd (B.tail from) frames+      return res
+ src/Network/ZGossip.hs view
@@ -0,0 +1,95 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+module Network.ZGossip (+    zgossipServer+  , zgossipClient+  , zgossipZRE) where+++import Prelude hiding (putStrLn, take)+import Control.Monad hiding (join)+import Control.Monad.IO.Class+import Control.Concurrent+import Control.Concurrent.Async+import Control.Concurrent.STM++import Data.UUID+import qualified Data.Map as M+import qualified Data.Set as S+import qualified Data.ByteString.Char8 as B+import qualified Data.ByteString.Lazy as BL++import Data.ZGossip+import Network.ZRE.Types (API(DoDiscover))+import Network.ZRE.Utils (bshow)++import Network.ZGossip.ZMQ+import Network.ZGossip.Types++import System.ZMQ4.Endpoint++zgossipClient :: Key -> Endpoint -> Endpoint -> (ZGSMsg -> IO ()) -> IO ()+zgossipClient uuid endpoint ourEndpoint handler = do+  gossipQ <- atomically $ newTBQueue 10+  atomically $ mapM_ (writeTBQueue gossipQ) [Hello]+  pa <- async $ forever $ do+    atomically $ writeTBQueue gossipQ $ Publish uuid (pEndpoint ourEndpoint) 600+    -- publish every 50s+    threadDelay $ 1000000*50++  link pa+  zgossipDealer endpoint uuid gossipQ handler++zgossipServer :: Endpoint -> IO ()+zgossipServer endpoint = do+  gossipS <- atomically $ newTVar emptyGossipState++  let expire = forever $ do+        atomically $ modifyTVar gossipS $ \x -> x { gossipPairs = M.mapMaybe ttlUpdate (gossipPairs x) }+        threadDelay 1000000+        where+           ttlUpdate (_, 0)   = Nothing+           ttlUpdate (v, ttl) = Just (v, ttl - 1)++  ea <- async expire+  link ea++  zgossipRouter endpoint (serverHandle gossipS)++serverHandle :: TVar ZGossipState -> Peer -> ZGSCmd -> IO [(Peer, ZGSCmd)]+serverHandle s from Hello = do+  atomically $ modifyTVar s $ \x -> x { gossipPeers = S.insert from (gossipPeers x) }+  st <- atomically $ readTVar s+  dbg ["Hello from", tryUUID from]+  -- send all the k,v pairs to this client+  return [(from, cvtPub pub) | pub <- M.toList $ gossipPairs st ]+serverHandle s from pub@(Publish k v ttl) = do+  atomically $ modifyTVar s $ \x -> x { gossipPairs = M.insert k (v, ttl) (gossipPairs x) }+  st <- atomically $ readTVar s+  dbg ["Publish from", tryUUID from, tryUUID k, "=", v, "( ttl", bshow ttl, ")"]++  -- republish this to all other clients+  return [(to, pub) | to <- M.keys $ gossipPairs st, to /= from ]+serverHandle _ from Ping = do+  dbg ["Ping from", tryUUID from]+  return [(from, PingOk)]+serverHandle _ _ PingOk = return []+serverHandle _ _ Invalid = return []++tryUUID :: B.ByteString -> B.ByteString+tryUUID x = maybe x toASCIIBytes (fromByteString $ BL.fromStrict x)++dbg :: [B.ByteString] -> IO ()+dbg = B.putStrLn . (B.intercalate " ")++-- send DoDiscover ZRE API messages on new Publish message+zgossipZRE :: TBQueue API -> ZGSMsg -> IO ()+zgossipZRE q ZGSMsg{..} = handlePublish zgsCmd+  where handlePublish (Publish k v _) = do+          case fromByteString $ BL.fromStrict k of+            Nothing ->  liftIO $ B.putStrLn "Can't parse zgossip uuid"+            Just uuid -> do+             case parseAttoEndpoint v of+               (Left _err) -> liftIO $ B.putStrLn "Can't parse zgossip endpoint"+               (Right endpoint) -> atomically $ writeTBQueue q (DoDiscover uuid endpoint)+        handlePublish _ = return ()
+ src/Network/ZGossip/Types.hs view
@@ -0,0 +1,20 @@+module Network.ZGossip.Types (+    ZGossipState(..)+  , emptyGossipState+  , cvtPub) where++import qualified Data.Map as M+import qualified Data.Set as S++import Data.ZGossip++data ZGossipState = ZGossipState {+    gossipPeers   :: S.Set Peer+  , gossipPairs   :: M.Map Key (Value, TTL)+} deriving (Show)++emptyGossipState :: ZGossipState+emptyGossipState = ZGossipState S.empty M.empty++cvtPub :: (Key, (Value, TTL)) -> ZGSCmd+cvtPub (k, (v, ttl)) = Publish k v ttl
+ src/Network/ZGossip/ZMQ.hs view
@@ -0,0 +1,74 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}++{-# LANGUAGE FlexibleContexts #-}+module Network.ZGossip.ZMQ (zgossipDealer, zgossipRouter) where++import Control.Monad+import Control.Concurrent.Async+import Control.Concurrent.STM+import Control.Monad.IO.Class+import qualified System.ZMQ4.Monadic as ZMQ+import qualified Data.ByteString.Char8 as B+import qualified Data.List.NonEmpty as NE++import Data.Maybe++import Data.ZGossip+import System.ZMQ4.Endpoint++zgossipDealer :: Control.Monad.IO.Class.MonadIO m+              => Endpoint+              -> B.ByteString+              -> TBQueue ZGSCmd+              -> (ZGSMsg -> IO ())+              -> m a+zgossipDealer endpoint ourUUID peerQ handler = ZMQ.runZMQ $ do+  d <- ZMQ.socket ZMQ.Dealer+  ZMQ.setLinger (ZMQ.restrict (1 :: Int)) d+  -- Never block on sending; we use an infinite HWM and buffer as many+  -- messages as needed in outgoing pipes. Note that the maximum number+  -- is the overall tuple set size+  ZMQ.setSendHighWM (ZMQ.restrict (0 :: Int)) d+  ZMQ.setSendTimeout (ZMQ.restrict (0 :: Int)) d+  ZMQ.setIdentity (ZMQ.restrict $ ourUUID) d+  ZMQ.connect d $ B.unpack $ pEndpoint endpoint+  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)++  let recv = forever $ do+        input <- ZMQ.receiveMulti d+        case parseZGS input of+           (Left err, _) -> do+             liftIO $ print $ "Malformed gossip message received: " ++ err+             liftIO $ print input+           (Right msg@ZGSMsg{..}, _) -> do+             liftIO $ handler msg++  sa <- ZMQ.async spam+  void $ ZMQ.async recv+  liftIO $ wait sa++zgossipRouter :: (Foldable t, Control.Monad.IO.Class.MonadIO m)+              => Endpoint+              -> (B.ByteString+              -> ZGSCmd+              -> IO (t (B.ByteString, ZGSCmd)))+              -> m a+zgossipRouter endpoint handler = ZMQ.runZMQ $ do+  sock <- ZMQ.socket ZMQ.Router+  ZMQ.bind sock $ B.unpack $ pEndpoint endpoint++  forever $ do+     input <- ZMQ.receiveMulti sock+     case parseZGS input of+        (Left err, _) -> liftIO $ print $ "Malformed gossip message received: " ++ err+        (Right ZGSMsg{..}, _) -> do+            --liftIO $ print msg+            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)+
+ src/Network/ZRE.hs view
@@ -0,0 +1,292 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+module Network.ZRE (+    runZre+  , runZre'+  , readZ+  , writeZ+  , unReadZ+  , defaultConf+  , API(..)+  , Event(..)+  , ZRE+  , Z.Group+  , zjoin+  , zleave+  , zshout+  , zshout'+  , zwhisper+  , zdebug+  , znodebug+  , zquit+  , pEndpoint+  , toASCIIBytes) where++import Prelude hiding (putStrLn, take)+import Control.Monad hiding (join)+import Control.Concurrent+import Control.Concurrent.Async+import Control.Concurrent.STM++import Data.UUID+import Data.UUID.V1+import Data.Maybe++import qualified Data.Set as S+import qualified Data.Map as M+import qualified Data.ByteString.Char8 as B++import qualified Data.ZRE as Z+import Network.ZRE.Beacon+import Network.ZRE.Utils+import Network.ZRE.Peer+import Network.ZRE.ZMQ+import Network.ZRE.Types+import System.ZMQ4.Endpoint++import Network.ZGossip++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+  cfg <- execParser opts+  print cfg+  runZre' cfg app+  where+    opts = info (parseCfg <**> helper)+      ( fullDesc+     <> progDesc "ZRE"+     <> header "zre tools" )++getIfaces :: [B.ByteString]+          -> IO [(B.ByteString, B.ByteString, B.ByteString)]+getIfaces ifcs = do+  case ifcs of+    [] -> do+      dr <- getDefRoute+      case dr of+        Nothing -> exitFail "Unable to get default route"+        Just (_route, iface) -> do+          i <- getIfaceReport iface+          return $ [i]+    x  -> do+      forM x getIfaceReport++runIface :: Show a+         => TVar ZREState+         -> Int+         -> (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+    ifcs <- getIfaces zreInterfaces++    u <- maybeM (exitFail "Unable to get UUID") return nextUUID+    let uuid = uuidByteString u++    case ifcs of+      [] -> exitFail "No interfaces found"+      ifaces@((_ifcname, ipv4, _ipv6):_) -> do+        zrePort <- randPort ipv4++        let zreEndpoint = newTCPEndpoint ipv4 zrePort+        print zreEndpoint++        zreName <- getName zreNamed++        inQ <- atomically $ newTBQueue 1000+        outQ <- atomically $ newTBQueue 1000++        s <- newZREState zreName zreEndpoint u inQ outQ++        -- FIXME: support multiple gossip clients+        case zreZGossip of+          Nothing -> return ()+          Just end -> void $ async $ zgossipClient uuid end zreEndpoint (zgossipZRE outQ)++        (mCastAddr:_) <- toAddrInfo zreMCast+        _beaconAsync <- async $ beacon mCastAddr uuid zrePort+        _beaconRecvAsync <- async $ beaconRecv s zreMCast+        apiAsync <- async $ api s+        _userAppAsync <- async $ runZ app inQ outQ++        mapM_ (runIface s zrePort) ifaces++        wait apiAsync+        --wait userAppAsync+        return ()++api :: TVar ZREState -> IO ()+api s = do+  a <- atomically $ readTVar s >>= readTBQueue . zreOut+  handleApi s a+  case a of+    DoQuit -> return ()+    _ -> api s++handleApi :: TVar ZREState -> API -> IO ()+handleApi s act = do+  case act of+    DoJoin group -> atomically $ do+      incGroupSeq+      modifyTVar s $ \x -> x { zreGroups = S.insert group (zreGroups x) }+      st <- readTVar s+      msgAllJoin s group (zreGroupSeq st)++    DoLeave group -> atomically $ do+      incGroupSeq+      modifyTVar s $ \x -> x { zreGroups = S.delete group (zreGroups x) }+      st <- readTVar s+      msgAllLeave s group (zreGroupSeq st)++    DoShout group msg -> atomically $ shoutGroup s group msg+    DoShoutMulti group mmsg -> atomically $ shoutGroupMulti s group mmsg+    DoWhisper uuid msg -> atomically $ whisperPeerUUID s uuid msg++    DoDiscover uuid endpoint -> do+      mp <- atomically $ lookupPeer s uuid+      case mp of+        Just _ -> return ()+        Nothing -> do+          void $ makePeer s uuid $ newPeerFromEndpoint endpoint++    DoDebug bool -> atomically $ modifyTVar s $ \x -> x { zreDebug = bool }++    DoQuit -> do+      -- FIXME: wait for empty peer queues+      threadDelay (sec (1.0 :: Float))+  where+    incGroupSeq = modifyTVar s $ \x -> x { zreGroupSeq = (zreGroupSeq x) + 1 }++-- handles incoming ZRE messages+-- creates peers, updates state+inbox :: TVar ZREState -> Z.ZREMsg -> IO ()+inbox s msg@Z.ZREMsg{..} = do+  let uuid = fromJust msgFrom++  -- print msg , "state pre-msg", printAll s++  mpt <- atomically $ lookupPeer s uuid+  case mpt of+    Nothing -> do+      case msgCmd of+        -- if the peer is not known but a message is HELLO we create a new+        -- peer, for other messages we don't know the endpoint to connect to+        h@(Z.Hello _endpoint _groups _groupSeq _name _headers) -> do+          peer <- makePeer s uuid $ newPeerFromHello h+          atomically $ updatePeer peer $ \x -> x { peerSeq = (peerSeq x) + 1 }+        -- silently drop any other messages+        _ -> return ()++    (Just peer) -> do+      atomically $ updateLastHeard peer $ fromJust msgTime++      -- destroy/re-start peer when this doesn't match+      p <- atomically $ readTVar peer+      case peerSeq p == msgSeq of+        True -> do+          -- rename to peerExpectSeq, need to update at line 127 too+          atomically $ updatePeer peer $ \x -> x { peerSeq = (peerSeq x) + 1 }+          handleCmd s msg peer+        _ -> do+          atomically $ emitdbg s "sequence mismatch, recreating peer"+          recreatePeer (peerUUID p) msgCmd++  -- "state post-msg", printAll s+  where+    recreatePeer uuid h@(Z.Hello _ _ _ _ _) = do+          destroyPeer s uuid+          peer <- makePeer s uuid $ newPeerFromHello h+          atomically $ updatePeer peer $ \x -> x { peerSeq = (peerSeq x) + 1 }+    recreatePeer uuid _ = destroyPeer s uuid++handleCmd :: TVar ZREState -> Z.ZREMsg -> TVar Peer -> IO ()+handleCmd s Z.ZREMsg{msgFrom=(Just from), msgTime=(Just time), msgCmd=cmd} peer = do+      case cmd of+        (Z.Whisper content) -> atomically $ do+          emit s $ Whisper from content time+          emitdbg s $ B.intercalate " " ["whisper", B.concat content]++        Z.Shout group content -> atomically $ do+          emit s $ Shout from group content time+          emitdbg s $ B.intercalate " " ["shout for group", group, ">", B.concat content]++        Z.Join group groupSeq -> atomically $ do+          joinGroup s peer group groupSeq+          emitdbg s $ B.intercalate " " ["join", group, bshow groupSeq]++        Z.Leave group groupSeq -> atomically $ do+          leaveGroup s peer group groupSeq+          emitdbg s $ B.intercalate " " ["leave", group, bshow groupSeq]++        Z.Ping -> atomically $ do+          msgPeer peer Z.PingOk+          emitdbg s $ "ping"+        Z.PingOk -> return ()+        Z.Hello endpoint groups groupSeq name headers -> do+          -- if this peer was already registered+          -- (e.g. from beacon) update appropriate data+          atomically $ do+            joinGroups s peer groups groupSeq+            updatePeer peer $ \x -> x {+                         peerName = Just name+                       , peerHeaders = headers+                       }+            p <- readTVar peer+            emit s $ Ready (peerUUID p) name groups headers endpoint+            emitdbg s $ "update peer"+          return ()+handleCmd _ _ _ = return ()
+ src/Network/ZRE/Beacon.hs view
@@ -0,0 +1,73 @@+{-# LANGUAGE OverloadedStrings #-}+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.ByteString+import Network.SockAddr+import Network.Multicast++import Data.Maybe+import Data.UUID+import Data.Time.Clock+import qualified Data.Map as M+import qualified Data.ByteString.Char8 as B++import Data.ZRE+import Network.ZRE.Peer+import Network.ZRE.Types+import System.ZMQ4.Endpoint++beaconRecv :: TVar ZREState -> Endpoint -> IO b+beaconRecv s e = do+    sock <- multicastReceiver (B.unpack $ endpointAddr e) (fromIntegral $ fromJust $ endpointPort e)+    forever $ do+        (msg, addr) <- recvFrom sock 22+        case parseBeacon msg of+          (Left err, _remainder) -> print err+          (Right (_lead, _ver, uuid, port), _) -> do+            case addr of+              x@(SockAddrInet _hisport _host) -> do+                beaconHandle s (showSockAddrBS x) uuid (fromIntegral port)+              x@(SockAddrInet6 _hisport _ _host _) -> do+                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 ()+beaconHandle s addr uuid port = do+    st <- atomically $ readTVar s++    if uuid == zreUUID st+      then return () -- our own message+      else do+        case M.lookup uuid $ zrePeers st of+          (Just peer) -> do+            now <- getCurrentTime+            atomically $ updateLastHeard peer now+          Nothing -> do+            -- B.putStrLn $ B.concat ["New peer from beacon ", B.pack $ show uuid, " (", addr, ":", B.pack $ show port , ")"]+            void $ makePeer s uuid $ newPeerFromBeacon addr port+            return ()+++-- sends udp multicast beacons+beacon :: AddrInfo -> B.ByteString -> Port -> IO a+beacon addrInfo uuid port = do+    withSocketsDo $ do+      bracket (getSocket addrInfo) close (talk (addrAddress addrInfo) (zreBeacon uuid port))+  where+    getSocket addr = do+      s <- socket (addrFamily addr) Datagram defaultProtocol+      mapM_ (\x -> setSocketOption s x 1) [Broadcast, ReuseAddr, ReusePort]+      bind s (addrAddress addr)+      return s+    talk addr msg s =+      forever $ do+      void $ sendTo s msg addr+      threadDelay zreBeaconMs
+ src/Network/ZRE/Parse.hs view
@@ -0,0 +1,50 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+module Network.ZRE.Parse (parseApi, parseAttoApi) where++import Control.Applicative++import Data.UUID+import Data.Attoparsec.ByteString.Char8 as A+import qualified Data.ByteString.Char8 as B++import Network.ZRE.Types++parseAttoApi :: B.ByteString -> Either String API+parseAttoApi = A.parseOnly parseApi++parseApi :: Parser API+parseApi = do+  parseControl++parseControl :: Parser API+parseControl = char '/' *> parseCmd++parseCmd :: Parser API+parseCmd =+      DoJoin <$> (string "join" *> lskip *> word)+  <|> DoLeave <$> (string "leave" *> lskip *> word)+  <|> DoShout <$> (string "shout" *> lskip *> 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 = lskip *> word++lskip :: Parser ()+lskip = skipWhile (==' ')++word :: Parser B.ByteString+word = A.takeWhile (/=' ')++--uEOL :: Parser B.ByteString+--uEOL = A.takeTill (pure False)++uuid :: Parser UUID+uuid = do+  mx <- fromASCIIBytes <$> lw+  case mx of+        Nothing -> fail "no uuid"+        Just x -> return x
+ src/Network/ZRE/Peer.hs view
@@ -0,0 +1,302 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+module Network.ZRE.Peer (+    newPeerFromBeacon+  , newPeerFromHello+  , newPeerFromEndpoint+  , makePeer+  , destroyPeer+  , msgPeer+  , msgPeerUUID+  , msgAll+  , msgGroup+  , joinGroup+  , joinGroups+  , leaveGroup+  , leaveGroups+  , lookupPeer+  , updatePeer+  , updateLastHeard+  , printPeer+  , printAll+  , msgAllJoin+  , msgAllLeave+  , shoutGroup+  , shoutGroupMulti+  , whisperPeerUUID+  ) where++import Control.Monad+import Control.Monad.IO.Class+import Control.Concurrent+import Control.Concurrent.Async+import Control.Concurrent.STM+import qualified Data.Map as M+import qualified Data.Set as Set+import qualified Data.ByteString.Char8 as B+import Data.Time.Clock+import Data.UUID+import Data.ZRE()+import System.ZMQ4.Endpoint+import Data.ZRE+import Network.ZRE.Types hiding (Shout, Whisper)+import Network.ZRE.Utils+import Network.ZRE.ZMQ (zreDealer)++printPeer :: Peer -> B.ByteString+printPeer Peer{..} = B.intercalate " "+  ["Peer",+    bshow peerName,+    pEndpoint peerEndpoint,+    toASCIIBytes peerUUID,+    bshow peerSeq,+    bshow peerGroupSeq,+    bshow peerGroups,+    bshow peerLastHeard]++newPeer :: MonadIO m+        => TVar ZREState+        -> Endpoint+        -> UUID+        -> Set.Set Group+        -> GroupSeq+        -> Maybe Name+        -> Headers+        -> UTCTime+        -> STM (TVar Peer, Maybe (m a), Maybe (IO b))+newPeer s endpoint uuid groups groupSeq mname headers t = do+  st <- readTVar s+  peerQ <- newTBQueue 100+  writeTBQueue peerQ $ Hello (zreEndpoint st) (zreGroups st) (zreGroupSeq st) (zreName st) (zreHeaders st)++  let p = Peer {+              peerEndpoint  = endpoint+            , peerUUID      = uuid+            , peerSeq       = 1+            , peerGroups    = groups+            , peerGroupSeq  = 0+            , peerName      = mname+            , peerHeaders   = headers+            , peerAsync     = Nothing+            , peerAsyncPing = Nothing+            , peerQueue     = peerQ+            , peerLastHeard = t }+  np <- newTVar $ p++  modifyTVar s $ \x -> x { zrePeers = M.insert uuid np (zrePeers x) }++  emit s $ New uuid mname groups headers endpoint+  case mname of+    (Just name) -> emit s $ Ready uuid name groups headers endpoint+    Nothing -> return ()++  joinGroups s np groups groupSeq++  return $ (np, Just $ zreDealer endpoint (uuidByteString $ zreUUID st) peerQ, Just $ pinger s np)++newPeerFromBeacon :: MonadIO m+                  => Address+                  -> Port+                  -> UTCTime+                  -> UUID+                  -> TVar ZREState+                  -> STM (TVar Peer, Maybe (m a), Maybe (IO b))+newPeerFromBeacon addr port t uuid s = do+  emitdbg s "New peer from beacon"+  let endpoint = newTCPEndpoint addr port+  newPeer s endpoint uuid (Set.empty :: Groups) 0 Nothing M.empty t++newPeerFromHello :: MonadIO m+                 => ZRECmd+                 -> UTCTime+                 -> UUID+                 -> TVar ZREState+                 -> STM (TVar Peer, Maybe (m a), Maybe (IO b))+newPeerFromHello (Hello endpoint groups groupSeq name headers) t uuid s = do+  emitdbg s "New peer from hello"+  newPeer s endpoint uuid groups groupSeq (Just name) headers t+newPeerFromHello _ _ _ _ = fail "not a hello message"++newPeerFromEndpoint :: MonadIO m+                    => Endpoint+                    -> UTCTime+                    -> UUID+                    -> TVar ZREState+                    -> STM (TVar Peer, Maybe (m a), Maybe (IO b))+newPeerFromEndpoint endpoint t uuid s = do+  emitdbg s "New peer from endpoint"+  newPeer s endpoint uuid (Set.empty :: Groups) 0 Nothing M.empty t++makePeer :: TVar ZREState+            -> UUID+            -> (UTCTime+                -> UUID+                -> TVar ZREState+                -> STM (TVar Peer, Maybe (IO ()), Maybe (IO ())))+            -> IO (TVar Peer)+makePeer s uuid newPeerFn = do+  t <- getCurrentTime+  res <- atomically $ do+    st <- readTVar s+    case M.lookup uuid $ zrePeers st of+      (Just peer) -> return (peer, Nothing, Nothing)+      Nothing -> newPeerFn t uuid s++  case res of+    -- fixme: clumsy+    (peer, Just deal, Just ping) -> do+      a <- async deal+      b <- async ping+      atomically $ do+        updatePeer peer $ \x -> x { peerAsync = (Just a) }+        updatePeer peer $ \x -> x { peerAsyncPing = (Just b) }++      return peer+    (peer, _, _) -> return peer++destroyPeer :: TVar ZREState -> UUID -> IO ()+destroyPeer s uuid = do+  asyncs <- atomically $ do+    mpt <- lookupPeer s uuid+    case mpt of+      Nothing -> return []+      (Just peer) -> do+        Peer{..} <- readTVar peer+        modifyTVar s $ \x -> x { zrePeers = M.delete uuid (zrePeers x) }+        leaveGroups s peer peerGroups peerGroupSeq+        emit s $ Quit uuid peerName++        return [peerAsync, peerAsyncPing]++  -- this is called from pinger so no more code is executed after this point+  mapM_ cancelM asyncs+  where+    cancelM Nothing = return ()+    cancelM (Just a) = cancel a++pinger :: TVar ZREState -> TVar Peer -> IO b+pinger s peer = forever $ do+  Peer{..} <- atomically $ readTVar peer+  now <- getCurrentTime+  if diffUTCTime now peerLastHeard > deadPeriod+    then destroyPeer s peerUUID+    else do+      let tdiff = diffUTCTime now peerLastHeard+      if tdiff > quietPeriod+        then do+          atomically $ writeTBQueue peerQueue $ Ping+          threadDelay quietPingRate+        else do+          threadDelay $ sec (quietPeriod - tdiff)++lookupPeer :: TVar ZREState -> UUID -> STM (Maybe (TVar Peer))+lookupPeer s uuid = do+  st <- readTVar s+  return $ M.lookup uuid $ zrePeers st++updatePeer :: TVar Peer -> (Peer -> Peer) -> STM ()+updatePeer peer fn = modifyTVar peer fn++--updatePeerUUID :: TVar ZREState -> UUID -> (Peer -> Peer) -> STM ()+--updatePeerUUID s uuid fn = do+--  st <- readTVar s+--  case M.lookup uuid $ zrePeers st of+--    Nothing -> return ()+--    (Just peer) -> updatePeer peer fn++updateLastHeard :: TVar Peer -> UTCTime -> STM ()+updateLastHeard peer val = updatePeer peer $ \x -> x { peerLastHeard = val }++-- join `peer` to `group`, update group sequence nuber to `groupSeq`+joinGroup :: TVar ZREState -> TVar Peer -> Group -> GroupSeq -> STM ()+joinGroup s peer group groupSeq = do+  updatePeer peer $ \x -> x { peerGroups = Set.insert group (peerGroups x) }+  updatePeer peer $ \x -> x { peerGroupSeq = groupSeq }+  p <- readTVar peer+  emit s $ GroupJoin (peerUUID p) group+  modifyTVar s $ \x -> x { zrePeerGroups = M.alter (f p) group $ zrePeerGroups x }+  where+    f p Nothing = Just $ M.fromList [(peerUUID p, peer)]+    f p (Just old) = Just $ M.insert (peerUUID p) peer old++joinGroups :: TVar ZREState -> TVar Peer -> Set.Set Group -> GroupSeq -> STM ()+joinGroups s peer groups groupSeq = do+  mapM_ (\x -> joinGroup s peer x groupSeq) $ Set.toList groups++leaveGroup :: TVar ZREState -> TVar Peer -> Group -> GroupSeq -> STM ()+leaveGroup s peer group groupSeq = do+  updatePeer peer $ \x -> x { peerGroups = Set.delete group (peerGroups x) }+  updatePeer peer $ \x -> x { peerGroupSeq = groupSeq }+  p <- readTVar peer+  emit s $ GroupLeave (peerUUID p) group+  modifyTVar s $ \x -> x { zrePeerGroups = M.alter (f p) group $ zrePeerGroups x }+  where+    f _ Nothing = Nothing+    f p (Just old) = nEmpty $ M.delete (peerUUID p) old+    nEmpty pmap | M.null pmap = Nothing+    nEmpty pmap = Just pmap++leaveGroups :: TVar ZREState -> TVar Peer -> Set.Set Group -> GroupSeq -> STM ()+leaveGroups s peer groups groupSeq = do+  mapM_ (\x -> leaveGroup s peer x groupSeq) $ Set.toList groups++msgPeer :: TVar Peer -> ZRECmd -> STM ()+msgPeer peer msg = do+  p <- readTVar peer+  writeTBQueue (peerQueue p) msg++msgPeerUUID :: TVar ZREState -> UUID -> ZRECmd -> STM ()+msgPeerUUID s uuid msg = do+  st <- readTVar s+  case M.lookup uuid $ zrePeers st of+    Nothing -> return ()+    (Just peer) -> do+      msgPeer peer msg+      return ()++msgAll :: TVar ZREState -> ZRECmd -> STM ()+msgAll s msg = do+  st <- readTVar s+  mapM_ (flip msgPeer msg) (zrePeers st)++msgGroup :: TVar ZREState -> Group -> ZRECmd -> STM ()+msgGroup s groupname msg = do+  st <- readTVar s+  case M.lookup groupname $ zrePeerGroups st of+    Nothing -> return () -- XXX: should report no such group error?+    (Just group) -> do+      mapM_ (flip msgPeer msg) group++shoutGroup :: TVar ZREState -> Group -> B.ByteString -> STM ()+shoutGroup s group msg = msgGroup s group $ Shout group [msg]++shoutGroupMulti :: TVar ZREState -> Group -> Content -> STM ()+shoutGroupMulti s group mmsg = msgGroup s group $ Shout group mmsg++msgAllJoin :: TVar ZREState -> Group -> GroupSeq -> STM ()+msgAllJoin s group sq = msgAll s $ Join group sq++msgAllLeave :: TVar ZREState -> Group -> GroupSeq -> STM ()+msgAllLeave s group sq = msgAll s $ Leave group sq++whisperPeerUUID :: TVar ZREState -> UUID -> B.ByteString -> STM ()+whisperPeerUUID s uuid msg = msgPeerUUID s uuid $ Whisper [msg]++printPeers :: M.Map k (TVar Peer) -> IO ()+printPeers x = do+  mapM_ ePrint $ M.elems x+  where+    ePrint pt = do+      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, "->"]+  printPeers v++printAll :: TVar ZREState -> IO ()+printAll s = do+  st <- atomically $ readTVar s+  printPeers $ zrePeers st+  mapM_ printGroup $ M.toList $ zrePeerGroups st
+ src/Network/ZRE/Types.hs view
@@ -0,0 +1,223 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE TypeFamilies #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}+module Network.ZRE.Types where++import Control.Monad.Reader+import Control.Monad.Base+import Control.Monad.Trans.Control+import Control.Concurrent.Async+import Control.Concurrent.STM+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.ZRE hiding (Shout, Whisper) -- (Name, Seq, Group, Groups, GroupSeq, Headers, Content, ZRECmd, ZREMsg)+import System.ZMQ4.Endpoint+++isec :: (Num a) => a -> a+isec  = (*1000000)+sec  :: (RealFrac a) => a -> Int+sec  = round . isec+msec :: (RealFrac a) => a -> Int+msec = round . (*1000)++-- send beacon every 0.9 seconds+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+  } deriving (Show)++defMCastEndpoint :: Endpoint+defMCastEndpoint = newUDPEndpoint "225.25.25.25" 5670++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+  }++data Event =+    New UUID (Maybe Name) Groups Headers Endpoint+  | Ready UUID Name Groups Headers Endpoint+  | GroupJoin UUID Group+  | GroupLeave UUID Group+  | Quit UUID (Maybe Name)+  | Message ZREMsg+  | Shout UUID Group Content UTCTime+  | Whisper UUID Content UTCTime+  | Debug B.ByteString+  deriving (Show)++data API =+    DoJoin Group+  | DoLeave Group+  | DoShout Group B.ByteString+  | DoShoutMulti Group [B.ByteString]+  | DoWhisper UUID B.ByteString+  | DoDiscover UUID Endpoint+  | DoDebug Bool+  | DoQuit+  deriving (Show)++type Peers = M.Map UUID (TVar Peer)+type PeerGroups = M.Map Group Peers++type EventQueue = TBQueue Event+type APIQueue = TBQueue API+++data ZREState = ZREState {+    zreUUID       :: UUID+  , zrePeers      :: Peers+  , zrePeerGroups :: PeerGroups+  , zreEndpoint   :: Endpoint+  , zreGroups     :: Groups+  , zreGroupSeq   :: GroupSeq+  , zreName       :: Name+  , zreHeaders    :: Headers+  , zreDebug      :: Bool+  , zreIn         :: EventQueue+  , zreOut        :: APIQueue+  , zreIfaces     :: M.Map B.ByteString [Async ()]+  }++data Peer = Peer {+    peerEndpoint  :: Endpoint+  , peerUUID      :: UUID+  , peerSeq       :: Seq+  , peerGroups    :: Groups+  , peerGroupSeq  :: GroupSeq+  , peerName      :: Maybe Name+  , peerHeaders   :: Headers+  , peerAsync     :: Maybe (Async ())+  , peerAsyncPing :: Maybe (Async ())+  , peerQueue     :: TBQueue ZRECmd+  , peerLastHeard :: UTCTime+  }+  deriving (Show)++instance Show a => Show (TBQueue a) where+  show = pure "TBQueue"++instance Show a => Show (Async a) where+  show = pure "Async"++newtype ZRE a = Z {+  runZ' :: ReaderT (EventQueue, APIQueue) IO a+}+  deriving (Functor, Applicative, Monad, MonadIO,+    MonadBase IO,+    MonadReader (EventQueue, APIQueue))++instance MonadBaseControl IO ZRE where+  type StM ZRE a = a+  liftBaseWith f = Z $ liftBaseWith $ \q -> f (q . runZ')+  restoreM = Z . restoreM++runZ :: ZRE a -> EventQueue -> APIQueue -> IO a+runZ app events api = runReaderT (runZ' app) (events, api)++readZ :: ZRE (Event)+readZ = do+  (e, _) <- ask+  v <- liftIO $ atomically $ readTBQueue e+  return v++unReadZ :: Event -> ZRE ()+unReadZ x = do+  (e, _) <- ask+  void $ liftIO $ atomically $ unGetTBQueue e x++writeZ :: API -> ZRE ()+writeZ x = do+  (_, a) <- ask+  liftIO $ atomically $ writeTBQueue a x++zjoin :: Group -> ZRE ()+zjoin = writeZ . DoJoin++zleave :: Group -> ZRE ()+zleave = writeZ . DoLeave++zshout :: Group -> B.ByteString -> ZRE ()+zshout group msg = writeZ $ DoShout group msg++zshout' :: Group -> [B.ByteString] -> ZRE ()+zshout' group msgs = writeZ $ DoShoutMulti group msgs++zwhisper :: UUID -> B.ByteString -> ZRE ()+zwhisper uuid msg = writeZ $ DoWhisper uuid msg++zdebug :: ZRE ()+zdebug = writeZ $ DoDebug True++znodebug :: ZRE ()+znodebug = writeZ $ DoDebug False++zquit :: ZRE ()+zquit = writeZ $ DoQuit++maybeM :: Monad m => m b -> (a -> m b) -> m (Maybe a) -> m b+maybeM err f value = value >>= maybe err f++newZREState :: Name+            -> Endpoint+            -> UUID+            -> EventQueue+            -> APIQueue+            -> IO (TVar ZREState)+newZREState name endpoint u inQ outQ = atomically $ newTVar $+  ZREState {+    zreUUID = u+    , zrePeers = M.empty+    , zrePeerGroups = M.empty+    , zreEndpoint = endpoint+    , zreGroups = S.empty+    , zreGroupSeq = 0+    , zreName = name+    , zreHeaders = M.empty+    , zreDebug = False+    , zreIn = inQ+    , zreOut = outQ+    , zreIfaces = M.empty+    }
+ src/Network/ZRE/Utils.hs view
@@ -0,0 +1,105 @@+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE OverloadedStrings #-}+module Network.ZRE.Utils (+    uuidByteString+  , exitFail+  , bshow+  , getDefRoute+  , getIface+  , getIfaceReport+  , getName+  , randPort+  , emit+  , emitdbg) where+++import System.Exit+import System.Process+import System.Random+import System.ZMQ4.Endpoint+import Network.BSD (getHostName)+import Network.Info+import Network.ZRE.Types+import Control.Concurrent.STM+import Control.Exception+import Network.Socket hiding (Debug)++import Data.UUID (UUID, toByteString)+import Data.Maybe+import qualified Data.ByteString.Char8 as B+import qualified Data.ByteString.Lazy as BL++uuidByteString :: UUID -> B.ByteString+uuidByteString = BL.toStrict . toByteString++exitFail :: B.ByteString -> IO b+exitFail msg = do+  B.putStrLn msg+  exitFailure++bshow :: (Show a) => a -> B.ByteString+bshow = B.pack . show++getDefRoute :: IO (Maybe (B.ByteString, B.ByteString))+getDefRoute = do+  ipr <- fmap lines $ readProcess "ip" ["route"] []+  return $ listToMaybe $ catMaybes $ map getDef (map words ipr)+  where+    getDef ("default":"via":gw:"dev":dev:_) = Just (B.pack gw, B.pack dev)+    getDef _ = Nothing++getIface :: B.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 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 "" = fmap B.pack getHostName+getName x  = return x++randPort :: B.ByteString -> IO Port+randPort ip = loop (100 :: Int)+  where+    loop cnt = do+      port <- randomRIO (41000, 41100)+      (xAddr:_) <- getAddrInfo Nothing (Just $ B.unpack ip) (Just $ show port)+      esocket <- try $ getSocket xAddr+      case esocket :: Either IOException Socket of+        Left e+            | cnt <= 1 -> error $ concat+                [ "Unable to bind to random port, last tried was "+                , show port+                , ". Exception was: "+                , show e+                ]+            | otherwise -> do+                loop $! cnt - 1++        Right s -> do+          close s+          return port++    getSocket addr = do+     s <- socket (addrFamily addr) Stream defaultProtocol+     bind s (addrAddress addr)+     return s++emit :: TVar ZREState -> Event -> STM ()+emit s x = do+  st <- readTVar s+  writeTBQueue (zreIn st) x++emitdbg :: TVar ZREState -> B.ByteString -> STM ()+emitdbg s x = do+  st <- readTVar s+  case zreDebug st of+    True -> writeTBQueue (zreIn st) $ Debug x+    _ -> return ()
+ src/Network/ZRE/ZMQ.hs view
@@ -0,0 +1,51 @@+{-# LANGUAGE OverloadedStrings #-}+module Network.ZRE.ZMQ (zreRouter, zreDealer) where++import Control.Monad+import Control.Concurrent.STM+import Control.Monad.IO.Class+import qualified System.ZMQ4.Monadic as ZMQ+import qualified Data.ByteString.Char8 as B+import qualified Data.List.NonEmpty as NE+import Data.Time.Clock.POSIX++import Data.ZRE+import System.ZMQ4.Endpoint++zreDealer :: Control.Monad.IO.Class.MonadIO m+          => Endpoint+          -> B.ByteString+          -> TBQueue ZRECmd+          -> m a+zreDealer endpoint ourUUID peerQ = ZMQ.runZMQ $ do+  d <- ZMQ.socket ZMQ.Dealer+  ZMQ.setLinger (ZMQ.restrict (1 :: Int)) d+  -- The sender MAY set a high-water mark (HWM) of, for example, 100 messages per second (if the timeout period is 30 second, this means a HWM of 3,000 messages).+  ZMQ.setSendHighWM (ZMQ.restrict $ (30 * 100 :: Int)) d+  ZMQ.setSendTimeout (ZMQ.restrict (0 :: Int)) d+  -- prepend '1' in front of 16bit UUID, ZMQ.restrict would do that for us but protocol requires it+  ZMQ.setIdentity (ZMQ.restrict $ B.cons '1' ourUUID) d+  ZMQ.connect d $ B.unpack $ pEndpoint endpoint+  loop d 1 -- sequence number must start with 1+  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)+           loop d (x+1)++zreRouter :: Control.Monad.IO.Class.MonadIO m+          => Endpoint+          -> (ZREMsg -> IO a1)+          -> m a+zreRouter endpoint handler = ZMQ.runZMQ $ do+  sock <- ZMQ.socket ZMQ.Router+  ZMQ.bind sock $ B.unpack $ pEndpoint endpoint+  forever $ do+     input <- ZMQ.receiveMulti sock+     now <- liftIO $ getCurrentTime+     case parseZRE input of+        (Left err, _) -> liftIO $ print $ "Malformed message received: " ++ err+        (Right msg, _) -> do+          let updateTime = \x -> x { msgTime = Just now }+          void $ liftIO $ handler (updateTime msg)+          return ()
+ src/System/ZMQ4/Endpoint.hs view
@@ -0,0 +1,126 @@+{-# LANGUAGE OverloadedStrings #-}+module System.ZMQ4.Endpoint (+    parseAttoEndpoint+  , parseAttoTCPEndpoint+  , parseAttoUDPEndpoint+  , pTransport+  , pEndpoint+  , endpointAddr+  , endpointPort+  , endpointTransport+  , newEndpoint+  , newEndpointPort+  , newEndpointAddrInfo+  , newEndpointPortAddrInfo+  , newTCPEndpoint+  , newTCPEndpointAddrInfo+  , newUDPEndpoint+  , toAddrInfo+  , Port+  , Address+  , Transport(..)+  , Endpoint(..)) where++import Control.Applicative+import Data.Attoparsec.ByteString.Char8 as A+import qualified Data.ByteString.Char8 as B+import Data.Char (toLower)++import Network.Socket  -- only need addrAddress+import Network.SockAddr (showSockAddrBS)++type Port = Int+type Address = B.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++pEndpoint :: Endpoint -> B.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]++newEndpoint :: Transport -> Address -> Endpoint+newEndpoint transport addr = newEndpointPort' transport addr Nothing++newEndpointAddrInfo :: Transport -> AddrInfo -> Endpoint+newEndpointAddrInfo transport addr = newEndpointPortAddrInfo' transport addr Nothing++newEndpointPort' :: Transport -> Address -> Maybe Port -> Endpoint+newEndpointPort' transport addr port = Endpoint transport addr port++newEndpointPortAddrInfo' :: Transport -> AddrInfo -> Maybe Port -> Endpoint+newEndpointPortAddrInfo' transport addr port = newEndpointPort' transport (showSockAddrBS $ 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)++newTCPEndpoint :: Address -> Port -> Endpoint+newTCPEndpoint addr port = newEndpointPort TCP addr port++newUDPEndpoint :: Address -> Port -> Endpoint+newUDPEndpoint addr port = newEndpointPort UDP addr port++newTCPEndpointAddrInfo :: AddrInfo -> Port -> Endpoint+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++parseTransport :: Parser Transport+parseTransport = do+  t <- A.takeWhile (/=':')+  _ <- string "://"+  r <- case t of+    "tcp" -> pure TCP+    "ipc" -> pure IPC+    "inproc" -> pure InProc+    "pgm" -> pure PGM+    "epgm" -> pure EPGM+    _ -> fail $ "Unknown transport" ++ (B.unpack t)++  return r++parseAddress :: Parser Address+parseAddress = A.takeWhile(/=':')++parsePort :: Parser Port+parsePort = do+  _ <- char ':'+  d <- decimal+  return d++parseEndpoint :: Parser Endpoint+parseEndpoint = Endpoint <$> parseTransport <*> parseAddress <*> optional parsePort++parseTCPEndpoint :: Parser Endpoint+parseTCPEndpoint = Endpoint <$> pure TCP <*> parseAddress <*> optional parsePort++parseUDPEndpoint :: Parser Endpoint+parseUDPEndpoint = Endpoint <$> pure UDP <*> parseAddress <*> optional parsePort++parseAttoEndpoint :: B.ByteString -> Either String Endpoint+parseAttoEndpoint = A.parseOnly parseEndpoint++parseAttoTCPEndpoint :: B.ByteString -> Either String Endpoint+parseAttoTCPEndpoint = A.parseOnly parseTCPEndpoint++parseAttoUDPEndpoint :: B.ByteString -> Either String Endpoint+parseAttoUDPEndpoint = A.parseOnly parseUDPEndpoint++endpointAddr :: Endpoint -> Address+endpointAddr (Endpoint _ a _) = a++endpointPort :: Endpoint -> Maybe Port+endpointPort (Endpoint _ _ p) = p++endpointTransport :: Endpoint -> Transport+endpointTransport (Endpoint t _ _) = t
+ test/Spec.hs view
@@ -0,0 +1,2 @@+main :: IO ()+main = putStrLn "Test suite not yet implemented"
+ zre.cabal view
@@ -0,0 +1,151 @@+name:                zre+version:             0.1.0.0+synopsis:            ZRE protocol implementation+description:         See README.rst+--homepage:            https://+license:             BSD3+license-file:        LICENSE+author:              Richard Marko+maintainer:          srk@48.io+copyright:           2016 Richard Marko+category:            Web+build-type:          Simple+-- extra-source-files:+cabal-version:       >=1.10++library+  hs-source-dirs:      src+  exposed-modules:       Network.ZRE+                       , Network.ZRE.Beacon+                       , Network.ZRE.Parse+                       , Network.ZRE.Peer+                       , Network.ZRE.Utils+                       , Network.ZRE.Types+                       , Network.ZRE.ZMQ+                       , Network.ZGossip+                       , Network.ZGossip.ZMQ+                       , Network.ZGossip.Types+                       , Data.ZRE+                       , Data.ZMQParse+                       , Data.ZGossip+                       , System.ZMQ4.Endpoint+  build-depends:       base >= 4.7 && < 5+                     , async+                     , attoparsec+                     , network+                     , network-info+                     , network-multicast+                     , binary+                     , binary-strict+                     , bytestring+                     , containers+                     , mtl+                     , monad-control+                     , optparse-applicative+                     , transformers-base+                     , sockaddr+                     , process+                     , random+                     , stm+                     , time+                     , uuid+                     , zeromq4-haskell+  default-language:    Haskell2010+  ghc-options:         -Wall++executable zre+  hs-source-dirs:      app+  main-is:             Main.hs+  ghc-options:         -threaded -rtsopts -with-rtsopts=-N+  build-depends:       base+                     , bytestring+                     , async+                     , lifted-async+                     , zre+  default-language:    Haskell2010++executable mzre+  hs-source-dirs:      app+  main-is:             Monadic.hs+  ghc-options:         -threaded -rtsopts -with-rtsopts=-N+  build-depends:       base+                     , bytestring+                     , mtl+                     , stm+                     , lifted-async+                     , zre+  default-language:    Haskell2010++executable zreworker+  hs-source-dirs:      app+  main-is:             Worker.hs+  ghc-options:         -threaded -rtsopts -with-rtsopts=-N+  build-depends:       base+                     , async+                     , bytestring+                     , mtl+                     , monad-control+                     , lifted-async+                     , zre+  default-language:    Haskell2010++executable zgossip_server+  hs-source-dirs:      app+  main-is:             ZGossipSrv.hs+  ghc-options:         -threaded -rtsopts -with-rtsopts=-N+  build-depends:       base+                     , async+                     , bytestring+                     , mtl+                     , monad-control+                     , lifted-async+                     , zre+  default-language:    Haskell2010++executable zrematch+  hs-source-dirs:      app+  main-is:             Match.hs+  ghc-options:         -threaded -rtsopts -with-rtsopts=-N+  build-depends:       base+                     , async+                     , bytestring+                     , mtl+                     , monad-control+                     , lifted-async+                     , zre+  default-language:    Haskell2010++executable zretime+  hs-source-dirs:      app+  main-is:             Time.hs+  ghc-options:         -threaded -rtsopts -with-rtsopts=-N+  build-depends:       base+                     , bytestring+                     , time+                     , lifted-async+                     , zre+  default-language:    Haskell2010++executable zrecat+  hs-source-dirs:      app+  main-is:             Cat.hs+  ghc-options:         -threaded -rtsopts -with-rtsopts=-N+  build-depends:       base+                     , bytestring+                     , time+                     , lifted-async+                     , zre+  default-language:    Haskell2010++test-suite zre-test+  type:                exitcode-stdio-1.0+  hs-source-dirs:      test+  main-is:             Spec.hs+  build-depends:       base+                     , zre+  ghc-options:         -threaded -rtsopts -with-rtsopts=-N+  default-language:    Haskell2010++source-repository head+  type:     git+  location: https://git.48.io/zre/