Factored the cache and retry pattern into module SendMessage

3 files changed, 33 insertions, 75 deletions

diff --git a/SendMessage.hs b/Presence/SendMessage.hs
index 6e8ea2b9..6e8ea2b9 100644
--- a/SendMessage.hs
+++ b/Presence/SendMessage.hs
diff --git a/Presence/XMPP.hs b/Presence/XMPP.hs
index a469c08e..339cc6f2 100644
--- a/Presence/XMPP.hs
+++ b/Presence/XMPP.hs
@@ -1,6 +1,7 @@
 {-# LANGUAGE OverloadedStrings   #-}
 {-# LANGUAGE FlexibleContexts    #-}
 {-# LANGUAGE ViewPatterns        #-}
+{-# LANGUAGE TypeFamilies        #-}
 module XMPP 
     ( module XMPPTypes
     , listenForXmppClients
@@ -9,6 +10,10 @@ module XMPP
     , seekRemotePeers
     , quitListening
     , OutBoundMessage(..)
+    , OutgoingConnections
+    , CachedMessages
+    , toPeer
+    , newOutgoingConnections
     , sendMessage
     )  where
 
@@ -17,14 +22,13 @@ import XMPPTypes
 import ByteStringOperators
 import ControlMaybe
 import XMLToByteStrings
-
+import SendMessage
 
 import Data.Maybe (catMaybes)
 import Data.HList
 import Network.Socket 
     ( Family
     , connect
-    , socketToHandle
     , sClose
     , Socket(..)
     , socket
@@ -33,11 +37,6 @@ import Network.Socket
 import Network.BSD 
     ( getProtocolNumber
     )
-import System.IO
-    ( BufferMode(..)
-    , IOMode(..)
-    , hSetBuffering
-    )
 import Control.Concurrent.STM
 import Data.Conduit
 import Data.ByteString (ByteString)
@@ -45,7 +44,6 @@ import qualified Data.ByteString.Lazy.Char8 as L
     ( putStrLn
     , fromChunks
     )
-import Control.Concurrent (forkIO)
 import Control.Concurrent.Async
 import Control.Exception as E
     ( handle
@@ -72,11 +70,6 @@ import           Data.Set as Set (Set,(\\))
 import qualified Data.Set as Set
 import qualified Data.Map as Map
 import           Data.Map as Map (Map)
-import GHC.Conc 
-    ( threadStatus
-    , ThreadStatus(..)
-    , ThreadId
-    )
 
 textToByteString x = L.fromChunks [S.encodeUtf8 x]
 
@@ -914,65 +907,28 @@ data CachedMessages = CachedMessages
     , probes :: Map JID (Set (Bool,JID)) -- False means solicitation rather than probe
     , approvals :: Map JID (Set (Bool,JID) ) -- False means rejection rather than approval
     }
-newCache = CachedMessages Map.empty Map.empty Map.empty
-
-connect_to_server chan peer = (>> return ()) . runMaybeT $ do
-    let port = 5269 :: Int
-    -- We'll cache Presence notifications until the socket
-    -- is ready.
-    cached <- liftIO $ newIORef newCache
-
-    let mmInsert val key mm = Map.alter f key mm
-            where 
-                f Nothing    = Just $ Set.singleton val
-                f (Just set) = Just $ Set.insert val set
-    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' = mmInsert  (True,from) to $ probes cache
-                    writeIORef cached (cache { probes=probes' })
-        cacheCmd (Solicitation from to) cached = do
-                    cache <- readIORef cached
-                    let probes' = mmInsert (False,from) to $ probes cache
-                    writeIORef cached (cache { probes=probes' })
-        cacheCmd (Approval from to) cached = do
-                    cache <- readIORef cached
-                    let approvals' = mmInsert (True,from) to $ approvals cache
-                    writeIORef cached (cache { approvals=approvals' })
-        cacheCmd (Rejection from to) cached = do
-                    cache <- readIORef cached
-                    let approvals' = mmInsert (False,from) to $ approvals cache
-                    writeIORef cached (cache { approvals=approvals' })
-
-    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 newCache
-                return (cache,packetSink h)
-            MaybeT $ handleOutgoingToPeer (restrictSocket sock) cache chan snk
-
-        liftIO $ cacheCmd retry cached
-        liftIO $ putStrLn $ "retrying " ++ show retry
-        sendmsgs
+
+instance CommandCache CachedMessages where
+    type CacheableCommand CachedMessages = OutBoundMessage
+    emptyCache = CachedMessages Map.empty Map.empty Map.empty
+
+    updateCache (OutBoundPresence (Presence jid Offline)) cache = 
+                cache { presences=Map.delete jid . presences $ cache }
+    updateCache (OutBoundPresence p@(Presence jid st)) cache =
+                cache { presences=Map.insert jid st . presences $ cache }
+    updateCache (PresenceProbe from to) cache =
+                cache { probes = mmInsert  (True,from) to $ probes cache }
+    updateCache (Solicitation from to) cache =
+                cache { probes= mmInsert (False,from) to $ probes cache }
+    updateCache (Approval from to) cache =
+                cache { approvals= mmInsert (True,from) to $ approvals cache }
+    updateCache (Rejection from to) cache =
+                cache { approvals= mmInsert (False,from) to $ approvals cache }
+
+mmInsert val key mm = Map.alter f key mm
+ where 
+    f Nothing    = Just $ Set.singleton val
+    f (Just set) = Just $ Set.insert val set
 
 
 greetPeer =
@@ -1139,6 +1095,7 @@ connect' addr port = do
         )
 
 
+{-
 sendMessage  :: TVar (Map Peer (TChan OutBoundMessage, ThreadId)) -> OutBoundMessage -> Peer -> IO ()
 sendMessage cons msg peer0 = do
     let peer = discardPort peer0
@@ -1170,11 +1127,12 @@ sendMessage cons msg peer0 = do
     atomically $ writeTChan (fst entry) msg
     when is_new . atomically $
         readTVar cons >>= writeTVar cons . Map.insert peer entry
+-}
 
 
 
 seekRemotePeers :: JabberPeerSession config =>
-     XMPPPeerClass config -> TChan Presence -> TVar (Map Peer (TChan OutBoundMessage, ThreadId)) -> IO b0
+     XMPPPeerClass config -> TChan Presence -> OutgoingConnections CachedMessages -> IO b0
 seekRemotePeers config chan server_connections = do
     fix $ \loop -> do
         event <- atomically $ readTChan chan
diff --git a/Presence/main.hs b/Presence/main.hs
index 317dc71e..3d5ac046 100644
--- a/Presence/main.hs
+++ b/Presence/main.hs
@@ -89,7 +89,7 @@ data PresenceState = PresenceState
     , remoteUsers :: TVar (Map Peer (RefCount,TVar (MultiMap JabberUser (JabberResource,JabberShow))))
 
     -- outGoingConnections  - a set of channels that may be used to send messages to remote peers.
-    , outGoingConnections :: TVar (Map Peer (TChan OutBoundMessage, ThreadId)) 
+    , outGoingConnections :: OutgoingConnections CachedMessages
     }
 
 
@@ -108,7 +108,7 @@ newPresenceState hostname = atomically $ do
     locals_greedy <- newEmptyTMVar
     rchan <- newEmptyTMVar
     remotes <- newTVar (Map.empty)
-    server_connections <- newServerConnections
+    server_connections <- newOutgoingConnections toPeer
     return $ PresenceState hostname tty us subs locals_greedy rchan remotes server_connections