From 5d045775732fdc4af4bf67e441aa904de9632153 Mon Sep 17 00:00:00 2001 From: joe Date: Tue, 2 Jul 2013 20:00:30 -0400 Subject: better exception handling --- Presence/XMPP.hs | 170 ++++++++++++++++++++++++++++++++++++------------------- 1 file changed, 111 insertions(+), 59 deletions(-) (limited to 'Presence/XMPP.hs') diff --git a/Presence/XMPP.hs b/Presence/XMPP.hs index fd29473e..3f81f000 100644 --- a/Presence/XMPP.hs +++ b/Presence/XMPP.hs @@ -42,8 +42,6 @@ import System.IO , IOMode(..) , hSetBuffering ) -import Control.Exception - ( bracketOnError ) import Control.Concurrent.STM import Data.Conduit import qualified Data.Conduit.List as CL @@ -58,7 +56,12 @@ import qualified Data.ByteString.Lazy.Char8 as L ) import Control.Concurrent (forkIO,killThread) import Control.Concurrent.Async -import Control.Exception (handle,SomeException(..),finally) +import Control.Exception + ( handle + -- , SomeException(..) + , finally + , bracketOnError ) +import GHC.IO.Exception (IOException(..)) import Control.Monad.IO.Class import Control.Monad.Trans.Class import Control.Monad.Trans.Maybe @@ -548,7 +551,8 @@ handlePresenceProbe session stanza = do subs <- getSubscribers (peerSessionFactory session) user liftIO $ L.putStrLn $ "subscribers for "<++>bshow user<++>": " <++>bshow subs forM_ subs $ \jidstr -> do - handle (\(SomeException _) -> return ()) $ do + handle (\(IOError _ _ _ _ _ _) -> return ()) $ do + -- handle (\(SomeException _) -> return ()) $ do L.putStrLn $ "parsing " <++>jidstr sub <- parseHostNameJID jidstr putStrLn $ "comparing " ++show (peer sub , peerAddress session) @@ -615,46 +619,45 @@ data CachedMessages = CachedMessages connect_to_server chan peer = (>> return ()) . runMaybeT $ do let port = 5269 :: Int - - connected <- liftIO . async $ connect' (peerAddr peer) port - -- We'll cache Presence notifications until the socket -- is ready. cached <- liftIO $ newIORef (CachedMessages Map.empty Map.empty) - sock <- MaybeT . fix $ \loop -> do - e <- atomically $ orElse - (fmap Right $ waitSTM connected) - (fmap Left $ readTChan chan) - case e of - Left (OutBoundPresence (Presence jid Offline)) -> do - cache <- readIORef cached - writeIORef cached (cache { presences=Map.delete jid . presences $ cache }) - loop - Left (OutBoundPresence p@(Presence jid st)) -> do - cache <- readIORef cached - writeIORef cached (cache { presences=Map.insert jid st . presences $ cache }) - loop - Left (PresenceProbe from to) -> do - cache <- readIORef cached - let probes' = Map.adjust (Set.insert from) to $ probes cache - writeIORef cached (cache { probes=probes' }) - loop - {- - Left event -> do - L.putStrLn $ "REMOTE-OUT DISCARDED: " <++> bshow event - loop - -} - Right sock -> return sock - - liftIO $ do - h <- socketToHandle sock ReadWriteMode - hSetBuffering h NoBuffering - let snk = packetSink h - cache <- readIORef $ cached - -- hint garbage collector: we're done with this... - writeIORef cached (CachedMessages Map.empty Map.empty) - handleOutgoingToPeer (restrictSocket sock) cache chan snk + let cacheCmd (OutBoundPresence (Presence jid Offline)) cached = do + cache <- readIORef cached + writeIORef cached (cache { presences=Map.delete jid . presences $ cache }) + cacheCmd (OutBoundPresence p@(Presence jid st)) cached = do + cache <- readIORef cached + writeIORef cached (cache { presences=Map.insert jid st . presences $ cache }) + cacheCmd (PresenceProbe from to) cached = do + cache <- readIORef cached + let probes' = Map.adjust (Set.insert from) to $ probes cache + writeIORef cached (cache { probes=probes' }) + + fix $ \sendmsgs -> do + connected <- liftIO . async $ connect' (peerAddr peer) port + + sock <- MaybeT . fix $ \loop -> do + e <- atomically $ orElse + (fmap Right $ waitSTM connected) + (fmap Left $ readTChan chan) + case e of + Left cmd -> cacheCmd cmd cached >> loop + Right sock -> return sock + + retry <- do + (cache,snk) <- liftIO $ do + h <- socketToHandle sock ReadWriteMode + hSetBuffering h NoBuffering + cache <- readIORef $ cached + -- hint garbage collector: we're done with this... + writeIORef cached (CachedMessages Map.empty Map.empty) + return (cache,packetSink h) + MaybeT $ handleOutgoingToPeer (restrictSocket sock) cache chan snk + + liftIO $ cacheCmd retry cached + liftIO $ putStrLn $ "retrying " ++ show retry + sendmsgs greetPeer = @@ -689,46 +692,94 @@ presenceProbe sock fromjid tojid = do , EventEndElement "{jabber:server}presence" ] -toPeer sock cache chan = do +{- +toPeerChain + :: SocketLike sock => + sock + -> CachedMessages + -> TChan OutBoundMessage + -> Sink ByteString IO b + -> IO b +toPeerChain sock cache chan snk = toPeer sock cache chan $$ renderChunks =$ snk +-} + +toPeer + :: SocketLike sock => + sock + -> CachedMessages + -> TChan OutBoundMessage + -> (Maybe OutBoundMessage -> IO ()) + -> ConduitM i [Event] IO () +toPeer sock cache chan fail = do let -- log = liftIO . L.putStrLn . ("(>P) " <++>) - send xs = yield xs >> prettyPrint ">P: " xs + send xs = yield xs >> prettyPrint ">P: " xs -- >> return (3::Int) + checkConnection cmd = do + liftIO $ catch (getPeerName sock >> return ()) + (\_ -> fail . Just $ cmd) + sendPresence presence = do + r <- lift $ xmlifyPresenceForPeer sock presence + {- + liftIO $ do + p' <- catch (fmap (Just . RemotePeer) $ getPeerName sock) + (\_ -> (fail . Just . OutBoundPresence $ presence) >> return Nothing) + L.putStrLn $ "sending Presence to " <++?> fmap showPeer p' + -} + let cmd = OutBoundPresence presence + checkConnection cmd + yieldOr r (fail . Just $ cmd) + prettyPrint ">P: " r + sendProbe from to = do + r <- liftIO $ presenceProbe sock from to + let cmd = PresenceProbe from to + checkConnection cmd + yieldOr r (fail . Just $ cmd) + prettyPrint ">P: " r + send greetPeer forM_ (Map.assocs . presences $ cache) $ \(jid,st) -> do - r <- lift $ xmlifyPresenceForPeer sock (Presence jid st) - send r + sendPresence (Presence jid st) forM_ (Map.assocs . probes $ cache) $ \(to,froms) -> do forM_ (Set.toList froms) $ \from -> do liftIO $ L.putStrLn "sending cached probe..." - r <- liftIO $ presenceProbe sock from to - send r + sendProbe from to fix $ \loop -> do event <- lift . atomically $ readTChan chan case event of - OutBoundPresence p -> do - r <- lift $ xmlifyPresenceForPeer sock p - send r + OutBoundPresence p -> sendPresence p PresenceProbe from to -> do liftIO $ L.putStrLn "sending live probe..." - r <- liftIO $ presenceProbe sock from to - send r + sendProbe from to loop send goodbyePeer +handleOutgoingToPeer + :: SocketLike sock => + sock + -> CachedMessages + -> TChan OutBoundMessage + -> Sink ByteString IO () + -> IO (Maybe OutBoundMessage) handleOutgoingToPeer sock cache chan snk = do p <- getPeerName sock L.putStrLn $ "(>P) connected " <++> showPeer (RemotePeer p) + failed <- newIORef Nothing + let failure cmd = do + writeIORef failed cmd + putStrLn $ "Failed: " ++ show cmd finally ( #ifdef RENDERFLUSH - toPeer sock cache chan - $$ flushList - =$= renderBuilderFlush def - =$= builderToByteStringFlush - =$= discardFlush - =$ snk + handle (\(IOError _ _ _ _ _ _) -> return ()) $ + toPeer sock cache chan failure + $$ flushList + =$= renderBuilderFlush def + =$= builderToByteStringFlush + =$= discardFlush + =$ snk #else - toPeer sock cache chan $$ renderChunks =$ snk + handle (\(IOError _ _ _ _ _ _) -> return ()) $ toPeer sock cache chan failure $$ renderChunks =$ snk #endif ) $ L.putStrLn $ "(>P) disconnected " <++> showPeer (RemotePeer p) + readIORef failed connect' :: SockAddr -> Int -> IO (Maybe Socket) connect' addr port = do @@ -743,7 +794,8 @@ connect' addr port = do -} let getport (SockAddrInet port _) = port getport (SockAddrInet6 port _ _ _) = port - let doException (SomeException e) = do + let doException e@(IOError _ _ _ _ _ _) = do + -- let doException (SomeException e) = do L.putStrLn $ "\nFailed to reach "<++> showPeer (RemotePeer addr) <++> " on port "<++>bshow port<++>": " <++> bshow e return Nothing handle doException -- cgit v1.2.3