diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,20 @@ +Copyright (c) 2015 Alfredo Di Napoli + +Permission is hereby granted, free of charge, to any person obtaining +a copy of this software and associated documentation files (the +"Software"), to deal in the Software without restriction, including +without limitation the rights to use, copy, modify, merge, publish, +distribute, sublicense, and/or sell copies of the Software, and to +permit persons to whom the Software is furnished to do so, subject to +the following conditions: + +The above copyright notice and this permission notice shall be included +in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. +IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, +TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE +SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/network-transport-amqp.cabal b/network-transport-amqp.cabal new file mode 100644 --- /dev/null +++ b/network-transport-amqp.cabal @@ -0,0 +1,81 @@ +name: network-transport-amqp +version: 0.1.0.0 +synopsis: AMQP-based transport layer for distributed-process (aka Cloud Haskell) +description: AMQP-based transport layer for distributed-process (aka Cloud Haskell) +license: MIT +license-file: LICENSE +author: Alfredo Di Napoli +maintainer: alfredo.dinapoli@gmail.com +category: Network +build-type: Simple +-- extra-source-files: +cabal-version: >=1.10 + +Source-Repository head + Type: git + Location: https://github.com/adinapoli/network-transport-ampq + +flag distributed-process-tests + description: Build test suites that require distributed-process to be installed. + default: False + +library + exposed-modules: + Network.Transport.AMQP + other-modules: + Network.Transport.AMQP.Internal.Types + -- other-extensions: + build-depends: + base >= 4.0 && < 5.0 + , amqp >= 0.8.3 && < 0.13 + , cereal >= 0.4.0.0 + , async >= 0.2.0 + , string-conv + , stm >= 2.4 + , stm-chans >= 0.3 + , network-transport >= 0.4.0 + , exceptions >= 0.3 + , bytestring < 0.11 + , containers >= 0.5.0.0 + , uuid >= 1.2.6 + , text >= 0.11 + , lens-family + , lens-family-th + + if impl(ghc < 7.8) + build-depends: + ghc-prim + + hs-source-dirs: src + default-language: Haskell2010 + ghc-options: + -Wall + default-extensions: DeriveDataTypeable + +Test-Suite amqp-tests + type: exitcode-stdio-1.0 + main-is: TestAMQP.hs + build-depends: base >= 4.4 && < 5, + network-transport >= 0.4, + network-transport-amqp, + amqp, + tasty, + tasty-hunit, + network-transport-tests >= 0.1.0.1 + ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind + hs-source-dirs: tests + default-language: Haskell2010 + +Test-Suite api-tests + type: exitcode-stdio-1.0 + main-is: TestAPI.hs + build-depends: base >= 4.4 && < 5, + network-transport >= 0.4, + network-transport-amqp, + amqp, + tasty, + tasty-hunit, + network-transport-tests >= 0.1.0.1 + ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind + hs-source-dirs: tests + default-language: Haskell2010 diff --git a/src/Network/Transport/AMQP.hs b/src/Network/Transport/AMQP.hs new file mode 100644 --- /dev/null +++ b/src/Network/Transport/AMQP.hs @@ -0,0 +1,766 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE CPP #-} + +#if ! MIN_VERSION_base(4,7,0) +{-# LANGUAGE MagicHash #-} +{-# LANGUAGE UnboxedTuples #-} +#endif + +{-# LANGUAGE LambdaCase #-} +{-# OPTIONS_GHC -fno-warn-type-defaults #-} +{-# OPTIONS_GHC -fno-warn-unused-imports #-} -- Cut noise for post-AMP imports + +{--| + A Network Transport Layer for `distributed-process` + based on AMQP and single-owner queues +--} + +module Network.Transport.AMQP ( + createTransport + , AMQPParameters(..) + ) where + +import Network.Transport.AMQP.Internal.Types + +import qualified Network.AMQP as AMQP +import qualified Network.AMQP.Types as AMQP +import qualified Data.Text as T +import qualified Data.Set as Set +import Data.UUID.V4 +import Data.UUID (toString, toWords) +import Control.Concurrent.STM +import Control.Concurrent.STM.TMChan +import Data.Bits +import Data.IORef +import Data.Monoid +import qualified Data.Map.Strict as Map +import Data.ByteString (ByteString) +import Data.Foldable hiding (mapM_) +import qualified Data.ByteString.Lazy as BL +import qualified Data.ByteString as B +import Data.String.Conv +import Data.Serialize +import Network.Transport +import Network.Transport.Internal (mapIOException, asyncWhenCancelled) +import Control.Applicative +import Control.Concurrent (yield, threadDelay) +import Control.Concurrent.Async as Async +import Control.Concurrent.MVar +import Control.Monad +import Lens.Family2 +import System.IO (hPutStrLn , stderr) +import Text.Printf +import Control.Monad.Catch hiding (catches, Handler(..)) +import Control.Exception (mapException, catches, throwIO, Handler(..), evaluate) + +#if ! MIN_VERSION_base(4,7,0) +import GHC.MVar +import GHC.Base +import GHC.PrimopWrappers +#endif + +-------------------------------------------------------------------------------- +-- Utility functions +-------------------------------------------------------------------------------- + +encode' :: AMQPMessage -> BL.ByteString +encode' = encodeLazy + +-------------------------------------------------------------------------------- +decode' :: AMQP.Message -> Either String AMQPMessage +decode' = decodeLazy . AMQP.msgBody + +-------------------------------------------------------------------------------- +onValidEndPoint :: LocalEndPoint -> (ValidLocalEndPointState -> IO ()) -> IO () +onValidEndPoint lep f = withMVar (localState lep) $ \case + LocalEndPointValid v -> f v + _ -> return () + +-------------------------------------------------------------------------------- +onValidRemote :: RemoteEndPoint -> (ValidRemoteEndPointState -> IO ()) -> IO () +onValidRemote rep f = withMVar (remoteState rep) $ \case + RemoteEndPointValid v -> f v + _ -> return () + +-------------------------------------------------------------------------------- +apiNewEndPoint :: AMQPInternalState + -> IO (Either (TransportError NewEndPointErrorCode) EndPoint) +apiNewEndPoint is@AMQPInternalState{..} = try $ mapAMQPException (TransportError NewEndPointFailed . show) $ do + let AMQPParameters{..} = istate_params + modifyMVar istate_tstate $ \tst -> case tst of + TransportClosed -> throwM $ TransportError NewEndPointFailed "Transport is closed." + TransportValid vst@(ValidTransportState _ oldMap) -> do + -- If a valid transport was found, create a new LocalEndPoint + (ep@LocalEndPoint{..}, chan) <- endPointCreate (Map.size oldMap) is + let newEp = EndPoint { + receive = atomically $ do + mx <- readTMChan chan + case mx of + Nothing -> error "receive: channel is closed" + Just x -> return x + , address = localAddress + , connect = apiConnect is ep + , closeEndPoint = apiCloseEndPoint is ep + , newMulticastGroup = return . Left $ newMulticastGroupError + , resolveMulticastGroup = return . Left . const resolveMulticastGroupError + } + return (TransportValid $ + over tstateEndPoints (Map.insert localAddress ep) vst + , newEp) + where + newMulticastGroupError = + TransportError NewMulticastGroupUnsupported "Multicast not supported" + resolveMulticastGroupError = + TransportError ResolveMulticastGroupUnsupported "Multicast not supported" + +------------------------------------------------------------------------------- +endPointCreate :: Int -> AMQPInternalState -> IO (LocalEndPoint, TMChan Event) +endPointCreate newId is@AMQPInternalState{..} = do + -- Each new `EndPoint` has a new Rabbit queue. Think of it as a + -- heavyweight connection. Subsequent connections are multiplexed + -- using Rabbit's exchanges. + uuid <- toS . toString <$> nextRandom + let qName = maybe uuid toS (transportEndpoint istate_params) + newChannel <- AMQP.openChannel (transportConnection istate_params) + + -- https://www.rabbitmq.com/ttl.html + -- Using a TTL for a heavyweight connection we ensure we will do + -- the appropriate cleanup even if we do not shutdown correctly an + -- endpoint or the transport layer. + (ourEndPoint,_,_) <- AMQP.declareQueue newChannel $ AMQP.newQueue { + AMQP.queueName = qName <> ":" <> T.pack (show newId) + , AMQP.queuePassive = False + , AMQP.queueDurable = False + , AMQP.queueExclusive = False + , AMQP.queueHeaders = queueHeaders + } + + newExchange <- toExchangeName (toAddress ourEndPoint) + AMQP.declareExchange newChannel $ AMQP.newExchange { + AMQP.exchangeName = newExchange + , AMQP.exchangeType = "direct" + , AMQP.exchangePassive = False + , AMQP.exchangeDurable = False + , AMQP.exchangeAutoDelete = True + } + + AMQP.bindQueue newChannel ourEndPoint newExchange mempty + + chOut <- newTMChanIO + lep <- LocalEndPoint <$> pure (toAddress ourEndPoint) + <*> pure (AMQPExchange newExchange) + <*> newEmptyMVar + <*> newEmptyMVar + opened <- newIORef True + mask $ \restore -> do + restore (startReceiver is lep chOut newChannel) + AMQP.addChannelExceptionHandler newChannel $ \_ -> do + finaliseEndPoint lep False + putMVar (localState lep) $ LocalEndPointValid + (ValidLocalEndPointState chOut newChannel opened newCounter Map.empty) + return (lep, chOut) + where + queueHeaders = AMQP.FieldTable (Map.fromList [("x-expires", AMQP.FVInt32 3000)]) + +-------------------------------------------------------------------------------- +finaliseEndPoint :: LocalEndPoint -> Bool -> IO () +finaliseEndPoint ourEp requestedByUser = do + join $ withMVar (localState ourEp) $ \case + LocalEndPointClosed -> afterP () + LocalEndPointValid v@ValidLocalEndPointState{..} -> do + writeIORef _localOpened False + return $ do + void $ Async.mapConcurrently (remoteEndPointClose False ourEp) + $ v ^. localRemotes + when requestedByUser $ void $ tryPutMVar (localDone ourEp) () + void $ swapMVar (localState ourEp) LocalEndPointClosed + +-------------------------------------------------------------------------------- +toExchangeName :: EndPointAddress -> IO T.Text +toExchangeName eA = do + uuid <- toString <$> nextRandom + return $ fromAddress eA <> ":" <> T.pack uuid + + +-------------------------------------------------------------------------------- +startReceiver :: AMQPInternalState + -> LocalEndPoint + -> TMChan Event + -> AMQP.Channel + -> IO () +startReceiver tr@AMQPInternalState{..} lep@LocalEndPoint{..} _localChan ch = do + void $ AMQP.consumeMsgs ch (fromAddress localAddress) AMQP.NoAck $ \(msg,_) -> do + case decode' msg of + Left _ -> return () + Right PoisonPill -> do + finaliseEndPoint lep True + Right (MessageData cId rawMsg) -> do + atomically $ writeTMChan _localChan $ Received cId rawMsg + Right (MessageConnect theirAddress) -> do + void $ createOrGetRemoteEndPoint tr lep theirAddress + Right (MessageInitConnection theirAddress theirId rel) -> do + join $ modifyMVar localState $ \case + LocalEndPointValid vst@ValidLocalEndPointState{..} -> do + case Map.lookup theirAddress _localRemotes of + Nothing -> return (LocalEndPointValid vst + , throwM $ InvariantViolated (RemoteEndPointLookupFailed theirAddress)) + Just rep -> modifyMVar (remoteState rep) $ \case + RemoteEndPointFailed -> + throwM $ InvariantViolated (RemoteEndPointMustBeValid theirAddress) + RemoteEndPointClosed -> + throwM $ InvariantViolated (RemoteEndPointMustBeValid theirAddress) + RemoteEndPointClosing{} -> + throwM $ InvariantViolated (RemoteEndPointMustBeValid theirAddress) + w@RemoteEndPointValid{} -> do + conn <- AMQPConnection <$> pure lep + <*> pure rep + <*> pure rel + <*> newMVar (AMQPConnectionValid $ ValidAMQPConnection Nothing Nothing i) + <*> newEmptyMVar + w' <- registerRemoteEndPoint (succ i) w + return ( w' + , ( LocalEndPointValid (set localConnections (Counter (succ i) (Map.insert (succ i) conn m)) vst) + , return ()) + ) + z@(RemoteEndPointPending w) -> do + conn <- AMQPConnection <$> pure lep + <*> pure rep + <*> pure rel + <*> newMVar (AMQPConnectionValid $ ValidAMQPConnection Nothing Nothing i) + <*> newEmptyMVar + modifyIORef w (\xs -> (registerRemoteEndPoint (succ i)) : xs) + return ( z + , ( LocalEndPointValid (set localConnections (Counter (succ i) (Map.insert (succ i) conn m)) vst) + , return ()) + ) + + where + (Counter i m) = vst ^. localConnections + _ -> throwM $ InvariantViolated (LocalEndPointMustBeValid localAddress) + where + registerRemoteEndPoint :: ConnectionId -> RemoteEndPointState -> IO RemoteEndPointState + registerRemoteEndPoint i RemoteEndPointFailed = do + atomically $ do + writeTMChan _localChan (ConnectionOpened i rel theirAddress) + writeTMChan _localChan (ConnectionClosed i) + return RemoteEndPointFailed + registerRemoteEndPoint i RemoteEndPointClosed = do + atomically $ do + writeTMChan _localChan (ConnectionOpened i rel theirAddress) + writeTMChan _localChan (ConnectionClosed i) + return RemoteEndPointClosed + registerRemoteEndPoint i (RemoteEndPointClosing x) = do + atomically $ do + writeTMChan _localChan (ConnectionOpened i rel theirAddress) + writeTMChan _localChan (ConnectionClosed i) + return $ RemoteEndPointClosing x + registerRemoteEndPoint _ RemoteEndPointPending{} = + throwM $ InvariantViolated $ RemoteEndPointCannotBePending (theirAddress) + registerRemoteEndPoint i (RemoteEndPointValid vst@(ValidRemoteEndPointState exg chl _ s _)) = do + publish chl exg (MessageInitConnectionOk localAddress theirId i) + atomically $ writeTMChan _localChan (ConnectionOpened i rel theirAddress) + return $ RemoteEndPointValid vst { _remoteIncomingConnections = Set.insert i s } + + -- + -- MessageCloseConnection + -- + Right (MessageCloseConnection idx) -> do + join $ modifyMVar localState $ \case + LocalEndPointValid v -> + case v ^. localConnectionAt idx of + Nothing -> return (LocalEndPointValid v, return ()) + Just conn -> do + old <- swapMVar (_connectionState conn) AMQPConnectionFailed + let setter = over (localConnections . cntValue) (Map.delete idx) + return ( LocalEndPointValid $ setter v + , case old of + AMQPConnectionFailed -> return () + AMQPConnectionInit -> return () -- throwM InvariantViolation + AMQPConnectionClosed -> return () + AMQPConnectionValid (ValidAMQPConnection{}) -> do + atomically $ writeTMChan _localChan (ConnectionClosed idx) + connectionCleanup (_connectionRemoteEndPoint conn) idx) + LocalEndPointClosed -> return (LocalEndPointClosed, return ()) + + -- + -- MessageInitConnectionOk + -- + Right (MessageInitConnectionOk theirAddress ourId theirId) -> do + join $ withMVar localState $ \case + LocalEndPointValid vst -> + case vst ^. localRemoteAt theirAddress of + Nothing -> return (return ()) -- TODO: send message to the host + Just rep -> modifyMVar (remoteState rep) $ \case + RemoteEndPointFailed -> return (RemoteEndPointFailed, return ()) + RemoteEndPointClosed -> + throwM $ InvariantViolated $ RemoteEndPointShouldBeValidOrClosed theirAddress + RemoteEndPointClosing{} -> + throwM $ InvariantViolated $ RemoteEndPointShouldBeValidOrClosed theirAddress + t@(RemoteEndPointValid (ValidRemoteEndPointState exg chl (Counter x m) s z)) -> return $ + case ourId `Map.lookup` m of + Nothing -> (t, return ()) -- TODO: send message to the hostv + Just c -> + (RemoteEndPointValid (ValidRemoteEndPointState exg chl (Counter x (ourId `Map.delete` m)) s (z+1)) + , do modifyMVar_ (_connectionState c) $ \case + AMQPConnectionFailed -> return AMQPConnectionFailed + AMQPConnectionInit -> + return $ AMQPConnectionValid (ValidAMQPConnection (Just exg) (Just chl) theirId) + AMQPConnectionClosed -> do + publish chl exg (MessageCloseConnection theirId) + return AMQPConnectionClosed + AMQPConnectionValid _ -> + throwM $ (IncorrectState "RemoteEndPoint should be closed") + void $ tryPutMVar (_connectionReady c) () + ) + RemoteEndPointPending p -> + return (RemoteEndPointPending p + , throwM $ (IncorrectState "RemoteEndPoint should be closed")) + LocalEndPointClosed -> return $ return () + + -- + -- MessageEndPointClose + -- + Right (MessageEndPointClose theirAddress True) -> do + getRemoteEndPoint lep theirAddress >>= + traverse_ (\rep -> do + onValidRemote rep $ \ValidRemoteEndPointState{..} -> + publish _remoteChannel _remoteExchange (MessageEndPointCloseOk localAddress) + remoteEndPointClose True lep rep) + + Right (MessageEndPointClose theirAddress False) -> do + getRemoteEndPoint lep theirAddress >>= + traverse_ (\rep -> do + mst <- cleanupRemoteEndPoint lep rep Nothing + case mst of + Nothing -> return () + Just st -> onValidEndPoint lep $ \vst -> do + atomically $ writeTMChan (vst ^. localChan) $ + ErrorEvent $ TransportError (EventConnectionLost theirAddress) "MessageEndPointClose: Exception on remote side" + closeRemoteEndPoint lep rep st) + + -- + -- MessageEndPointCloseOk + -- + Right (MessageEndPointCloseOk theirAddress) -> do + getRemoteEndPoint lep theirAddress >>= + traverse_ (\rep -> do + state <- swapMVar (remoteState rep) RemoteEndPointClosed + closeRemoteEndPoint lep rep state) + +-------------------------------------------------------------------------------- +-- | Close, all network connections. +closeRemoteEndPoint :: LocalEndPoint + -> RemoteEndPoint + -> RemoteEndPointState + -> IO () +closeRemoteEndPoint lep rep state = do + step1 + step2 state + where + step1 = uninterruptibleMask_ $ do + -- NOTE: This step1 function is different + -- from Tweag's one, which used "modifyMVar_", but + -- on this implementation it seems to always deadlock + -- on the "SendException" test. + mbSt <- tryReadMVar (localState lep) + case mbSt of + Nothing -> return () + Just vst -> do + newState <- case vst of + LocalEndPointValid v -> return $ + LocalEndPointValid (over localRemotes (Map.delete (remoteAddress rep)) v) + c -> return c + void $ tryPutMVar (localState lep) newState + step2 (RemoteEndPointClosing (ClosingRemoteEndPoint (AMQPExchange _) _ rd)) = do + _ <- readMVar rd + return () -- No AMQP cleanup needed (cfr. TTL) + step2 _ = return () + + +-------------------------------------------------------------------------------- +connectionCleanup :: RemoteEndPoint -> ConnectionId -> IO () +connectionCleanup rep cid = modifyMVar_ (remoteState rep) $ \case + RemoteEndPointValid v -> return $ + RemoteEndPointValid $ over remoteIncomingConnections (Set.delete cid) v + c -> return c + +-------------------------------------------------------------------------------- +-- | Asynchronous operation, shutdown of the remote end point may take a while +apiCloseEndPoint :: AMQPInternalState + -> LocalEndPoint + -> IO () +apiCloseEndPoint AMQPInternalState{..} LocalEndPoint{..} = + mask_ $ either errorLog return <=< tryAMQP $ do + -- we don't close endpoint here because other threads, + -- should be able to access local endpoint state + old <- readMVar localState + case old of + LocalEndPointValid ValidLocalEndPointState{..} -> do + -- close channel, no events will be received + atomically $ do + writeTMChan _localChan EndPointClosed + closeTMChan _localChan + publish _localChannel localExchange PoisonPill + LocalEndPointClosed -> return () + + takeMVar localDone + void $ swapMVar localState LocalEndPointClosed + modifyMVar_ istate_tstate $ \case + TransportClosed -> return TransportClosed + TransportValid v -> return + $ TransportValid (over tstateEndPoints (Map.delete localAddress) v) + + +-------------------------------------------------------------------------------- +toAddress :: T.Text -> EndPointAddress +toAddress = EndPointAddress . toS + +-------------------------------------------------------------------------------- +fromAddress :: EndPointAddress -> T.Text +fromAddress = toS . endPointAddressToByteString + +-------------------------------------------------------------------------------- +apiConnect :: AMQPInternalState + -> LocalEndPoint + -> EndPointAddress -- ^ Remote address + -> Reliability -- ^ Reliability (ignored) + -> ConnectHints -- ^ Hints + -> IO (Either (TransportError ConnectErrorCode) Connection) +apiConnect is@AMQPInternalState{..} lep@LocalEndPoint{..} theirAddress reliability _ = fmap (either Left id) $ try $ do + mapAMQPException (TransportError ConnectFailed . show) $ do + eRep <- createOrGetRemoteEndPoint is lep theirAddress + case eRep of + Left _ -> return $ Left $ TransportError ConnectFailed "LocalEndPoint is closed." + Right rep -> do + conn <- AMQPConnection <$> pure lep + <*> pure rep + <*> pure reliability + <*> newMVar AMQPConnectionInit + <*> newEmptyMVar + let apiConn = Connection + { send = apiSend conn + , close = apiClose conn + } + join $ modifyMVar (remoteState 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 + newState <- handshake conn w + return (newState, waitReady conn apiConn) + RemoteEndPointPending z -> do + modifyIORef z (\zs -> handshake 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 + AMQPConnectionInit{} -> return $ yield >> waitReady conn apiConn + AMQPConnectionValid{} -> afterP $ Right apiConn + AMQPConnectionFailed{} -> afterP $ Left $ TransportError ConnectFailed "Connection failed." + AMQPConnectionClosed{} -> afterP $ Left $ TransportError ConnectFailed "Connection closed" + handshake _ RemoteEndPointClosed = return RemoteEndPointClosed + handshake _ RemoteEndPointPending{} = + throwM $ TransportError ConnectFailed "Connection pending." + handshake _ (RemoteEndPointClosing x) = return $ RemoteEndPointClosing x + handshake _ RemoteEndPointFailed = return RemoteEndPointFailed + handshake conn (RemoteEndPointValid (ValidRemoteEndPointState exg ch (Counter i m) s z)) = do + publish ch exg (MessageInitConnection localAddress i' reliability) + return $ RemoteEndPointValid (ValidRemoteEndPointState exg ch (Counter i' (Map.insert i' conn m)) s z) + where i' = succ i + +-------------------------------------------------------------------------------- +getRemoteEndPoint :: LocalEndPoint -> EndPointAddress -> IO (Maybe RemoteEndPoint) +getRemoteEndPoint ourEp theirAddr = do + withMVar (localState ourEp) $ \case + LocalEndPointValid v -> return $ theirAddr `Map.lookup` _localRemotes v + LocalEndPointClosed -> return Nothing + +-------------------------------------------------------------------------------- +createOrGetRemoteEndPoint :: AMQPInternalState + -> LocalEndPoint + -> EndPointAddress + -> IO (Either AMQPError RemoteEndPoint) +createOrGetRemoteEndPoint AMQPInternalState{..} ourEp theirAddr = do + join $ modifyMVar (localState ourEp) $ \case + LocalEndPointValid v@ValidLocalEndPointState{..} -> do + opened <- readIORef _localOpened + if opened + then do + case v ^. localRemoteAt theirAddr of + Nothing -> create v + Just rep -> do + withMVar (remoteState rep) $ \case + RemoteEndPointFailed -> do + 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 + newChannel <- AMQP.openChannel (transportConnection istate_params) + newExchange <- toExchangeName theirAddr + AMQP.declareExchange newChannel $ AMQP.newExchange { + AMQP.exchangeName = newExchange + , AMQP.exchangeType = "direct" + , AMQP.exchangePassive = False + , AMQP.exchangeDurable = False + , AMQP.exchangeAutoDelete = True + } + + AMQP.bindQueue newChannel (fromAddress theirAddr) newExchange mempty + + state <- newMVar . RemoteEndPointPending =<< newIORef [] + opened <- newIORef False + let rep = RemoteEndPoint theirAddr state opened + return ( LocalEndPointValid + $ over localRemotes (Map.insert theirAddr rep) v + , initialize newChannel (AMQPExchange newExchange) rep >> return (Right rep)) + + ourAddr = localAddress ourEp + + initialize ch exg rep = do + publish ch exg $ MessageConnect ourAddr + let v = ValidRemoteEndPointState exg ch newCounter Set.empty 0 + modifyMVar_ (remoteState rep) $ \case + RemoteEndPointPending p -> do + z <- foldM (\y f -> f y) (RemoteEndPointValid v) . Prelude.reverse =<< readIORef p + modifyIORef (remoteOpened rep) (const True) + return z + RemoteEndPointValid _ -> + throwM $ InvariantViolated (RemoteEndPointMustBeValid theirAddr) + RemoteEndPointClosed -> return RemoteEndPointClosed + RemoteEndPointClosing j -> return (RemoteEndPointClosing j) + RemoteEndPointFailed -> return RemoteEndPointFailed + +-------------------------------------------------------------------------------- +publish :: AMQP.Channel + -> AMQPExchange + -> AMQPMessage + -> IO () +publish chn (AMQPExchange e) msg = do + AMQP.publishMsg chn e mempty --TODO: Do we need a routing key? + (AMQP.newMsg { AMQP.msgBody = encode' msg + , AMQP.msgDeliveryMode = Just AMQP.NonPersistent + }) + +-------------------------------------------------------------------------------- +-- TODO: Deal with exceptions and error at the broker level. +apiSend :: AMQPConnection + -> [ByteString] + -> IO (Either (TransportError SendErrorCode) ()) +apiSend (AMQPConnection us them _ st _) msg = do + msgs <- try $ mapM_ evaluate msg + case msgs of + Left ex -> do cleanup + return $ Left $ TransportError SendFailed (show (ex::SomeException)) + -- TODO: Check that, in case of AMQP-raised exceptions, we are still + -- doing the appropriate cleanup. + Right _ -> (fmap Right send_) `catches` + [ Handler $ \ex -> -- TransportError - return, as all require + -- actions were performed + return $ Left (ex :: TransportError SendErrorCode) + , Handler $ \ex -> do -- AMQPError appeared exception + cleanup + return $ Left $ TransportError SendFailed (show (ex::AMQPError)) + , Handler $ \ex -> do -- AMQPException appeared exception + cleanup + return $ Left $ TransportError SendFailed (show (ex::AMQP.AMQPException)) + ] + + where + send_ :: IO () + send_ = join $ withMVar (remoteState them) $ \x -> case x of + RemoteEndPointPending{} -> + return $ yield >> send_ + 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 _ -> withMVar st $ \case + AMQPConnectionInit -> return $ yield >> send_ + AMQPConnectionClosed -> + throwIO $ TransportError SendClosed "Connection is closed" + AMQPConnectionFailed -> + throwIO $ TransportError SendFailed "Connection is failed" + AMQPConnectionValid (ValidAMQPConnection (Just exg) (Just ch) idx) -> do + res <- try $ publish ch exg (MessageData idx msg) + case res of + Right _ -> afterP () + Left (_ :: SomeException) -> do + rep <- cleanupRemoteEndPoint us them Nothing + traverse_ (\z -> do + onValidEndPoint us $ \w -> atomically $ writeTMChan (_localChan w) $ + ErrorEvent $ TransportError (EventConnectionLost (remoteAddress them)) "apiSend: Exception on remote side" + closeRemoteEndPoint us them z) rep + throwIO $ TransportError SendFailed "Connection broken." + + cleanup = do + void $ cleanupRemoteEndPoint us them + (Just $ \v -> publish (_remoteChannel v) (_remoteExchange v) $ + MessageEndPointClose (localAddress us) False) + onValidEndPoint us $ \v -> do + atomically $ writeTMChan (_localChan v) $ ErrorEvent $ TransportError + (EventConnectionLost (remoteAddress them)) "Exception on send." + +-------------------------------------------------------------------------------- +apiClose :: AMQPConnection -> IO () +apiClose (AMQPConnection _ them _ st _) = + either errorLog return <=< tryAMQP $ uninterruptibleMask_ $ do + join $ modifyMVar st $ \case + AMQPConnectionValid (ValidAMQPConnection _ _ idx) -> do + return (AMQPConnectionClosed, do + modifyMVar_ (remoteState them) $ \case + v@RemoteEndPointValid{} -> notify idx v + v@(RemoteEndPointPending p) -> modifyIORef p (\xs -> notify idx : xs) >> return v + v -> return v + ) + _ -> return (AMQPConnectionClosed, return ()) + where + notify _ RemoteEndPointFailed = return RemoteEndPointFailed + notify _ (RemoteEndPointClosing x) = return $ RemoteEndPointClosing x + notify _ RemoteEndPointClosed = return RemoteEndPointClosed + notify _ RemoteEndPointPending{} = + throwM $ InvariantViolated (RemoteEndPointMustBeValid (remoteAddress them)) + notify idx w@(RemoteEndPointValid (ValidRemoteEndPointState exg ch _ _ _)) = do + publish ch exg (MessageCloseConnection idx) + return w + +-------------------------------------------------------------------------------- +-- | Close all endpoint connections, return previous state in case +-- if it was alive, for further cleanup actions. +cleanupRemoteEndPoint :: LocalEndPoint + -> RemoteEndPoint + -> (Maybe (ValidRemoteEndPointState -> IO ())) + -> IO (Maybe RemoteEndPointState) +cleanupRemoteEndPoint lep rep actions = do + modifyMVar (localState lep) $ \case + LocalEndPointValid v -> do + modifyIORef (remoteOpened rep) (const False) + oldState <- swapMVar (remoteState rep) newState + case oldState of + RemoteEndPointValid w -> do + let cn = w ^. (remotePendingConnections . cntValue) + traverse_ (\c -> void $ swapMVar (_connectionState c) AMQPConnectionFailed) cn + cn' <- foldM + (\c' idx -> case idx `Map.lookup` cn of + Nothing -> return c' + Just c -> do + void $ swapMVar (_connectionState c) AMQPConnectionFailed + return $ over cntValue (Map.delete idx) c' + ) + (v ^. localConnections) + (Set.toList (_remoteIncomingConnections w)) + case actions of + Nothing -> return () + Just f -> f w + return ( LocalEndPointValid (set localConnections cn' v) + , Just oldState) + _ -> return (LocalEndPointValid v, Nothing) + c -> return (c, Nothing) + where + newState = RemoteEndPointFailed + +-------------------------------------------------------------------------------- +remoteEndPointClose :: Bool -> LocalEndPoint -> RemoteEndPoint -> IO () +remoteEndPointClose silent lep rep = do + modifyIORef (remoteOpened rep) (const False) + join $ modifyMVar (remoteState rep) $ \o -> case o of + RemoteEndPointFailed -> return (o, return ()) + RemoteEndPointClosed -> return (o, return ()) + RemoteEndPointClosing (ClosingRemoteEndPoint _ _ l) -> return (o, void $ readMVar l) + RemoteEndPointPending _ -> do + let err = (error "Pending actions should not be executed") + closing err err o + -- TODO: store socket, or delay + RemoteEndPointValid v -> closing (_remoteExchange v) (_remoteChannel v) o + where + closing exg ch old = do + lock <- newEmptyMVar + return (RemoteEndPointClosing (ClosingRemoteEndPoint exg ch lock), go lock old) + + go lock old@(RemoteEndPointValid (ValidRemoteEndPointState c amqpCh _ s i)) = do + -- close all connections + void $ cleanupRemoteEndPoint lep rep Nothing + withMVar (localState lep) $ \case + LocalEndPointClosed -> return () + LocalEndPointValid ValidLocalEndPointState{..} -> do + -- notify about all connections close (?) do we really want it? + traverse_ (atomically . writeTMChan _localChan . ConnectionClosed) (Set.toList s) + -- if we have outgoing connections, then we have connection error + when (i > 0) $ do + let msg = "Remote end point closed, but " <> show i <> " outgoing connection(s)" + atomically $ writeTMChan _localChan $ ErrorEvent $ TransportError (EventConnectionLost (remoteAddress rep)) msg + + unless silent $ do + publish amqpCh c (MessageEndPointClose (localAddress lep) True) + yield + void $ Async.race (readMVar lock) (threadDelay 1000000) + void $ tryPutMVar lock () + closeRemoteEndPoint lep rep old + return () + go _ _ = return () + +-------------------------------------------------------------------------------- +createTransport :: AMQPParameters -> IO Transport +createTransport params@AMQPParameters{..} = do + let validTState = ValidTransportState transportConnection Map.empty + tState <- newMVar (TransportValid validTState) + let iState = AMQPInternalState params tState + return Transport { + newEndPoint = apiNewEndPoint iState + , closeTransport = apiCloseTransport iState + } + +-------------------------------------------------------------------------------- +-- Do not close the externally-passed AMQP connection, +-- or it will compromise users sharing it! +apiCloseTransport :: AMQPInternalState -> IO () +apiCloseTransport is = do + old <- swapMVar (istate_tstate is) TransportClosed + case old of + TransportClosed -> return () + TransportValid (ValidTransportState _ mp) -> either errorLog return <=< tryAMQP $ do + traverse_ (apiCloseEndPoint is) mp + +-------------------------------------------------------------------------------- +afterP :: a -> IO (IO a) +afterP = return . return + +-------------------------------------------------------------------------------- +-- | 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-amqp] Unhandled error: %s" $ show s) + +-------------------------------------------------------------------------------- +promoteAMQPException :: a -> a +promoteAMQPException = mapException DriverError + +-------------------------------------------------------------------------------- +tryAMQP :: (MonadCatch m) => m a -> m (Either AMQPError a) +tryAMQP = try . promoteAMQPException + +-------------------------------------------------------------------------------- +mapAMQPException :: (Exception e) => (AMQPError -> e) -> a -> a +mapAMQPException = mapException + +#if ! MIN_VERSION_base(4,7,0) +-------------------------------------------------------------------------------- +tryReadMVar :: MVar a -> IO (Maybe a) +tryReadMVar (MVar m) = IO $ \ s -> + case tryReadMVar# m s of + (# s', 0#, _ #) -> (# s', Nothing #) -- MVar is empty + (# s', _, a #) -> (# s', Just a #) -- MVar is full +#endif diff --git a/src/Network/Transport/AMQP/Internal/Types.hs b/src/Network/Transport/AMQP/Internal/Types.hs new file mode 100644 --- /dev/null +++ b/src/Network/Transport/AMQP/Internal/Types.hs @@ -0,0 +1,200 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE RankNTypes #-} +{-# OPTIONS_GHC -fno-warn-orphans #-} +{-# OPTIONS_GHC -fno-warn-missing-signatures #-} + +module Network.Transport.AMQP.Internal.Types + (module Network.Transport.AMQP.Internal.Types) where + +import qualified Network.AMQP as AMQP +import qualified Data.Text as T +import Data.Map.Strict (Map) +import qualified Data.Map.Strict as Map +import Data.Set (Set) +import Data.IORef +import GHC.Generics (Generic) +import Data.ByteString (ByteString) +import Data.Serialize +import Data.Typeable +import Network.Transport +import Control.Concurrent.MVar +import Control.Exception +import Control.Concurrent.STM.TMChan +import Lens.Family2.TH +import Lens.Family2 + +-------------------------------------------------------------------------------- +-- Data Types +-- Largely inspired to: `network-transport-zeromq` by Tweag I/O and +-- `network-transport-tcp` by Well Typed et al. +-------------------------------------------------------------------------------- + +data AMQPParameters = AMQPParameters { + transportConnection :: AMQP.Connection + , transportMultiCastEndpoint :: !T.Text + , transportEndpoint :: !(Maybe T.Text) + -- ^ The queue and exchange name. If not specified, will be randomised. + } + +-------------------------------------------------------------------------------- +data TransportState + = TransportValid ValidTransportState + | TransportClosed + +-------------------------------------------------------------------------------- +data ValidTransportState = ValidTransportState { + _tstateConnection :: AMQP.Connection + , _tstateEndPoints :: !(Map EndPointAddress LocalEndPoint) + } + +-------------------------------------------------------------------------------- +data AMQPInternalState = AMQPInternalState { + istate_params :: !AMQPParameters + , istate_tstate :: !(MVar TransportState) + } + +-------------------------------------------------------------------------------- +data LocalEndPoint = LocalEndPoint + { localAddress :: !EndPointAddress + , localExchange :: !AMQPExchange + , localDone :: !(MVar ()) + , localState :: !(MVar LocalEndPointState) + } + +-------------------------------------------------------------------------------- +data LocalEndPointState = + LocalEndPointValid !ValidLocalEndPointState + | LocalEndPointClosed + +-------------------------------------------------------------------------------- +data ValidLocalEndPointState = ValidLocalEndPointState + { + _localChan :: !(TMChan Event) + , _localChannel :: !AMQP.Channel + , _localOpened :: !(IORef Bool) + , _localConnections :: !(Counter ConnectionId AMQPConnection) + , _localRemotes :: !(Map EndPointAddress RemoteEndPoint) + } + +-------------------------------------------------------------------------------- +data Counter a b = Counter + { _cntNext :: !a + , _cntValue :: !(Map a b) + } + +-------------------------------------------------------------------------------- +newCounter :: Counter ConnectionId AMQPConnection +newCounter = Counter 0 Map.empty + +-------------------------------------------------------------------------------- +data AMQPConnection = AMQPConnection + { _connectionLocalEndPoint :: !LocalEndPoint + , _connectionRemoteEndPoint :: !RemoteEndPoint + , _connectionReliability :: !Reliability + , _connectionState :: !(MVar AMQPConnectionState) + , _connectionReady :: !(MVar ()) + } + +newtype AMQPExchange = AMQPExchange T.Text deriving (Show, Eq) + +-------------------------------------------------------------------------------- +data AMQPConnectionState = + AMQPConnectionInit + | AMQPConnectionValid !ValidAMQPConnection + | AMQPConnectionClosed + | AMQPConnectionFailed + +-------------------------------------------------------------------------------- +data ValidAMQPConnection = ValidAMQPConnection + { _amqpExchange :: !(Maybe AMQPExchange) + , _amqpChannel :: !(Maybe AMQP.Channel) + , _amqpConnectionId :: !ConnectionId + } + +-------------------------------------------------------------------------------- +data RemoteEndPoint = RemoteEndPoint + { remoteAddress :: !EndPointAddress + , remoteState :: !(MVar RemoteEndPointState) + , remoteOpened :: !(IORef Bool) + } + +-------------------------------------------------------------------------------- +data ClosingRemoteEndPoint = ClosingRemoteEndPoint + { _closingRemoteExchange :: !AMQPExchange + , _closingRemoteChannel :: !AMQP.Channel + , _closingRemoteDone :: !(MVar ()) + } + +-------------------------------------------------------------------------------- +data RemoteEndPointState + = RemoteEndPointValid ValidRemoteEndPointState + | RemoteEndPointClosed + | RemoteEndPointFailed + | RemoteEndPointPending (IORef [RemoteEndPointState -> IO RemoteEndPointState]) + | RemoteEndPointClosing ClosingRemoteEndPoint + +-------------------------------------------------------------------------------- +data ValidRemoteEndPointState = ValidRemoteEndPointState + { _remoteExchange :: !AMQPExchange + , _remoteChannel :: !AMQP.Channel + , _remotePendingConnections :: !(Counter ConnectionId AMQPConnection) + , _remoteIncomingConnections :: !(Set ConnectionId) + , _remoteOutgoingCount :: !Int + } + +-------------------------------------------------------------------------------- +makeLenses ''ValidTransportState +makeLenses ''ValidLocalEndPointState +makeLenses ''ValidRemoteEndPointState +makeLenses ''AMQPConnection +makeLenses ''ValidAMQPConnection +makeLenses ''Counter + +-------------------------------------------------------------------------------- +-- Lenses +-- +-------------------------------------------------------------------------------- +localConnectionAt :: Phantom f => ConnectionId -> LensLike' f ValidLocalEndPointState (Maybe AMQPConnection) +localConnectionAt idx = localConnections . cntValue . to (Map.lookup idx) + +localRemoteAt :: Phantom f => EndPointAddress -> LensLike' f ValidLocalEndPointState (Maybe RemoteEndPoint) +localRemoteAt eA = localRemotes . to (Map.lookup eA) + +-------------------------------------------------------------------------------- +data AMQPMessage + = MessageConnect !EndPointAddress -- ^ Connection greeting + | MessageInitConnection !EndPointAddress !ConnectionId !Reliability + | MessageInitConnectionOk !EndPointAddress !ConnectionId !ConnectionId + | MessageCloseConnection !ConnectionId + | MessageData !ConnectionId ![ByteString] + | MessageEndPointClose !EndPointAddress !Bool + | MessageEndPointCloseOk !EndPointAddress + | PoisonPill + deriving (Show, Generic) + +deriving instance Generic EndPointAddress +instance Serialize EndPointAddress +deriving instance Generic Reliability +instance Serialize Reliability +instance Serialize AMQPMessage + +data AMQPError + = InvariantViolated InvariantViolation + | IncorrectState String + | ConnectionFailed + | DriverError AMQP.AMQPException + deriving (Typeable, Show) + +instance Exception AMQPError + +data InvariantViolation = + RemoteEndPointLookupFailed EndPointAddress + | RemoteEndPointCannotBePending EndPointAddress + | RemoteEndPointShouldBeValidOrClosed EndPointAddress + | RemoteEndPointMustBeValid EndPointAddress + | LocalEndPointMustBeValid EndPointAddress + deriving Show diff --git a/tests/TestAMQP.hs b/tests/TestAMQP.hs new file mode 100644 --- /dev/null +++ b/tests/TestAMQP.hs @@ -0,0 +1,48 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# OPTIONS_GHC -fno-warn-unused-imports #-} -- Cut noise for post-AMP imports +module Main where + +import Control.Applicative +import Network.Transport +import Network.Transport.AMQP +import Network.AMQP +import Network.Transport.Tests +import Network.Transport.Tests.Auxiliary (runTests) + +main :: IO () +main = do + conn <- openConnection "localhost" "/" "guest" "guest" + let amqpParameters = AMQPParameters conn "simple-multicast" Nothing + testTransportCompliance (Right <$> createTransport amqpParameters) + +-- | These tests that our transport layer it's compliant to the +-- official specification. +testTransportCompliance :: IO (Either String Transport) -> IO () +testTransportCompliance newTransport = do + Right transport <- newTransport + runTests + [ + ("PingPong", testPingPong transport numPings) + , ("EndPoints", testEndPoints transport numPings) + , ("Connections", testConnections transport numPings) + , ("CloseOneConnection", testCloseOneConnection transport numPings) + , ("CloseOneDirection", testCloseOneDirection transport numPings) + , ("CloseReopen", testCloseReopen transport numPings) + , ("ParallelConnects", testParallelConnects transport (numPings * 2)) + , ("SendAfterClose", testSendAfterClose transport (numPings * 2)) + , ("Crossing", testCrossing transport numPings) + , ("CloseTwice", testCloseTwice transport (numPings * 2)) + , ("ConnectToSelf", testConnectToSelf transport numPings) + , ("ConnectToSelfTwice", testConnectToSelfTwice transport numPings) + , ("CloseSelf", testCloseSelf newTransport) + , ("CloseEndPoint", testCloseEndPoint transport numPings) + , ("CloseTransport", testCloseTransport newTransport) + , ("ExceptionOnReceive", testExceptionOnReceive newTransport) + , ("SendException", testSendException newTransport) + , ("Kill", testKill newTransport 30) + -- testKill test have a timeconstraint so n-t-amqp + -- fails to work with required speed, we need to + -- reduce a number of tests here. (Same limitation as n-t-0MQ) + ] + where + numPings = 10 :: Int diff --git a/tests/TestAPI.hs b/tests/TestAPI.hs new file mode 100644 --- /dev/null +++ b/tests/TestAPI.hs @@ -0,0 +1,139 @@ +{-# LANGUAGE OverloadedStrings #-} +module Main where + +import Control.Monad ( replicateM ) + +import Network.Transport +import Network.Transport.AMQP +import Network.AMQP +import Test.Tasty +import Test.Tasty.HUnit + +main :: IO () +main = defaultMain $ + testGroup "API tests" + [ testCase "simple" test_simple + --, testCase "connection break" test_connectionBreak + --, testCase "test multicast" test_multicast + --, testCase "connect to non existent host" test_nonexists + --, testCase "connect with prior knowledge" test_prior + ] + +newTransport :: IO Transport +newTransport = do + conn <- openConnection "localhost" "/" "guest" "guest" + let amqpTransport = AMQPParameters conn "simple-multicast" Nothing + createTransport amqpTransport + +test_simple :: IO () +test_simple = do + transport <- newTransport + Right ep1 <- newEndPoint transport + Right ep2 <- newEndPoint transport + Right c1 <- connect ep1 (address ep2) ReliableOrdered defaultConnectHints + Right c2 <- connect ep2 (address ep1) ReliableOrdered defaultConnectHints + Right _ <- send c1 ["123"] + Right _ <- send c2 ["321"] + close c1 + close c2 + [ConnectionOpened _ ReliableOrdered _, Received _ ["321"], ConnectionClosed _] <- replicateM 3 $ receive ep1 + [ConnectionOpened _ ReliableOrdered _, Received _ ["123"], ConnectionClosed _] <- replicateM 3 $ receive ep2 + closeTransport transport + +--test_connectionBreak :: IO () +--test_connectionBreak = do +-- (zmq, transport) <- +-- createTransportExposeInternals defaultZMQParameters "127.0.0.1" +-- Right ep1 <- newEndPoint transport +-- Right ep2 <- newEndPoint transport +-- Right ep3 <- newEndPoint transport +-- +-- Right c21 <- connect ep1 (address ep2) ReliableOrdered defaultConnectHints +-- Right c22 <- connect ep2 (address ep1) ReliableOrdered defaultConnectHints +-- Right c23 <- connect ep3 (address ep1) ReliableOrdered defaultConnectHints +-- +-- ConnectionOpened 1 ReliableOrdered _ <- receive ep2 +-- ConnectionOpened 1 ReliableOrdered _ <- receive ep1 +-- ConnectionOpened 2 ReliableOrdered _ <- receive ep1 +-- +-- breakConnection zmq (address ep1) (address ep2) +-- +-- ErrorEvent (TransportError (EventConnectionLost _ ) _) <- receive $ ep1 +-- ErrorEvent (TransportError (EventConnectionLost _ ) _) <- receive $ ep2 +-- +-- Left (TransportError SendFailed _) <- send c21 ["test"] +-- Left (TransportError SendFailed _) <- send c22 ["test"] +-- +-- Left (TransportError SendFailed _) <- send c23 ["test"] +-- ErrorEvent (TransportError (EventConnectionLost _) _ ) <- receive ep1 +-- Right c24 <- connect ep3 (address ep1) ReliableOrdered defaultConnectHints +-- Right () <- send c24 ["final"] +-- ConnectionOpened 3 ReliableOrdered _ <- receive ep1 +-- Received 3 ["final"] <- receive ep1 +-- closeTransport transport +-- +--test_multicast :: IO () +--test_multicast = do +-- transport <- createTransport defaultZMQParameters "127.0.0.1" +-- Right ep1 <- newEndPoint transport +-- Right ep2 <- newEndPoint transport +-- Right g1 <- newMulticastGroup ep1 +-- multicastSubscribe g1 +-- threadDelay 1000000 +-- multicastSend g1 ["test"] +-- ReceivedMulticast _ ["test"] <- receive ep1 +-- Right g2 <- resolveMulticastGroup ep2 (multicastAddress g1) +-- multicastSubscribe g2 +-- threadDelay 100000 +-- multicastSend g2 ["test-2"] +-- ReceivedMulticast _ ["test-2"] <- receive ep2 +-- ReceivedMulticast _ ["test-2"] <- receive ep1 +-- return () +-- +--test_auth :: IO () +--test_auth = do +-- tr2 <- +-- createTransport defaultZMQParameters{ zmqSecurityMechanism = +-- Just $ SecurityPlain "user" "password" } +-- "127.0.0.1" +-- Right ep3 <- newEndPoint tr2 +-- Right ep4 <- newEndPoint tr2 +-- Right c3 <- connect ep3 (address ep4) ReliableOrdered defaultConnectHints +-- Right _ <- send c3 ["4456"] +-- [ConnectionOpened 1 ReliableOrdered _, Received 1 ["4456"]] <- replicateM 2 $ receive ep4 +-- Right c4 <- connect ep3 (address ep4) ReliableOrdered defaultConnectHints +-- Right _ <- send c4 ["5567"] +-- [ConnectionOpened 2 ReliableOrdered _, Received 2 ["5567"]] <- replicateM 2 $ receive ep4 +-- return () +-- +--test_nonexists :: IO () +--test_nonexists = do +-- tr <- createTransport defaultZMQParameters "127.0.0.1" +-- Right ep <- newEndPoint tr +-- Left (TransportError ConnectFailed _) <- connect ep (EndPointAddress "tcp://129.0.0.1:7684") ReliableOrdered defaultConnectHints +-- closeTransport tr +-- +--test_cleanup :: IO () +--test_cleanup = do +-- (zmq, transport) <- +-- createTransportExposeInternals defaultZMQParameters "127.0.0.1" +-- x <- newIORef (0::Int) +-- Just _ <- registerCleanupAction zmq (modifyIORef x (+1)) +-- Just u <- registerCleanupAction zmq (modifyIORef x (+1)) +-- applyCleanupAction zmq u +-- closeTransport transport +-- 2 <- readIORef x +-- return () +-- +--test_prior :: IO () +--test_prior = do +-- (zmqa, _) <- createTransportExposeInternals defaultZMQParameters "127.0.0.1" +-- Right epa <- apiNewEndPoint defaultHints{hintsPort=Just 8888} zmqa +-- +-- (_, transportb) <- +-- createTransportExposeInternals defaultZMQParameters "127.0.0.1" +-- Right epb <- newEndPoint transportb +-- Right _ <- connect epb (EndPointAddress "tcp://127.0.0.1:8888") ReliableOrdered defaultConnectHints +-- (ConnectionOpened 1 ReliableOrdered _) <- receive epa +-- return () +--