1 files changed, 111 insertions, 59 deletions

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