diff --git a/courier.cabal b/courier.cabal --- a/courier.cabal +++ b/courier.cabal @@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/ name: courier -version: 0.1.0.4 +version: 0.1.0.5 synopsis: A message-passing library, intended for simplifying network applications description: Inspired by Erlang's simple message-passing facilities, courier provides roughly similar capabilities. Applications simply create one or more endpoints, @@ -17,7 +17,7 @@ maintainer: phil@haphazardhouse.net copyright: Copyright (c) 2013 Phil Hargett -category: Network +category: Network,Distributed Computing,Message-Oriented build-type: Simple cabal-version: >=1.8 @@ -68,6 +68,7 @@ cereal, directory, hslogger, + network-simple, -- this project's modules courier diff --git a/src/Network/Endpoints.hs b/src/Network/Endpoints.hs --- a/src/Network/Endpoints.hs +++ b/src/Network/Endpoints.hs @@ -32,7 +32,9 @@ unbindEndpoint, sendMessage, + sendMessage_, broadcastMessage, + broadcastMessage_, receiveMessage, receiveMessageTimeout, @@ -79,6 +81,7 @@ -- > name2 = "endpoint2" -- > resolver = resolverFromList [(name1,"localhost:2000"), -- > (name2,"localhost:2001")] +-- > transport <- newTCPTransport resolver -- > endpoint1 <- newEndpoint [transport] -- > endpoint2 <- newEndpoint [transport] -- > Right () <- bindEndpoint endpoint1 name1 @@ -86,6 +89,7 @@ -- > sendMessage endpoint1 name2 $ encode "hello world!" -- > msg <- receiveMessage endpoint2 -- > print msg +-- > shutdown transport {-| Endpoints are a locus of communication, used for sending and receive messages. @@ -163,11 +167,29 @@ return $ Right () {-| +A variant of 'sendMessage' for use when the return value can be ignored. + +-} +sendMessage_ :: Endpoint -> Name -> Message -> IO () +sendMessage_ endpoint name msg = do + _ <- sendMessage endpoint name msg + return () + +{-| Helper for sending a single 'Message' to several 'Endpoint's. -} broadcastMessage :: Endpoint -> [Name] -> Message -> IO [(Either String ())] broadcastMessage endpoint names msg = do mapM (\name -> sendMessage endpoint name msg) names + +{-| +Variant of 'broadcastMessage' that ignores the results of sending. + +-} +broadcastMessage_ :: Endpoint -> [Name] -> Message -> IO () +broadcastMessage_ endpoint names msg = do + _ <- broadcastMessage endpoint names msg + return () {-| Receive the next 'Message' sent to the 'Endpoint', blocking until a message is available. diff --git a/src/Network/Transport/TCP.hs b/src/Network/Transport/TCP.hs --- a/src/Network/Transport/TCP.hs +++ b/src/Network/Transport/TCP.hs @@ -40,7 +40,7 @@ import GHC.Generics -import Network.Socket (HostName,ServiceName,Socket,sClose,accept) +import Network.Socket (sClose,accept) import Network.Simple.TCP hiding (accept) import System.Log.Logger @@ -140,9 +140,14 @@ unbind = tcpUnbind listener address } where - tcpListen address port = listen HostAny port $ \(socket,_) -> do - tcpAccept address socket + tcpListen address port = + listen HostAny port $ \(socket,_) -> + catch (do + tcpAccept address socket) + (\e -> do + warningM _log $ "Listen error: " ++ (show (e :: SomeException))) tcpAccept address socket = do + infoM _log $ "Listening for connections on " ++ (show address) ++ ": " ++ (show socket) (client,clientAddress) <- accept socket _ <- async $ tcpDispatch address client clientAddress tcpAccept address socket @@ -153,12 +158,15 @@ Nothing -> sClose client Just (IdentifyMessage clientAddress) -> do infoM _log $ "Identified " ++ (show clientAddress) - msngr <- newMessenger client clientAddress (tcpInbound transport) + clientSocket <- atomically $ newTMVar client + msngr <- newMessenger clientSocket clientAddress (tcpInbound transport) found <- atomically $ do msngrs <- readTVar $ tcpMessengers transport return $ M.lookup clientAddress msngrs case found of - Just _ -> closeMessenger msngr + Just _ -> do + infoM _log $ "Already have messenger for " ++ (show clientAddress) + closeMessenger msngr Nothing -> do addMessenger transport clientAddress msngr tcpIdentify client clientAddress = do @@ -177,38 +185,50 @@ tcpSendTo :: TCPTransport -> Name -> Message -> IO () tcpSendTo transport name msg = do - Just address <- resolve (tcpResolver transport) name - let env = encode $ Envelope { - envelopeDestination = name, - envelopeContents = msg - } - amsngr <- atomically $ do - msngrs <- readTVar $ tcpMessengers transport - return $ M.lookup address msngrs - case amsngr of - Nothing -> do - let (host,port) = parseTCPAddress address - infoM _log $ "Connecting to " ++ (show address) - (socket,_) <- connectSock host port - infoM _log $ "Connected to " ++ (show address) - msngr <- newMessenger socket address (tcpInbound transport) - addMessenger transport address msngr - -- identify all bindings - identifyAll msngr - deliver msngr env - return () - Just msngr -> deliver msngr env - where - deliver msngr message = atomically $ writeTQueue (messengerOut msngr) message - identifyAll msngr = do - bindings <- atomically $ readTVar $ tcpBindings transport - boundAddresses <- mapM (resolve $ tcpResolver transport) (M.keys bindings) - let uniqueAddresses = S.toList $ S.fromList boundAddresses - mapM_ (identify msngr) uniqueAddresses - identify msngr maybeUniqueAddress= do - case maybeUniqueAddress of - Nothing -> return() - Just uniqueAddress -> deliver msngr $ encode $ IdentifyMessage uniqueAddress + isLocal <- local + if isLocal + then return () + else remote + where + local = do + found <- atomically $ do + bindings <- readTVar $ tcpBindings transport + return $ M.lookup name bindings + case found of + Nothing -> return False + Just mbox -> do + atomically $ writeTQueue mbox msg + return True + remote = do + Just address <- resolve (tcpResolver transport) name + let env = encode $ Envelope { + envelopeDestination = name, + envelopeContents = msg + } + amsngr <- atomically $ do + msngrs <- readTVar $ tcpMessengers transport + return $ M.lookup address msngrs + case amsngr of + Nothing -> do + msngrs <- atomically $ readTVar $ tcpMessengers transport + infoM _log $ "No messenger for " ++ (show address) ++ " in " ++ (show msngrs) + socketVar <- atomically $ newEmptyTMVar + msngr <- newMessenger socketVar address (tcpInbound transport) + addMessenger transport address msngr + identifyAll msngr + deliver msngr env + return () + Just msngr -> deliver msngr env + deliver msngr message = atomically $ writeTQueue (messengerOut msngr) message + identifyAll msngr = do + bindings <- atomically $ readTVar $ tcpBindings transport + boundAddresses <- mapM (resolve $ tcpResolver transport) (M.keys bindings) + let uniqueAddresses = S.toList $ S.fromList boundAddresses + mapM_ (identify msngr) uniqueAddresses + identify msngr maybeUniqueAddress= do + case maybeUniqueAddress of + Nothing -> return() + Just uniqueAddress -> deliver msngr $ encode $ IdentifyMessage uniqueAddress tcpShutdown :: TCPTransport -> IO () tcpShutdown transport = do @@ -220,19 +240,20 @@ mapM_ sClose $ M.elems listeners infoM _log $ "Closing dispatcher" mapM_ cancel $ S.toList $ tcpDispatchers transport + mapM_ wait $ S.toList $ tcpDispatchers transport data Messenger = Messenger { messengerOut :: Mailbox, messengerAddress :: Address, messengerSender :: Async (), messengerReceiver :: Async (), - messengerSocket :: Socket + messengerSocket :: TMVar Socket } instance Show Messenger where - show msngr = "Messenger(" ++ (show $ messengerAddress msngr) ++ "," ++ (show $ messengerSocket msngr) ++ ")" + show msngr = "Messenger(" ++ (show $ messengerAddress msngr) ++ ")" -newMessenger :: Socket -> Address -> Mailbox -> IO Messenger +newMessenger :: TMVar Socket -> Address -> Mailbox -> IO Messenger newMessenger socket address inc = do out <- newMailbox sndr <- async $ sender socket address out @@ -257,33 +278,61 @@ closeMessenger msngr = do cancel $ messengerSender msngr cancel $ messengerReceiver msngr - sClose $ messengerSocket msngr + open <- atomically $ tryTakeTMVar $ messengerSocket msngr + case open of + Just socket -> sClose socket + Nothing -> return () + -sender :: Socket -> Address -> Mailbox -> IO () -sender socket address mailbox = sendMessages +sender :: TMVar Socket -> Address -> Mailbox -> IO () +sender socketVar address mailbox = sendMessages where - sendMessages = do + sendMessages = do + reconnect catch (do infoM _log $ "Waiting to send to " ++ (show address) msg <- atomically $ readTQueue mailbox infoM _log $ "Sending message to " ++ (show address) - send socket $ encode (B.length msg) - infoM _log $ "Length sent" - send socket msg - infoM _log $ "Message sent to" ++ (show address) - ) (\e -> do - errorM _log $ "Send error " ++ (show (e :: SomeException)) - throw e) + connected <- atomically $ tryReadTMVar socketVar + case connected of + Just socket -> do + send socket $ encode (B.length msg) + infoM _log $ "Length sent" + send socket msg + infoM _log $ "Message sent to" ++ (show address) + Nothing -> return () + ) (\e -> do + warningM _log $ "Send error: " ++ (show (e :: SomeException)) + disconnect) sendMessages + reconnect = do + -- TODO need a timeout here, in case connecting always fails + connected <- atomically $ tryReadTMVar socketVar + case connected of + Just _ -> return () + Nothing -> do + let (host,port) = parseTCPAddress address + infoM _log $ "Connecting to " ++ (show host) ++ ":" ++ (show port) -- (show address) + (socket,sockAddr) <- connectSock host port + infoM _log $ "Connected to " ++ (show address) ++ ": " ++ (show sockAddr) + atomically $ putTMVar socketVar socket + disconnect = do + connected <- atomically $ tryTakeTMVar socketVar + case connected of + Just socket -> sClose socket + Nothing -> return () + dispatcher :: TVar (M.Map Name Mailbox) -> Mailbox -> IO () dispatcher bindings mbox = dispatchMessages where - dispatchMessages = do - infoM _log $ "Dispatching messages" - env <- atomically $ readTQueue mbox - dispatchMessage env - dispatchMessages + dispatchMessages = catch (do + infoM _log $ "Dispatching messages" + env <- atomically $ readTQueue mbox + dispatchMessage env + dispatchMessages) + (\e -> do + warningM _log $ "Dispatch error: " ++ (show (e :: SomeException))) dispatchMessage env = do infoM _log $ "Dispatching message" let envelopeOrErr = decode env @@ -301,19 +350,25 @@ writeTQueue dest msg return () -receiver :: Socket -> Address -> Mailbox -> IO () -receiver socket address mailbox = receiveMessages +receiver :: TMVar Socket -> Address -> Mailbox -> IO () +receiver socketVar address mailbox = receiveMessages where - receiveMessages = do + receiveMessages = catch (do infoM _log $ "Waiting to receive from " ++ (show address) + socket <- atomically $ readTMVar socketVar maybeMsg <- receiveMessage socket infoM _log $ "Received message from " ++ (show address) case maybeMsg of - Nothing -> return () - Just msg -> atomically $ writeTQueue mailbox msg + Nothing -> do + sClose socket + return () + Just msg -> do + atomically $ writeTQueue mailbox msg + receiveMessages) (\e -> do + warningM _log $ "Receive error: " ++ (show (e :: SomeException))) receiveMessage :: Socket -> IO (Maybe Message) -receiveMessage socket = do +receiveMessage socket = catch (do maybeLen <- recv socket 8 -- TODO must figure out what defines length of an integer in bytes case maybeLen of Nothing -> do @@ -322,7 +377,9 @@ Just len -> do maybeMsg <- recv socket $ msgLength (decode len) infoM _log $ "Received message" - return maybeMsg + return maybeMsg) (\e -> do + warningM _log $ "Receive error: " ++ (show (e :: SomeException)) + throw e) where msgLength (Right size) = size msgLength (Left err) = error err diff --git a/tests/TestTCP.hs b/tests/TestTCP.hs --- a/tests/TestTCP.hs +++ b/tests/TestTCP.hs @@ -25,68 +25,121 @@ _log :: String _log = "test.transport.tcp" +testDelay :: Int +testDelay = 1 * 1000000 + +address1 :: Address +address1 = "localhost:2001" + +address2 :: Address +address2 = "localhost:2002" + tests :: [Test.Framework.Test] tests = [ testCase "tcp-endpoints+transport" testEndpointTransport, testCase "tcp-bind-unbind" testEndpointBindUnbind, testCase "tcp-send-receive" testEndpointSendReceive, - testCase "tcp-send-receive-reply" testEndpointSendReceiveReply + testCase "tcp-double-send-receive" testEndpointDoubleSendReceive, + testCase "tcp-send-receive-reply" testEndpointSendReceiveReply, + testCase "tcp-multiple-send-receive-reply" testEndpointMultipleSendReceiveReply, + testCase "tcp-local-send-receive-reply" testEndpointLocalSendReceiveReply ] testEndpointTransport :: Assertion testEndpointTransport = do let name1 = "endpoint1" name2 = "endpoint2" - let resolver = resolverFromList [(name1,"localhost:2000"), - (name2,"localhost:2001")] - transport <- newTCPTransport resolver - _ <- newEndpoint [transport] - return () + let resolver = resolverFromList [(name1,address1), + (name2,address2)] + bracket (newTCPTransport resolver) + shutdown + (\transport -> do + _ <- newEndpoint [transport] + return ()) testEndpointBindUnbind :: Assertion testEndpointBindUnbind = do let name1 = "endpoint1" name2 = "endpoint2" - let resolver = resolverFromList [(name1,"localhost:2000"), - (name2,"localhost:2001")] - transport <- newTCPTransport resolver - endpoint <- newEndpoint [transport] - Right () <- bindEndpoint endpoint name1 - unbound <- unbindEndpoint endpoint name1 - case unbound of - Left err -> assertFailure $ "Unbind failed: " ++ err - Right () -> assertBool "Unbind succeeded" True - return () + let resolver = resolverFromList [(name1,address1), + (name2,address2)] + bracket (newTCPTransport resolver) + shutdown + (\transport -> do + endpoint <- newEndpoint [transport] + Right () <- bindEndpoint endpoint name1 + unbound <- unbindEndpoint endpoint name1 + case unbound of + Left err -> assertFailure $ "Unbind failed: " ++ err + Right () -> assertBool "Unbind succeeded" True + return ()) testEndpointSendReceive :: Assertion testEndpointSendReceive = do let name1 = "endpoint1" name2 = "endpoint2" - let resolver = resolverFromList [(name1,"localhost:2000"), - (name2,"localhost:2001")] + let resolver = resolverFromList [(name1,address1), + (name2,address2)] bracket (newTCPTransport resolver) - (\transport1 -> shutdown transport1) + shutdown (\transport1 -> bracket (newTCPTransport resolver) - (\transport2 -> shutdown transport2) + shutdown (\transport2 -> do endpoint1 <- newEndpoint [transport1] endpoint2 <- newEndpoint [transport2] Right () <- bindEndpoint endpoint1 name1 Right () <- bindEndpoint endpoint2 name2 - threadDelay $ 1 * 1000000 - _ <- sendMessage endpoint1 name2 $ encode "hello!" - Just msg <- receiveMessageTimeout endpoint2 (2 * 1000000) + threadDelay testDelay + + infoM _log "Sending message from 1 to 2" + sendMessage_ endpoint1 name2 $ encode "hello!" + Just msg <- receiveMessageTimeout endpoint2 testDelay assertEqual "Received message not same as sent" (Right "hello!") (decode msg) + Right () <- unbindEndpoint endpoint1 name1 + Right () <- unbindEndpoint endpoint2 name2 return ())) +testEndpointDoubleSendReceive :: Assertion +testEndpointDoubleSendReceive = do + let name1 = "endpoint1" + name2 = "endpoint2" + name3 = "endpoint3" + let resolver = resolverFromList [(name1,address1), + (name2,address2), + (name3,address1)] + bracket (newTCPTransport resolver) + shutdown + (\transport1 -> do + bracket (newTCPTransport resolver) + shutdown + (\transport2 -> do + endpoint1 <- newEndpoint [transport1] + endpoint2 <- newEndpoint [transport2] + endpoint3 <- newEndpoint [transport1] + Right () <- bindEndpoint endpoint1 name1 + Right () <- bindEndpoint endpoint2 name2 + Right () <- bindEndpoint endpoint3 name3 + threadDelay testDelay + + verifiedSend endpoint1 endpoint2 name1 name2 "hello" + verifiedSend endpoint3 endpoint2 name3 name2 "ciao!" + + Right () <- unbindEndpoint endpoint1 name1 + + verifiedSend endpoint3 endpoint2 name3 name2 "hi!" + + Right () <- unbindEndpoint endpoint2 name2 + Right () <- unbindEndpoint endpoint3 name3 + return ())) + testEndpointSendReceiveReply :: Assertion testEndpointSendReceiveReply = do let name1 = "endpoint1" name2 = "endpoint2" - let resolver = resolverFromList [(name1,"localhost:2002"), - (name2,"localhost:2003")] + let resolver = resolverFromList [(name1,address1), + (name2,address2)] bracket (newTCPTransport resolver) shutdown (\transport1 -> @@ -97,17 +150,74 @@ endpoint2 <- newEndpoint [transport2] Right () <- bindEndpoint endpoint1 name1 Right () <- bindEndpoint endpoint2 name2 - threadDelay $ 1 * 1000000 + threadDelay testDelay - infoM _log "Sending message from 1 to 2" - _ <- sendMessage endpoint1 name2 $ encode "hello!" - Just msg1 <- receiveMessageTimeout endpoint2 (1 * 1000000) - assertEqual "Received message not same as sent" (Right "hello!") (decode msg1) - - infoM _log "Sending message from 2 to 1" - _ <- sendMessage endpoint2 name1 $ encode "hi!" - Just msg2 <- receiveMessageTimeout endpoint1 (1 * 1000000) - assertEqual "Received message not same as sent" (Right "hi!") (decode msg2) + verifiedSend endpoint1 endpoint2 name1 name2 "hello" + verifiedSend endpoint2 endpoint1 name2 name1 "hi!" + + Right () <- unbindEndpoint endpoint1 name1 + Right () <- unbindEndpoint endpoint2 name2 return ())) +testEndpointLocalSendReceiveReply :: Assertion +testEndpointLocalSendReceiveReply = do + let name1 = "endpoint1" + name2 = "endpoint2" + let resolver = resolverFromList [(name1,address1), + (name2,address1)] + bracket (newTCPTransport resolver) + shutdown + (\transport1 -> do + endpoint1 <- newEndpoint [transport1] + endpoint2 <- newEndpoint [transport1] + Right () <- bindEndpoint endpoint1 name1 + Right () <- bindEndpoint endpoint2 name2 + threadDelay testDelay + + verifiedSend endpoint1 endpoint2 name1 name2 "hello" + verifiedSend endpoint2 endpoint1 name2 name1 "hi!" + + Right () <- unbindEndpoint endpoint1 name1 + Right () <- unbindEndpoint endpoint2 name2 + + return ()) + +testEndpointMultipleSendReceiveReply :: Assertion +testEndpointMultipleSendReceiveReply = do + let name1 = "endpoint1" + name2 = "endpoint2" + let resolver = resolverFromList [(name1,address1), + (name2,address2)] + bracket (newTCPTransport resolver) + shutdown + (\transport1 -> + bracket (newTCPTransport resolver) + shutdown + (\transport2 -> do + endpoint1 <- newEndpoint [transport1] + endpoint2 <- newEndpoint [transport2] + Right () <- bindEndpoint endpoint1 name1 + Right () <- bindEndpoint endpoint2 name2 + threadDelay testDelay + + roundtrip endpoint1 endpoint2 name1 name2 + roundtrip endpoint2 endpoint1 name2 name1 + roundtrip endpoint1 endpoint2 name1 name2 + roundtrip endpoint2 endpoint1 name2 name1 + + Right () <- unbindEndpoint endpoint1 name1 + Right () <- unbindEndpoint endpoint2 name2 + + return ())) + where + roundtrip endpoint1 endpoint2 name1 name2 = do + verifiedSend endpoint1 endpoint2 name1 name2 "hello" + verifiedSend endpoint2 endpoint1 name2 name1 "hi!" + +verifiedSend :: Endpoint -> Endpoint -> Name -> Name -> String -> Assertion +verifiedSend endpoint1 endpoint2 name1 name2 msg = do + infoM _log $ "Sending message from " ++ name1 ++ " to " ++ name2 + sendMessage_ endpoint1 name2 $ encode msg + Just msg1 <- receiveMessageTimeout endpoint2 testDelay + assertEqual "Received message not same as sent" (Right msg) (decode msg1)