courier 0.1.0.4 → 0.1.0.5
raw patch · 4 files changed
+290/−100 lines, 4 filesPVP: minor bump suggested
API additions: PVP suggests at least a minor version bump
API changes (from Hackage documentation)
+ Network.Endpoints: broadcastMessage_ :: Endpoint -> [Name] -> Message -> IO ()
+ Network.Endpoints: sendMessage_ :: Endpoint -> Name -> Message -> IO ()
Files
- courier.cabal +3/−2
- src/Network/Endpoints.hs +22/−0
- src/Network/Transport/TCP.hs +120/−63
- tests/TestTCP.hs +145/−35
courier.cabal view
@@ -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
src/Network/Endpoints.hs view
@@ -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.
src/Network/Transport/TCP.hs view
@@ -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
tests/TestTCP.hs view
@@ -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)