-- |
-- Copyright: (C) 2014 EURL Tweag
--
-- 0MQ implemmentation of the network-transport API.
--
-- The 0MQ implementation guarantees that only a single connection (socket) will
-- be used between endpoints, provided that the addresses specified are
-- canonical. If /A/ connects to /B/ and reports its address as
-- @192.168.0.1:8080@ and /B/ subsequently connects tries to connect to /A/ as
-- @client1.local:http-alt@ then the transport layer will not realize that the
-- TCP connection can be reused.
--
-- This module is intended to be imported qualified.
{-# LANGUAGE BangPatterns #-}
module Network.Transport.ZMQ
( -- * Main API
createTransport
, ZMQParameters(..)
, SecurityMechanism(..)
, defaultZMQParameters
-- * ZeroMQ specific functionality
-- $zeromqs
, Hints(..)
, defaultHints
, apiNewEndPoint
-- * Internals
-- $internals
, createTransportExposeInternals
, breakConnectionEndPoint
, breakConnection
, unsafeConfigurePush
-- $cleanup
, registerCleanupAction
, registerValidCleanupAction
, applyCleanupAction
-- * Design
-- $design
-- ** Multicast
-- $multicast
) where
import Prelude hiding (sequence_)
import Network.Transport.ZMQ.Internal.Types
import qualified Network.Transport.ZMQ.Internal as ZMQ
import Control.Applicative
import Control.Category
( (>>>) )
import Control.Concurrent
( yield
, threadDelay
)
import qualified Control.Concurrent.Async as Async
import Control.Concurrent.MVar
import Control.Concurrent.STM
import Control.Concurrent.STM.TMChan
import Control.Monad
( void
, forever
, unless
, join
, foldM
, when
, (<=<)
, liftM2
)
import Control.Exception
( mapException
, catches
, throwIO
, Handler(..)
, evaluate
)
import Control.Monad.Catch
( finally
, try
, Exception
, MonadCatch
, SomeException(..)
, mask
, mask_
, uninterruptibleMask_
, throwM
)
import Data.Binary
import Data.ByteString (ByteString)
import qualified Data.ByteString as B
import qualified Data.ByteString.Lazy as BL
import qualified Data.ByteString.Char8 as B8
import Data.Foldable
( traverse_
, sequence_
)
import qualified Data.Foldable as Foldable
import Data.IORef
( newIORef
, modifyIORef
, readIORef
, writeIORef
, atomicModifyIORef'
)
import qualified Data.IntMap.Strict as IntMap
import Data.List.NonEmpty
import qualified Data.Map.Strict as Map
import qualified Data.Set as Set
import qualified Data.Traversable as Traversable
import Data.Typeable
import Data.Unique
import GHC.Generics
( Generic )
import Network.Transport
import System.IO
( fixIO
, hPutStrLn
, stderr
)
import System.ZMQ4
( Context )
import qualified System.ZMQ4 as ZMQ
import Data.Accessor ((^.), (^=), (^:) )
import Text.Printf
--------------------------------------------------------------------------------
--- Internal datatypes --
--------------------------------------------------------------------------------
-- $design
--
-- In the zeromq backend we are using following address scheme:
--
-- @
-- tcp:\/\/host:port\/
-- | | |
-- | +---+---------------- can be configured by user, one host,
-- | port pair per distributed process
-- | instance
-- +---------------------------- In future it will be possible to add other
-- schemas.
-- @
--
-- Transport specifies host that will be used, and port will be
-- automatically generated by the transport for each endpoint.
--
-- * Connections Reliability
--
-- Currently only reliable connections are supported. In case 'Unreliable' was
-- passed the connection will nonetheless be reliable, since it is not incorrect
-- to do so. This may be changed in future versions.
--
-- Network-transport-zeromq maintains one thread for each endpoint that is
-- used to read incoming requests. All sends and connection requests are
-- handled from the user thread and are mostly asynchronous.
--
-- Each endpoint obtains one pull socket and one push socket for each remote end
-- point. All lightweight threads abrigged into one heavyweight connection.
--
-- Virually connections looks like the following plot:
--
-- @
-- +------------------+ +--------------+
-- | Endpoint | | Endpoint |
-- | | | |
-- | +------+ +------+ |
-- | | pull |<-----------------------| push | |
-- + +------+ +------+ |
-- | | +--------------------+ | |
-- + +------+/~~~~ connection 1 ~~~~\\+------+ |
-- | | push |~~~~~ connection 2 ~~~~~| pull | |
-- | +------+\\ /+------+ |
-- | | +--------------------+ | |
-- +------------------+ +--------------+
-- @
--
-- Physically ZeroMQ may choose better representations and mapping on a real
-- connections.
--
-- The ZeroMQ library takes care of reliability. It keeps message in a queue
-- such that in case of connection break, it may be restarted with no message
-- loss. So heartbeating procedure should be introduced on the top on
-- network-transport, and 'breakConnection' should be used to notify connection
-- death. However if High Water Mark will be reached connection to endpoint
-- considered failed and connection break procedure started automatically.
-- Naming conventions.
-- api* -- functions that can be called by user
-- localEndPoint -- internal functions that are used in endpoints
-- remoteEndPoint -- internal functions that are used in endpoints
--
-- Transport
-- |--- Valid transport itself
-- | |--- newEndpoint
-- | |--- close
-- |--- Closed
--
-- LocaEndPoint
-- RemoteEndPoint
-- Connection
-- Connection closing procedure.
-- Once endpoint is gracefully closed by user or by transport close
-- 'RemoteEndPoint' goes into 'RemoteEndPointClosing' state and sends
-- a message to remote endpoint if link is known as alive, in
-- 'RemoteEndPointClosing' all messages are ignored and 'EndPointClosed'
-- is returned. Uppon a 'EndPointClose' delivery endpoint replies with
-- 'EndPointClose' message, and starts cleanup procedure (marking link
-- as not alive). When endpoint receives 'EndPointCloseOk' message it
-- moved 'RemoteEndPoint' to close state and removes all data structures.
-- Together with marking a endpoint as closed asynchronous timeout is
-- set, and if 'EndPointCloseOk' is not delivered withing that timeperiod
-- 'EndPoint' is closed.
--
-- endpoint-1 endpoint-2
-- 1. mark as closing
-- [RemoteEndPoint:Closing] ----endPointClose-> [remoteEndPoint:Valid]
-- 2. mark as closed
-- <-endPointCloseOk- [remoteEndPoint:Closed]
-- 2. mark as closed
-- [RemoteEndPoint:Closed] 4. cleanup remote end point
--
-- 'EndPoint' can be closed in a two ways: normally and abnormally (in case
-- of exception or invariant vionation on a remove side). If 'EndPoint' is
-- closed normally all opened connections will recevice ConnectionClosed
-- events, otherwise.
--
-- Multicast group schema:
--
-- +--------------+
-- | EndPoint |<----------------+
-- +--------------+ | events
-- | |
-- +------------------+ |
-- | |
-- +------------------------------+
-- | Multicast Group (Server) |
-- +--+ +--+ +--+ +---+
-- | Rep | | Pub | | Sub |
-- +-----+ +-----+ +-----+
-- ^ | ^
-- request | | |
-- +------------+ +--------+ messages
-- | |
-- | +-----------+
-- | | messages
-- +-----+ +-----+
-- | Req | | Sub |
-- +---+ +--+ +--+
-- | Multicast Group |
-- +---------------------+
--
-- Resolve Group automatically subscribes to the group to receive events,
-- however unless 'subscribe' will be called endpoint will not reveive any
-- message. This is done in order to keep ability to notify receiver about
-- multcast group close.
--
-- Current solution incorrectly keeps track of incoming connections. This means
-- that we can't really guarantee that a "close"" message is delivered to every
-- recipient and it disconnected correctly.
-- $internals Internal functions provide additional ZeroMQ specific set of
-- configuration options. This functionality should be used with caution as it
-- may break required socket properties.
-- | Messages.
data ZMQMessage
= MessageConnect !EndPointAddress -- ^ Connection greeting
| MessageInitConnection !EndPointAddress !ConnectionId !Reliability
| MessageInitConnectionOk !EndPointAddress !ConnectionId !ConnectionId
| MessageCloseConnection !ConnectionId
| MessageData !ConnectionId
| MessageEndPointClose !EndPointAddress !Bool
| MessageEndPointCloseOk !EndPointAddress
deriving (Generic)
instance Binary ZMQMessage
data ZMQError
= InvariantViolation String
| IncorrectState String
| ConnectionFailed
| DriverError ZMQ.ZMQError
deriving (Typeable, Show)
instance Exception ZMQError
-- | Create 0MQ based transport.
createTransport :: ZMQParameters
-> ByteString -- ^ Transport address (IP or hostname)
-> IO Transport
createTransport z b = (fmap snd) (createTransportExposeInternals z b)
-- | You should probably not use this function (used for unit testing only)
createTransportExposeInternals
:: ZMQParameters -- ^ Configuration parameters for ZeroMQ
-> ByteString -- ^ Host name or IP address
-> IO (TransportInternals, Transport)
createTransportExposeInternals params host = do
ctx <- ZMQ.context
mtid <- Traversable.sequenceA $
fmap (\(SecurityPlain user pass) -> ZMQ.authManager ctx user pass) $
zmqSecurityMechanism params
transport <- TransportInternals
<$> pure addr
<*> (newMVar =<< mkTransportState ctx mtid)
<*> pure params
return $ (transport, Transport
{ newEndPoint = apiNewEndPoint defaultHints transport
, closeTransport = apiTransportClose transport
})
where
addr = B.concat ["tcp://", host]
-- | Close Transport on the current node.
-- This operation is synchronous.
apiTransportClose :: TransportInternals -> IO ()
apiTransportClose transport = mask_ $ do
old <- swapMVar (transportState transport) TransportClosed
case old of
TransportClosed -> return ()
TransportValid v -> either errorLog return <=< tryZMQ $ do
traverse_ (liftM2 (>>) Async.cancel Async.waitCatch)
(v ^. transportAuth)
traverse_ (apiCloseEndPoint transport)
(v ^. transportEndPoints)
sequence_ =<< atomicModifyIORef' (v ^. transportSockets) (\x -> (IntMap.empty, x))
ZMQ.term (v ^. transportContext)
-- | Creates a new endpoint on the transport specified and applies all hints.
apiNewEndPoint :: Hints -- ^ Hints to apply.
-> TransportInternals -- ^ Internal transport state.
-> IO (Either (TransportError NewEndPointErrorCode) EndPoint)
apiNewEndPoint hints transport = try $ mapZMQException (TransportError NewEndPointFailed . show) $
modifyMVar (transportState transport) $ \case
TransportClosed -> throwM $ TransportError NewEndPointFailed "Transport is closed."
TransportValid i -> do
(ep, chan) <- endPointCreate hints (transportParameters transport)
(i ^. transportContext)
(B8.unpack addr)
let !cntx = i ^. transportContext
return $
( TransportValid
. (transportEndPoints ^: (Map.insert (localEndPointAddress ep) ep))
$ i
, EndPoint
{ receive = atomically $ do
mx <- readTMChan chan
case mx of
Nothing -> error "channel is closed"
Just x -> return x
, address = localEndPointAddress ep
, connect = apiConnect (transportParameters transport) cntx ep
, closeEndPoint = apiCloseEndPoint transport ep
, newMulticastGroup = apiNewMulticastGroup defaultHints transport ep
, resolveMulticastGroup = apiResolveMulticastGroup transport ep
}
)
where
addr = transportAddress transport
-- | Asynchronous operation, shutdown of the remote end point may take a while
apiCloseEndPoint :: TransportInternals
-> LocalEndPoint
-> IO ()
apiCloseEndPoint transport lep = mask_ $ either errorLog return <=< tryZMQ $ do
-- we don't close endpoint here because other threads,
-- should be able to access local endpoint state
old <- readMVar (localEndPointState lep)
case old of
LocalEndPointValid (ValidLocalEndPoint x _ _ threadId _ _) -> do
-- close channel, no events will be received
atomically $ do
writeTMChan x EndPointClosed
closeTMChan x
Async.cancel threadId
void $ Async.waitCatch threadId
LocalEndPointClosed -> return ()
void $ swapMVar (localEndPointState lep) LocalEndPointClosed
modifyMVar_ (transportState transport) $ \case
TransportClosed -> return TransportClosed
TransportValid v -> return
$ TransportValid
. (transportEndPoints ^: (Map.delete (localEndPointAddress lep)))
$ v
endPointCreate :: Hints
-> ZMQParameters
-> Context
-> String
-> IO (LocalEndPoint, TMChan Event)
endPointCreate hints params ctx addr = promoteZMQException $ do
pull <- ZMQ.socket ctx ZMQ.Pull
case zmqSecurityMechanism params of
Nothing -> return ()
Just SecurityPlain{} -> do
ZMQ.setPlainServer True pull
ZMQ.setSendHighWM (ZMQ.restrict (zmqHighWaterMark params)) pull
port <- case hintsPort hints of
Nothing -> ZMQ.bindRandomPort pull addr
Just i -> do ZMQ.bind pull (addr ++ ":" ++ show i)
return i
chOut <- newTMChanIO
lep <- LocalEndPoint <$> pure (EndPointAddress $ B8.pack (addr ++ ":" ++ show port))
<*> newEmptyMVar
<*> pure port
opened <- newIORef True
mask $ \restore -> do
thread <- Async.async $ (restore (receiver pull lep chOut))
`finally` finalizeEndPoint lep port pull
putMVar (localEndPointState lep) $ LocalEndPointValid
(ValidLocalEndPoint chOut (Counter 0 Map.empty) Map.empty thread opened Map.empty)
return (lep, chOut)
where
finalizer pull ourEp = forever $ do
(cmd:_) <- ZMQ.receiveMulti pull
mask_ $ case decode' cmd of
MessageEndPointCloseOk theirAddress -> getRemoteEndPoint ourEp theirAddress >>=
traverse_ (\rep -> do
state <- swapMVar (remoteEndPointState rep) RemoteEndPointClosed
closeRemoteEndPoint ourEp rep state)
_ -> return () -- XXX: send exception
receiver :: ZMQ.Socket ZMQ.Pull
-> LocalEndPoint
-> TMChan Event
-> IO ()
receiver pull ourEp chan = forever $ mask_ $ do
(cmd:msgs) <- ZMQ.receiveMulti pull
case decode' cmd of
MessageData idx -> atomically $ writeTMChan chan (Received idx msgs)
MessageConnect theirAddress -> do
void $ createOrGetRemoteEndPoint params ctx ourEp theirAddress
MessageInitConnection theirAddress theirId rel -> do
join $ do
modifyMVar (localEndPointState ourEp) $ \case
LocalEndPointValid v ->
case v ^. localEndPointRemoteAt theirAddress of
Nothing -> return (LocalEndPointValid v, throwM $ InvariantViolation "Remote endpoint should exist.")
Just rep -> modifyMVar (remoteEndPointState rep) $ \case
RemoteEndPointFailed -> throwM $ InvariantViolation "RemoteEndPoint should be valid."
RemoteEndPointClosed -> throwM $ InvariantViolation "RemoteEndPoint should be valid."
RemoteEndPointClosing{} -> throwM $ InvariantViolation "RemoteEndPoint should be valid."
w@RemoteEndPointValid{} -> do
conn <- ZMQConnection <$> pure ourEp
<*> pure rep
<*> pure rel
<*> newMVar (ZMQConnectionValid $ ValidZMQConnection Nothing i)
<*> newEmptyMVar
w' <- register (succ i) w
return ( w'
, ( LocalEndPointValid (localEndPointConnections ^= Counter (succ i) (Map.insert (succ i) conn m) $ v)
, return ())
)
z@(RemoteEndPointPending w) -> do
conn <- ZMQConnection <$> pure ourEp
<*> pure rep
<*> pure rel
<*> newMVar (ZMQConnectionValid $ ValidZMQConnection Nothing i)
<*> newEmptyMVar
modifyIORef w (\xs -> (register (succ i)) : xs)
return ( z
, ( LocalEndPointValid (localEndPointConnections ^= Counter (succ i) (Map.insert (succ i) conn m) $ v)
, return ())
)
where
(Counter i m) = v ^. localEndPointConnections
_ -> throwM $ InvariantViolation "RemoteEndPoint should be valid."
where
register i RemoteEndPointFailed = do
atomically $ do
writeTMChan chan (ConnectionOpened i rel theirAddress)
writeTMChan chan (ConnectionClosed i)
return RemoteEndPointFailed
register i RemoteEndPointClosed = do
atomically $ do
writeTMChan chan (ConnectionOpened i rel theirAddress)
writeTMChan chan (ConnectionClosed i)
return RemoteEndPointClosed
register i (RemoteEndPointClosing x) = do
atomically $ do
writeTMChan chan (ConnectionOpened i rel theirAddress)
writeTMChan chan (ConnectionClosed i)
return $ RemoteEndPointClosing x
register _ RemoteEndPointPending{} = throwM $ InvariantViolation "RemoteEndPoint can't be pending."
register i (RemoteEndPointValid v@(ValidRemoteEndPoint sock _ s _)) = do
ZMQ.send sock [] $ encode' (MessageInitConnectionOk ourAddr theirId i)
atomically $ writeTMChan chan (ConnectionOpened i rel theirAddress)
return $ RemoteEndPointValid
v{_remoteEndPointIncommingConnections = Set.insert i s}
MessageCloseConnection idx -> join $ do
modifyMVar (localEndPointState ourEp) $ \case
LocalEndPointValid v ->
case v ^. localEndPointConnectionAt idx of
Nothing -> return (LocalEndPointValid v, return ())
Just conn -> do
old <- swapMVar (connectionState conn) ZMQConnectionFailed
return ( LocalEndPointValid (localEndPointConnections ^: counterValues ^: (Map.delete idx) $ v)
, case old of
ZMQConnectionFailed -> return ()
ZMQConnectionInit -> return () -- throwM InvariantViolation
ZMQConnectionClosed -> return ()
ZMQConnectionValid (ValidZMQConnection _ _) -> do
atomically $ writeTMChan chan (ConnectionClosed idx)
connectionCleanup (connectionRemoteEndPoint conn) idx)
LocalEndPointClosed -> return (LocalEndPointClosed, return ())
MessageInitConnectionOk theirAddress ourId theirId -> do
join $ withMVar (localEndPointState ourEp) $ \case
LocalEndPointValid v ->
case v ^. localEndPointRemoteAt theirAddress of
Nothing -> return (return ()) -- XXX: send message to the host
Just rep -> modifyMVar (remoteEndPointState rep) $ \case
RemoteEndPointFailed -> return (RemoteEndPointFailed, return ())
RemoteEndPointClosed -> throwM $ InvariantViolation "RemoteEndPoint should be valid or failed."
RemoteEndPointClosing{} -> throwM $ InvariantViolation "RemoteEndPoint should be valid or failed."
t@(RemoteEndPointValid (ValidRemoteEndPoint sock (Counter x m) s z)) -> return $
case ourId `Map.lookup` m of
Nothing -> (t, return ()) -- XXX: send message to the hostv
Just c ->
(RemoteEndPointValid (ValidRemoteEndPoint sock (Counter x (ourId `Map.delete` m)) s (z+1))
, do
modifyMVar_ (connectionState c) $ \case
ZMQConnectionFailed -> return ZMQConnectionFailed
ZMQConnectionInit -> return $ ZMQConnectionValid (ValidZMQConnection (Just sock) theirId)
ZMQConnectionClosed -> do
ZMQ.send sock [] $ encode' (MessageCloseConnection theirId)
return ZMQConnectionClosed
ZMQConnectionValid _ -> throwM $ InvariantViolation "RemoteEndPoint should be closed"
void $ tryPutMVar (connectionReady c) ()
)
RemoteEndPointPending p -> return (RemoteEndPointPending p, throwM $ InvariantViolation "RemoteEndPoint should be closed")
LocalEndPointClosed -> return $ return ()
MessageEndPointClose theirAddress True -> getRemoteEndPoint ourEp theirAddress >>=
traverse_ (\rep -> do
onValidRemote rep $ \v ->
ZMQ.send (remoteEndPointSocket v) [] (encode' $ MessageEndPointCloseOk $ localEndPointAddress ourEp)
remoteEndPointClose True ourEp rep)
MessageEndPointClose theirAddress False -> getRemoteEndPoint ourEp theirAddress >>=
traverse_ (\rep -> do
mst <- cleanupRemoteEndPoint ourEp rep Nothing
case mst of
Nothing -> return ()
Just st -> do
onValidEndPoint ourEp $ \v -> atomically $ writeTMChan (localEndPointChan v) $
ErrorEvent $ TransportError (EventConnectionLost theirAddress) "Exception on remote side"
closeRemoteEndPoint ourEp rep st)
MessageEndPointCloseOk theirAddress -> getRemoteEndPoint ourEp theirAddress >>=
traverse_ (\rep -> do
state <- swapMVar (remoteEndPointState rep) RemoteEndPointClosed
closeRemoteEndPoint ourEp rep state)
where
ourAddr = localEndPointAddress ourEp
finalizeEndPoint ourEp _port pull = do
join $ withMVar (localEndPointState ourEp) $ \case
LocalEndPointClosed -> afterP ()
LocalEndPointValid v -> do
writeIORef (localEndPointOpened v) False
return $ do
tid <- Async.async $ finalizer pull ourEp
void $ Async.mapConcurrently (remoteEndPointClose False ourEp)
$ v ^. localEndPointRemotes
Async.cancel tid
void $ Async.waitCatch tid
ZMQ.closeZeroLinger pull
void $ swapMVar (localEndPointState ourEp) LocalEndPointClosed
apiSend :: ZMQConnection -> [ByteString] -> IO (Either (TransportError SendErrorCode) ())
apiSend (ZMQConnection l e _ s _) b = do
eb <- try $ mapM_ evaluate b
case eb of
-- Tests in network-transport-tests require to convert exceptions
-- during evaluation of the argument to returned error values.
Left ex -> do cleanup
return $ Left $ TransportError SendFailed (show (ex::SomeException))
Right _ -> (fmap Right inner) `catches`
[ Handler $ \ex -> -- TransportError - return, as all require
-- actions were performed
return $ Left (ex :: TransportError SendErrorCode)
, Handler $ \ex -> do -- ZMQError appeared exception
cleanup
return $ Left $ TransportError SendFailed (show (ex::ZMQError))
]
where
inner :: IO ()
inner = join $ withMVar (remoteEndPointState e) $ \x -> case x of
RemoteEndPointPending{} -> return $ yield >> inner
RemoteEndPointFailed -> throwIO $ TransportError SendFailed "Remote end point is failed."
RemoteEndPointClosed -> throwIO $ TransportError SendFailed "Remote end point is closed."
RemoteEndPointClosing{} -> throwIO $ TransportError SendFailed "Remote end point is closing."
RemoteEndPointValid v -> withMVar s $ \case
ZMQConnectionInit -> return $ yield >> inner -- readMVar (connectionReady c) >> inner
ZMQConnectionClosed -> throwIO $ TransportError SendClosed "Connection is closed"
ZMQConnectionFailed -> throwIO $ TransportError SendFailed "Connection is failed"
ZMQConnectionValid (ValidZMQConnection _ idx) -> do
evs <- ZMQ.events (remoteEndPointSocket v)
if ZMQ.Out `elem` evs
then do ZMQ.sendMulti (remoteEndPointSocket v) $ encode' (MessageData idx) :| b
afterP ()
else return $ do
rep <- cleanupRemoteEndPoint l e Nothing
traverse_ (\z -> do
onValidEndPoint l $ \w -> atomically $ writeTMChan (localEndPointChan w) $
ErrorEvent $ TransportError (EventConnectionLost (remoteEndPointAddress e)) "Exception on remote side"
closeRemoteEndPoint l e z) rep
throwIO $ TransportError SendFailed "Connection broken."
cleanup = do
void $ cleanupRemoteEndPoint l e
(Just $ \v -> ZMQ.send (remoteEndPointSocket v) [] $ encode' (MessageEndPointClose (localEndPointAddress l) False))
onValidEndPoint l $ \v -> atomically $ do
writeTMChan (localEndPointChan v) $ ErrorEvent $ TransportError
(EventConnectionLost (remoteEndPointAddress e)) "Exception on send."
-- 'apiClose' function is asynchronous, as connection may not exists by the
-- time of the calling to this function. In this case function just marks
-- connection as closed, so all subsequent calls from the user side will
-- "think" that the connection is closed.
-- only after connection will be up.
apiClose :: ZMQConnection -> IO ()
apiClose (ZMQConnection _ e _ s _) = either errorLog return <=< tryZMQ $ uninterruptibleMask_ $ join $ do
modifyMVar s $ \case
ZMQConnectionValid (ValidZMQConnection _ idx) -> do
return (ZMQConnectionClosed, do
modifyMVar_ (remoteEndPointState e) $ \case
v@RemoteEndPointValid{} -> notify idx v
v@(RemoteEndPointPending p) -> modifyIORef p (\xs -> notify idx : xs) >> return v
v -> return v
)
_ -> return (ZMQConnectionClosed, return ())
where
notify _ RemoteEndPointFailed = return RemoteEndPointFailed
notify _ (RemoteEndPointClosing x) = return $ RemoteEndPointClosing x
notify _ RemoteEndPointClosed = return RemoteEndPointClosed
notify _ RemoteEndPointPending{} = throwM $ InvariantViolation "RemoteEndPoint can't be pending."
notify idx w@(RemoteEndPointValid (ValidRemoteEndPoint sock _ _ _)) = do
ZMQ.send sock [] $ encode' (MessageCloseConnection idx)
return w
apiConnect :: ZMQParameters
-> Context
-> LocalEndPoint
-> EndPointAddress
-> Reliability
-> ConnectHints
-> IO (Either (TransportError ConnectErrorCode) Connection)
apiConnect params ctx ourEp theirAddr reliability _hints = fmap (either Left id) $ try $
mapZMQException (TransportError ConnectFailed . show) $ do
eRep <- createOrGetRemoteEndPoint params ctx ourEp theirAddr
case eRep of
Left{} -> return $ Left $ TransportError ConnectFailed "LocalEndPoint is closed."
Right rep -> do
conn <- ZMQConnection <$> pure ourEp
<*> pure rep
<*> pure reliability
<*> newMVar ZMQConnectionInit
<*> newEmptyMVar
let apiConn = Connection
{ send = apiSend conn
, close = apiClose conn
}
join $ modifyMVar (remoteEndPointState rep) $ \w -> case w of
RemoteEndPointClosed -> do
return ( RemoteEndPointClosed
, return $ Left $ TransportError ConnectFailed "Transport is closed.")
RemoteEndPointClosing x -> do
return ( RemoteEndPointClosing x
, return $ Left $ TransportError ConnectFailed "RemoteEndPoint closed.")
RemoteEndPointValid _ -> do
s' <- go conn w
return (s', waitReady conn apiConn)
RemoteEndPointPending z -> do
modifyIORef z (\zs -> go conn : zs)
return ( RemoteEndPointPending z, waitReady conn apiConn)
RemoteEndPointFailed ->
return ( RemoteEndPointFailed
, return $ Left $ TransportError ConnectFailed "RemoteEndPoint failed.")
where
waitReady conn apiConn = join $ withMVar (connectionState conn) $ \case
ZMQConnectionInit{} -> return $ yield >> waitReady conn apiConn
ZMQConnectionValid{} -> afterP $ Right apiConn
ZMQConnectionFailed{} -> afterP $ Left $ TransportError ConnectFailed "Connection failed."
ZMQConnectionClosed{} -> throwM $ InvariantViolation "Connection closed."
ourAddr = localEndPointAddress ourEp
go conn s = inner conn s
inner _ RemoteEndPointClosed = return RemoteEndPointClosed
inner _ RemoteEndPointPending{} = throwM $ InvariantViolation "Connection pending."
inner _ (RemoteEndPointClosing x) = return $ RemoteEndPointClosing x
inner _ RemoteEndPointFailed = return RemoteEndPointFailed
inner conn (RemoteEndPointValid (ValidRemoteEndPoint sock (Counter i m) s z)) = do
ZMQ.send sock [] $ encode' (MessageInitConnection ourAddr i' reliability)
return $ RemoteEndPointValid (ValidRemoteEndPoint sock (Counter i' (Map.insert i' conn m)) s z)
where i' = succ i
getRemoteEndPoint :: LocalEndPoint -> EndPointAddress -> IO (Maybe RemoteEndPoint)
getRemoteEndPoint ourEp theirAddr = do
withMVar (localEndPointState ourEp) $ \case
LocalEndPointValid v -> return $ theirAddr `Map.lookup` (_localEndPointRemotes v)
LocalEndPointClosed -> return Nothing
createOrGetRemoteEndPoint :: ZMQParameters
-> Context
-> LocalEndPoint
-> EndPointAddress
-> IO (Either ZMQError RemoteEndPoint)
createOrGetRemoteEndPoint params ctx ourEp theirAddr = join $ do
modifyMVar (localEndPointState ourEp) $ \case
LocalEndPointValid v@(ValidLocalEndPoint _ _ _ _ o _) -> do
opened <- readIORef o
if opened
then do
case v ^. localEndPointRemoteAt theirAddr of
Nothing -> create v
Just rep -> do
withMVar (remoteEndPointState rep) $ \case
RemoteEndPointFailed -> create v
_ -> return (LocalEndPointValid v, return $ Right rep)
else return (LocalEndPointValid v, return $ Left $ IncorrectState "EndPointClosing")
LocalEndPointClosed ->
return ( LocalEndPointClosed
, return $ Left $ IncorrectState "EndPoint is closed"
)
where
create v = do
push <- ZMQ.socket ctx ZMQ.Push
case zmqSecurityMechanism params of
Nothing -> return ()
Just (SecurityPlain p u) -> do
ZMQ.setPlainPassword (ZMQ.restrict p) push
ZMQ.setPlainUserName (ZMQ.restrict u) push
state <- newMVar . RemoteEndPointPending =<< newIORef []
opened <- newIORef False
let rep = RemoteEndPoint theirAddr state opened
return ( LocalEndPointValid
. (localEndPointRemotes ^: (Map.insert theirAddr rep))
$ v
, initialize push rep >> return (Right rep))
ourAddr = localEndPointAddress ourEp
initialize push rep = do
lock <- newEmptyMVar
x <- Async.async $ do
takeMVar lock
ZMQ.connect push (B8.unpack $ endPointAddressToByteString theirAddr)
monitor <- ZMQ.monitor [ZMQ.ConnectedEvent] ctx push
putMVar lock ()
r <- Async.race (threadDelay 100000) (monitor True)
void $ monitor False
case r of
Left _ -> do
_ <- swapMVar (remoteEndPointState rep) RemoteEndPointFailed
Async.cancel x
ZMQ.closeZeroLinger push
Right _ -> do
ZMQ.send push [] $ encode' $ MessageConnect ourAddr
let v = ValidRemoteEndPoint push (Counter 0 Map.empty) Set.empty 0
modifyMVar_ (remoteEndPointState rep) $ \case
RemoteEndPointPending p -> do
z <- foldM (\y f -> f y) (RemoteEndPointValid v) . Prelude.reverse =<< readIORef p
modifyIORef (remoteEndPointOpened rep) (const True)
return z
RemoteEndPointValid _ -> throwM $ InvariantViolation "RemoteEndPoint valid."
RemoteEndPointClosed -> return RemoteEndPointClosed
RemoteEndPointClosing j -> return (RemoteEndPointClosing j)
RemoteEndPointFailed -> return RemoteEndPointFailed
-- | Close all endpoint connections, return previous state in case
-- if it was alive, for further cleanup actions.
cleanupRemoteEndPoint :: LocalEndPoint
-> RemoteEndPoint
-> (Maybe (ValidRemoteEndPoint -> IO ()))
-> IO (Maybe RemoteEndPointState)
cleanupRemoteEndPoint lep rep actions = modifyMVar (localEndPointState lep) $ \case
LocalEndPointValid v -> do
modifyIORef (remoteEndPointOpened rep) (const False)
oldState <- swapMVar (remoteEndPointState rep) newState
case oldState of
RemoteEndPointValid w -> do
let cn = w ^. (remoteEndPointPendingConnections >>> counterValues)
traverse_ (\c -> void $ swapMVar (connectionState c) ZMQConnectionFailed) cn
cn' <- foldM
(\c' idx -> case idx `Map.lookup` cn of
Nothing -> return c'
Just c -> do
void $ swapMVar (connectionState c) ZMQConnectionFailed
return $ (counterValues ^: (Map.delete idx)) $ c'
)
(v ^. localEndPointConnections)
(Set.toList (_remoteEndPointIncommingConnections w))
case actions of
Nothing -> return ()
Just f -> f w
return ( LocalEndPointValid (localEndPointConnections ^= cn' $ v)
, Just oldState)
_ -> return (LocalEndPointValid v, Nothing)
c -> return (c, Nothing)
where
newState = RemoteEndPointFailed
-- | Close, all network connections.
closeRemoteEndPoint :: LocalEndPoint -> RemoteEndPoint -> RemoteEndPointState -> IO ()
closeRemoteEndPoint lep rep state = step1 >> step2 state
where
step1 = modifyMVar_ (localEndPointState lep) $ \case
LocalEndPointValid v -> return $
LocalEndPointValid
. (localEndPointRemotes ^: (Map.delete (remoteEndPointAddress rep)))
$ v
c -> return c
step2 (RemoteEndPointValid v) = do
ZMQ.closeZeroLinger (remoteEndPointSocket v)
step2 (RemoteEndPointClosing (ClosingRemoteEndPoint sock rd)) = do
_ <- readMVar rd
ZMQ.closeZeroLinger sock
step2 _ = return ()
remoteEndPointClose :: Bool -> LocalEndPoint -> RemoteEndPoint -> IO ()
remoteEndPointClose silent lep rep = do
modifyIORef (remoteEndPointOpened rep) (const False)
join $ modifyMVar (remoteEndPointState rep) $ \o -> case o of
RemoteEndPointFailed -> return (o, return ())
RemoteEndPointClosed -> return (o, return ())
RemoteEndPointClosing (ClosingRemoteEndPoint _ l) -> return (o, void $ readMVar l)
RemoteEndPointPending _ -> closing (error "Pending actions should not be executed") o -- XXX: store socket, or delay
RemoteEndPointValid v -> closing (remoteEndPointSocket v) o
where
closing sock old = do
lock <- newEmptyMVar
return (RemoteEndPointClosing (ClosingRemoteEndPoint sock lock), go lock old)
go lock old@(RemoteEndPointValid (ValidRemoteEndPoint c _ s i)) = do
-- close all connections
void $ cleanupRemoteEndPoint lep rep Nothing
withMVar (localEndPointState lep) $ \case
LocalEndPointClosed -> return ()
LocalEndPointValid v -> do
-- notify about all connections close (?) do we really want it?
traverse_ (atomically . writeTMChan (localEndPointChan v) . ConnectionClosed) (Set.toList s)
-- if we have outgoing connections, then we have connection error
when (i > 0) $ atomically
$ writeTMChan (localEndPointChan v)
$ ErrorEvent $ TransportError (EventConnectionLost (remoteEndPointAddress rep)) "Remote end point closed."
-- notify other side about closing connection
unless silent $ do
ZMQ.send c [] (encode' (MessageEndPointClose (localEndPointAddress lep) True))
yield
void $ Async.race (readMVar lock) (threadDelay 1000000)
void $ tryPutMVar lock ()
closeRemoteEndPoint lep rep old
return ()
go _ _ = return ()
connectionCleanup :: RemoteEndPoint -> ConnectionId -> IO ()
connectionCleanup rep cid = modifyMVar_ (remoteEndPointState rep) $ \case
RemoteEndPointValid v -> return $
RemoteEndPointValid v{_remoteEndPointIncommingConnections = Set.delete cid (_remoteEndPointIncommingConnections v)}
c -> return c
encode' :: Binary a => a -> ByteString
encode' = B.concat . BL.toChunks . encode
decode' :: Binary a => ByteString -> a
decode' s = decode . BL.fromChunks $ [s]
onValidEndPoint :: LocalEndPoint -> (ValidLocalEndPoint -> IO ()) -> IO ()
onValidEndPoint lep f = withMVar (localEndPointState lep) $ \case
LocalEndPointValid v -> f v
_ -> return ()
onValidRemote :: RemoteEndPoint -> (ValidRemoteEndPoint -> IO ()) -> IO ()
onValidRemote rep f = withMVar (remoteEndPointState rep) $ \case
RemoteEndPointValid v -> f v
_ -> return ()
afterP :: a -> IO (IO a)
afterP = return . return
tryZMQ :: (MonadCatch m) => m a -> m (Either ZMQError a)
tryZMQ = try . promoteZMQException
mapZMQException :: (Exception e) => (ZMQError -> e) -> a -> a
mapZMQException = mapException
-- | Break endpoint connection, all endpoints that will be affected.
breakConnection :: TransportInternals
-> EndPointAddress
-> EndPointAddress
-> IO ()
breakConnection zmqt _from to = Foldable.sequence_ <=< withMVar (transportState zmqt) $ \case
TransportValid v -> Traversable.forM (v ^. transportEndPoints) $ \x ->
withMVar (localEndPointState x) $ \case
LocalEndPointValid w -> return $ Foldable.sequence_ $ flip Map.mapWithKey (w ^. localEndPointRemotes) $ \key rep ->
if onDeadHost key
then do
mz <- cleanupRemoteEndPoint x rep Nothing
flip traverse_ mz $ \z -> do
atomically $ writeTMChan (localEndPointChan w) $
ErrorEvent $ TransportError (EventConnectionLost key) "Manual connection break"
closeRemoteEndPoint x rep z
else return ()
LocalEndPointClosed -> afterP ()
TransportClosed -> return Map.empty
where
dh = B8.init . fst $ B8.spanEnd (/=':') (endPointAddressToByteString to)
onDeadHost = B8.isPrefixOf dh . endPointAddressToByteString
-- | Break connection between two endpoints, other endpoints on the same
-- remove will not be affected.
breakConnectionEndPoint :: TransportInternals
-> EndPointAddress
-> EndPointAddress
-> IO ()
breakConnectionEndPoint zmqt from to = one from to >> one to from
where
one f t = join $ withMVar (transportState zmqt) $ \case
TransportValid v -> case v ^. transportEndPointAt f of
Nothing -> afterP ()
Just x -> withMVar (localEndPointState x) $ \case
LocalEndPointValid w -> case w ^. localEndPointRemoteAt t of
Nothing -> afterP ()
Just y -> return $ do
mz <- cleanupRemoteEndPoint x y Nothing
case mz of
Nothing -> return ()
Just z -> do
onValidEndPoint x $ \j -> atomically $ writeTMChan (localEndPointChan j) $
ErrorEvent $ TransportError (EventConnectionLost to) "Exception on remote side"
closeRemoteEndPoint x y z
LocalEndPointClosed -> afterP ()
TransportClosed -> afterP ()
-- | Configure socket after creation
unsafeConfigurePush :: TransportInternals
-> EndPointAddress
-> EndPointAddress
-> (ZMQ.Socket ZMQ.Push -> IO ())
-> IO ()
unsafeConfigurePush zmqt from to f = withMVar (transportState zmqt) $ \case
TransportValid v -> Foldable.traverse_
(\x -> withMVar (localEndPointState x) $ \case
LocalEndPointValid w -> case w ^. localEndPointRemoteAt to of
Nothing -> return ()
Just y -> onValidRemote y $ f . remoteEndPointSocket
LocalEndPointClosed -> return ()
) (v ^. transportEndPointAt from)
TransportClosed -> return ()
-- | Create a new multicast group associated with an EndPoint.
--
-- For multicast addresses zeromq uses two sockets. Is uses Pub-Sub sockets for multicast
-- delivery and one Req-Rep socket for sending messages to the EndPoint. That
-- endpoint will retransmit messages to all subscribers.
--
-- If Hints are used to specify ports then the address will have the form:
--
-- > host:Port:ControlPort
--
-- where Port is 'hintsPort' and ControlPort is 'hintsControlPort'. If the hint port
-- is not specified then random ports will be used.
apiNewMulticastGroup :: Hints -- ^ Multicast group hints.
-> TransportInternals -- ^ Internal transport state.
-> LocalEndPoint -- ^ EndPoint that is associated with the multicast group.
-> IO ( Either (TransportError NewMulticastGroupErrorCode) MulticastGroup)
apiNewMulticastGroup hints zmq lep = withMVar (transportState zmq) $ \case
TransportClosed -> return $ Left $ TransportError NewMulticastGroupFailed "Transport is closed."
TransportValid vt -> modifyMVar (localEndPointState lep) $ \case
LocalEndPointClosed -> return (LocalEndPointClosed, Left $ TransportError NewMulticastGroupFailed "Transport is closed.")
LocalEndPointValid vl -> mask_ $ do
(pub, portPub, rep, portRep, wrkThread) <- mkPublisher vt
let addr = MulticastAddress $
B8.concat [transportAddress zmq,":",B8.pack (show portPub), ":",B8.pack (show portRep)]
subAddr = extractPubAddress addr
repAddr = extractRepAddress addr
-- subscriber api
sub <- ZMQ.socket (vt ^. transportContext) ZMQ.Sub
ZMQ.connect sub (B8.unpack subAddr)
ZMQ.subscribe sub ""
subscribed <- newIORef False
(subThread, mgroup) <- fixIO $ \ ~(tid,_) -> do
v <- newMVar (MulticastGroupValid (ValidMulticastGroup subscribed))
let mgroup = ZMQMulticastGroup v
(apiDeleteMulticastGroupLocal v lep addr rep repAddr pub sub subAddr (Just tid) wrkThread)
tid' <- Async.async $ do
repeatWhile $ do
(ctrl: msg) <- ZMQ.receiveMulti sub
if B8.null ctrl
then do opened <- readIORef subscribed
when opened $ atomically $ writeTMChan (localEndPointChan vl)
$ ReceivedMulticast addr msg
return True
else return False
return (tid', mgroup)
return ( LocalEndPointValid vl
, Right $ MulticastGroup
{ multicastAddress = addr
, deleteMulticastGroup = apiDeleteMulticastGroupLocal (multicastGroupState mgroup) lep addr rep repAddr pub sub subAddr (Just subThread) wrkThread
, maxMsgSize = Nothing
, multicastSend = apiMulticastSendLocal mgroup pub
, multicastSubscribe = apiMulticastSubscribe mgroup
, multicastUnsubscribe = apiMulticastUnsubscribe mgroup
, multicastClose = apiMulticastClose
}
)
where
mkPublisher vt = do
pub <- ZMQ.socket (vt^.transportContext) ZMQ.Pub
portPub <- case hintsPort hints of
Nothing -> ZMQ.bindRandomPort pub (B8.unpack $ transportAddress zmq)
Just i -> do ZMQ.bind pub (B8.unpack (transportAddress zmq) ++ ":" ++ show i)
return i
rep <- ZMQ.socket (vt^.transportContext) ZMQ.Rep
portRep <- case hintsControlPort hints of
Nothing -> ZMQ.bindRandomPort rep (B8.unpack $ transportAddress zmq)
Just i -> do ZMQ.bind rep (B8.unpack (transportAddress zmq) ++ ":" ++ show i)
return i
wrkThread <- Async.async $ forever $ do
msg <- ZMQ.receiveMulti rep
ZMQ.sendMulti pub $ "" :| msg
ZMQ.send rep [] "OK"
return (pub,portPub,rep,portRep, wrkThread)
apiResolveMulticastGroup :: TransportInternals -> LocalEndPoint -> MulticastAddress -> IO (Either (TransportError ResolveMulticastGroupErrorCode) MulticastGroup)
apiResolveMulticastGroup zmq lep addr = withMVar (transportState zmq) $ \case
TransportClosed -> return $ Left $ TransportError ResolveMulticastGroupFailed "Transport is closed."
TransportValid vt -> modifyMVar (localEndPointState lep) $ \case
LocalEndPointClosed -> return (LocalEndPointClosed, Left $ TransportError ResolveMulticastGroupFailed "Transport is closed.")
LocalEndPointValid vl -> mask_ $ do
-- socket allocation
req <- ZMQ.socket (vt^.transportContext) ZMQ.Req
ZMQ.connect req (B8.unpack reqAddr)
sub <- ZMQ.socket (vt^.transportContext) ZMQ.Sub
ZMQ.connect sub (B8.unpack subAddr)
ZMQ.subscribe sub ""
subscribed <- newIORef False
(subThread, mgroup) <- fixIO $ \ ~(tid,_) -> do
v <- newMVar (MulticastGroupValid (ValidMulticastGroup subscribed))
let mgroup = ZMQMulticastGroup v
(apiDeleteMulticastGroupRemote v lep addr req reqAddr sub subAddr (Just tid))
tid' <- Async.async $ repeatWhile $ do
(ctrl: msg) <- ZMQ.receiveMulti sub
if B8.null ctrl
then do opened <- readIORef subscribed
when opened $ atomically $ writeTMChan (localEndPointChan vl)
$ ReceivedMulticast addr msg
return True
else do apiDeleteMulticastGroupRemote v lep addr req reqAddr sub subAddr Nothing
return False
return (tid', mgroup)
return ( LocalEndPointValid
vl{_localEndPointMulticastGroups = Map.insert addr mgroup (_localEndPointMulticastGroups vl)}
, Right $ MulticastGroup
{ multicastAddress = addr
, deleteMulticastGroup =
apiDeleteMulticastGroupRemote (multicastGroupState mgroup) lep addr req reqAddr sub subAddr (Just subThread)
, maxMsgSize = Nothing
, multicastSend = apiMulticastSendRemote mgroup req
, multicastSubscribe = apiMulticastSubscribe mgroup
, multicastUnsubscribe = apiMulticastUnsubscribe mgroup
, multicastClose = apiMulticastClose
}
)
where
reqAddr = extractRepAddress addr
subAddr = extractPubAddress addr
apiDeleteMulticastGroupRemote :: MVar MulticastGroupState -> LocalEndPoint -> MulticastAddress
-> ZMQ.Socket ZMQ.Req -> ByteString -> ZMQ.Socket ZMQ.Sub -> ByteString -> Maybe (Async.Async ())
-> IO ()
apiDeleteMulticastGroupRemote mstate lep addr req _reqAddr sub _subAddr mtid = mask_ $ do
Foldable.traverse_ (\tid -> Async.cancel tid >> void (Async.waitCatch tid)) mtid
modifyMVar_ mstate $ \case
MulticastGroupClosed -> return MulticastGroupClosed
MulticastGroupValid v -> do
modifyIORef (_multicastGroupSubscribed v) (const False)
ZMQ.closeZeroLinger req
ZMQ.unsubscribe sub ""
ZMQ.closeZeroLinger sub
return MulticastGroupClosed
modifyMVar_ (localEndPointState lep) $ \case
LocalEndPointClosed -> return LocalEndPointClosed
LocalEndPointValid v -> return $ LocalEndPointValid
v{_localEndPointMulticastGroups = Map.delete addr (_localEndPointMulticastGroups v)}
apiDeleteMulticastGroupLocal :: MVar MulticastGroupState -> LocalEndPoint -> MulticastAddress
-> ZMQ.Socket ZMQ.Rep -> ByteString -> ZMQ.Socket ZMQ.Pub -> ZMQ.Socket ZMQ.Sub -> ByteString -> Maybe (Async.Async ())
-> Async.Async ()
-> IO ()
apiDeleteMulticastGroupLocal mstate lep addr rep _repAddr pub sub _pubAddr mtid wrk = mask_ $ do
Foldable.traverse_ (\tid -> Async.cancel tid >> void (Async.waitCatch tid)) mtid
void $ Async.cancel wrk >> Async.waitCatch wrk
modifyMVar_ mstate $ \case
MulticastGroupClosed -> return MulticastGroupClosed
MulticastGroupValid v -> do
modifyIORef (_multicastGroupSubscribed v) (const False)
ZMQ.sendMulti pub $ "C" :| []
threadDelay 50000
ZMQ.closeZeroLinger rep
ZMQ.unsubscribe sub ""
ZMQ.closeZeroLinger sub
ZMQ.closeZeroLinger pub
return MulticastGroupClosed
modifyMVar_ (localEndPointState lep) $ \case
LocalEndPointClosed -> return LocalEndPointClosed
LocalEndPointValid v -> return $ LocalEndPointValid
v{_localEndPointMulticastGroups = Map.delete addr (_localEndPointMulticastGroups v)}
apiMulticastSendLocal :: (ZMQ.Sender a) => ZMQMulticastGroup -> ZMQ.Socket a -> [ByteString] -> IO ()
apiMulticastSendLocal mgroup sock msg = withMVar (multicastGroupState mgroup) $ \case
MulticastGroupClosed -> return ()
MulticastGroupValid _ -> do
ZMQ.sendMulti sock $ "" :| msg
apiMulticastSendRemote :: (ZMQ.Sender a, ZMQ.Receiver a) => ZMQMulticastGroup -> ZMQ.Socket a -> [ByteString] -> IO ()
apiMulticastSendRemote _ _ [] = return ()
apiMulticastSendRemote mgroup sock (m:msg) = withMVar (multicastGroupState mgroup) $ \case
MulticastGroupClosed -> return ()
MulticastGroupValid _ -> do
ZMQ.sendMulti sock $ m :| msg
"OK" <- ZMQ.receive sock -- XXX: timeout
return ()
apiMulticastSubscribe :: ZMQMulticastGroup -> IO ()
apiMulticastSubscribe mgroup = withMVar (multicastGroupState mgroup) $ \case
MulticastGroupClosed -> return ()
MulticastGroupValid v -> do
modifyIORef (_multicastGroupSubscribed v) (const True)
apiMulticastUnsubscribe :: ZMQMulticastGroup -> IO ()
apiMulticastUnsubscribe mgroup = withMVar (multicastGroupState mgroup) $ \case
MulticastGroupClosed -> return ()
MulticastGroupValid v -> modifyIORef (_multicastGroupSubscribed v) (const False)
apiMulticastClose :: IO ()
apiMulticastClose = return ()
-- $cleanup
-- Cleanup API is prepared to store cleanup actions for example socket
-- close for objects that uses network-transport-zeromq resources. Theese
-- actions will be fired on transport close.
-- If you need to clean resource beforehead use 'applyCleanupAction'.
-- | Register action that will be fired when transport will be closed.
registerCleanupAction :: TransportInternals -> IO () -> IO (Maybe Unique)
registerCleanupAction zmq fn = withMVar (transportState zmq) $ \case
TransportValid v -> registerValidCleanupAction v fn
TransportClosed -> return Nothing
-- | Register action on a locked transport.
registerValidCleanupAction :: ValidTransportState -> IO () -> IO (Maybe Unique)
registerValidCleanupAction v fn = Just <$> do
u <- newUnique
atomicModifyIORef' (v ^. transportSockets)
(\m -> (IntMap.insert (hashUnique u) fn m, u))
-- | Call cleanup action before transport close.
applyCleanupAction :: TransportInternals -> Unique -> IO ()
applyCleanupAction zmq u = withMVar (transportState zmq) $ \case
TransportValid v -> mask_ $
traverse_ id =<< atomicModifyIORef' (v ^. transportSockets)
(liftA2 (,) (IntMap.delete (hashUnique u))
(IntMap.lookup (hashUnique u)))
TransportClosed -> return ()
extractRepAddress :: MulticastAddress -> ByteString
extractRepAddress (MulticastAddress bs) = B8.concat [a,":",p]
where
(x,p) = B8.spanEnd (/=':') bs
a = B8.init . fst . B8.spanEnd (/=':') . B8.init $ x
extractPubAddress :: MulticastAddress -> ByteString
extractPubAddress (MulticastAddress bs) = B8.init . fst $ B8.spanEnd (/=':') bs
repeatWhile :: Monad m => (m Bool) -> m ()
repeatWhile f = f >>= \x -> if x then repeatWhile f else return ()
-- | Map ZMQ.ZMQError to network-transport ZMQError
promoteZMQException :: a -> a
promoteZMQException = mapException DriverError
-- | Print error to standart output, this function should be used for
-- errors that could not be handled in a normal way.
errorLog :: Show a => a -> IO ()
errorLog s = hPutStrLn stderr (printf "[network-transport-zeromq] Unhandled error: %s" $ show s)
-- $zeromqs
-- network-transport ZeroMQ has a big number of additional options that can be used for socket
-- configuration and that are not exposed in network-transport abstraction layer. In order to
-- to use specific options of network-transport-zeromq, the function 'apiNewEndPoint' was introduced.
-- This function uses 'TransportInternals' and a notion of 'Hints' -- a special data type that contains
-- the possible configuration options.
--
-- __Example: Bootstrapping problem.__
--
-- Due to the network-transport-zeromq design, a new endpoint is bound to a new socket address,
-- making it impossible to bootstrap systems without an additional communication mechanism.
-- This can be avoided if the new function is used to create an endpoint on a specified port:
--
-- > (intenals, transport) <- createTransportExposeInternals defaultZMQParameters "127.0.0.1"
-- > ep <- apiNewEndpoint Hints{hintsPort=8888} internals
--
-- Allow to create a endpoint on a specified port and as a result bootstrap the system.