Generic connection manager for Tox and XMPP.

3 files changed, 135 insertions, 121 deletions

diff --git a/Presence/Presence.hs b/Presence/Presence.hs
index d2a6b6eb..15775857 100644
--- a/Presence/Presence.hs
+++ b/Presence/Presence.hs
@@ -1,6 +1,7 @@
-{-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE TupleSections #-}
-{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE ExistentialQuantification #-}
+{-# LANGUAGE LambdaCase                #-}
+{-# LANGUAGE OverloadedStrings         #-}
+{-# LANGUAGE TupleSections             #-}
 module Presence where
 
 import System.Environment
@@ -47,6 +48,7 @@ import PeerResolve
 import ConsoleWriter
 import ClientState
 import Util
+import qualified Connection
 
 isPeerKey :: ConnectionKey -> Bool
 isPeerKey k = case k of { PeerKey {} -> True ; _ -> False }
@@ -59,18 +61,26 @@ localJID user resource = do
     hostname <- textHostName
     return $ user <> "@" <> hostname <> "/" <> resource
 
-newPresenceState cw = atomically $ do
+data PresenceState = forall status. PresenceState
+    { clients         :: TVar (Map ConnectionKey ClientState)
+    , clientsByUser   :: TVar (Map Text LocalPresence)
+    , remotesByPeer   :: TVar (Map ConnectionKey
+                                   (Map UserName RemotePresence))
+    , server          :: TMVar (XMPPServer, Connection.Manager status Text)
+    , keyToChan       :: TVar (Map ConnectionKey Conn)
+    , consoleWriter   :: Maybe ConsoleWriter
+    }
+
+
+newPresenceState cw xmpp = atomically $ do
         clients <- newTVar Map.empty
         clientsByUser <- newTVar Map.empty
         remotesByPeer <- newTVar Map.empty
-        associatedPeers <- newTVar Map.empty
-        xmpp <- newEmptyTMVar
         keyToChan <- newTVar Map.empty
         return PresenceState
             { clients = clients
             , clientsByUser = clientsByUser
             , remotesByPeer = remotesByPeer
-            , associatedPeers = associatedPeers
             , keyToChan = keyToChan
             , server = xmpp
             , consoleWriter = cw
@@ -122,7 +132,7 @@ pcSingletonNetworkClient :: ConnectionKey
 pcSingletonNetworkClient key client =
     LocalPresence
     { networkClients = Map.singleton key client
-    } 
+    }
 
 pcInsertNetworkClient :: ConnectionKey -> ClientState -> LocalPresence -> LocalPresence
 pcInsertNetworkClient key client pc =
@@ -139,21 +149,6 @@ pcIsEmpty :: LocalPresence -> Bool
 pcIsEmpty pc = Map.null (networkClients pc)
 
 
-data PresenceState = PresenceState
-    { clients :: TVar (Map ConnectionKey ClientState)
-    , clientsByUser :: TVar (Map Text LocalPresence)
-    , remotesByPeer :: TVar (Map ConnectionKey
-                                 (Map UserName
-                                      RemotePresence))
-    , -- | These are the remote peers we are seeking to obtain a connection to.
-      -- See 'rosterGetStuff'.
-      associatedPeers :: TVar (Map SockAddr ())
-    , server :: TMVar XMPPServer
-    , keyToChan :: TVar (Map ConnectionKey Conn)
-    , consoleWriter :: Maybe ConsoleWriter
-    }
-
-
 
 getConsolePids :: PresenceState -> IO [(Text,ProcessID)]
 getConsolePids state = do
@@ -197,7 +192,7 @@ chooseResourceName state k addr desired = do
                                   mb
 
     localJID (clientUser client) (clientResource client)
-    
+
   where
     getTTYandPID muid = do
         -- us <- fmap (map (second fst) . Map.toList) . readTVarIO $ activeUsers state
@@ -210,9 +205,9 @@ chooseResourceName state k addr desired = do
         return (rsc,pid)
 
     getJabberUserForId muid =
-             maybe (return "nobody") 
-                   (\(uid,_) -> 
-                        handle (\(SomeException _) -> 
+             maybe (return "nobody")
+                   (\(uid,_) ->
+                        handle (\(SomeException _) ->
                                 return . (<> "uid.") . Text.pack . show $ uid)
                           $ do
                              user <- fmap userName $ getUserEntryForID uid
@@ -248,19 +243,13 @@ rosterGetStuff what state k = forClient state k (return [])
     $ \client -> do
     jids <- configText what (clientUser client)
     let hosts = map ((\(_,h,_)->h) . splitJID) jids
-    addrs <- resolveAllPeers hosts 
-    peers <- atomically $ readTVar (associatedPeers state)
-    addrs <- return $ addrs `Map.difference` peers
-    sv <- atomically $ takeTMVar $ server state
-    -- Grok peers to associate with from the roster:
-    forM_ (Map.keys addrs) $ \addr -> do
-        putStrLn $ "new addr: "++show addr
-        addPeer sv addr
-    -- Update local set of associated peers
-    atomically $ do
-        writeTVar (associatedPeers state) (addrs `Map.union` peers)
-        putTMVar (server state) sv
-    return jids
+    case state of
+      PresenceState { server = svVar } -> do
+        (sv,conns) <- atomically $ takeTMVar svVar
+        -- Grok peers to associate with from the roster:
+        forM_ hosts $ \host -> Connection.setPolicy conns host Connection.TryingToConnect
+        atomically $ putTMVar svVar (sv,conns)
+        return jids
 
 rosterGetBuddies :: PresenceState -> ConnectionKey -> IO [Text]
 rosterGetBuddies state k = rosterGetStuff ConfigFiles.getBuddies state k
@@ -309,7 +298,7 @@ sendProbesAndSolicitations state k laddr chan = do
     -- XXX: The following O(n²) nub may be a little
     --      too onerous.
     forM_ (nub xs) $ \(isbud,u,user) -> do
-        let make = if isbud then presenceProbe 
+        let make = if isbud then presenceProbe
                             else presenceSolicitation
             toh = peerKeyToText k
             jid = unsplitJID (u,toh,Nothing)
@@ -448,7 +437,7 @@ deliverMessage :: PresenceState
                 -> IO ()
                 -> StanzaWrap (LockedChan Event)
                 -> IO ()
-deliverMessage state fail msg = 
+deliverMessage state fail msg =
     case stanzaOrigin msg of
       NetworkOrigin senderk@(ClientKey {}) _ -> do
         -- Case 1.  Client -> Peer
@@ -509,14 +498,14 @@ deliverMessage state fail msg =
 
 
 setClientFlag :: PresenceState -> ConnectionKey -> Int8 -> IO ()
-setClientFlag state k flag = 
+setClientFlag state k flag =
     atomically $ do
         cmap <- readTVar (clients state)
         flip (maybe $ return ()) (Map.lookup k cmap) $ \client -> do
         setClientFlag0 client flag
 
 setClientFlag0 :: ClientState -> Int8 -> STM ()
-setClientFlag0 client flag = 
+setClientFlag0 client flag =
     modifyTVar' (clientFlags client) (\flgs -> flgs .|. flag)
 
 informSentRoster :: PresenceState -> ConnectionKey -> IO ()
@@ -528,7 +517,7 @@ subscribedPeers :: Text -> IO [SockAddr]
 subscribedPeers user = do
     jids <- configText ConfigFiles.getSubscribers user
     let hosts = map ((\(_,h,_)->h) . splitJID) jids
-    fmap Map.keys $ resolveAllPeers hosts 
+    fmap Map.keys $ resolveAllPeers hosts
 
 -- | this JID is suitable for peers, not clients.
 clientJID :: Conn -> ClientState -> Text
@@ -577,13 +566,13 @@ informPeerPresence :: PresenceState
                     -> IO ()
 informPeerPresence state k stanza = do
     -- Presence must indicate full JID with resource...
-    putStrLn $ "xmppInformPeerPresence checking from address..." 
+    putStrLn $ "xmppInformPeerPresence checking from address..."
     flip (maybe $ return ()) (stanzaFrom stanza) $ \from -> do
     let (muser,h,mresource) = splitJID from
     putStrLn $ "xmppInformPeerPresence from  = " ++ show from
     -- flip (maybe $ return ()) mresource       $ \resource -> do
     flip (maybe $ return ()) muser           $ \user -> do
-    
+
     clients <- atomically $ do
 
         -- Update remotesByPeer...
@@ -592,33 +581,33 @@ informPeerPresence state k stanza = do
             rp = case (presenceShow $ stanzaType stanza) of
                     Offline ->
                         maybe Map.empty
-                              (\resource -> 
+                              (\resource ->
                                   maybe (Map.empty)
                                      (Map.delete resource . resources)
                                    $ Map.lookup user umap)
                               mresource
-                    
+
                     _ ->maybe Map.empty
                               (\resource ->
-                                  maybe (Map.singleton resource stanza) 
+                                  maybe (Map.singleton resource stanza)
                                      (Map.insert resource stanza . resources )
                                     $ Map.lookup user umap)
                               mresource
             umap' = Map.insert user (RemotePresence rp) umap
-        
+
         flip (maybe $ return []) (case presenceShow $ stanzaType stanza of
                                      Offline -> Just ()
                                      _ -> mresource >> Just ())
             $ \_ -> do
         writeTVar (remotesByPeer state) $ Map.insert k umap' rbp
         -- TODO: Store or delete the stanza (remotesByPeer)
-        
+
         -- all clients, we'll filter available/authorized later
 
         ktc <- readTVar (keyToChan state)
         runTraversableT $ do
         (ck,client) <- liftMT $ fmap Map.toList $ readTVar (clients state)
-        con <- liftMaybe $ Map.lookup ck ktc 
+        con <- liftMaybe $ Map.lookup ck ktc
         return (ck,con,client)
     putStrLn $ "xmppInformPeerPresence (length clients="++show (length clients)++")"
     forM_ clients $ \(ck,con,client) -> do
@@ -631,7 +620,7 @@ informPeerPresence state k stanza = do
             let ClientKey laddr = ck
             (_,trip) <- multiplyJIDForClient laddr from
             return (map unsplitJID trip)
-                    
+
         putStrLn $ "sending to client: " ++ show (stanzaType stanza,froms)
         forM_ froms $ \from' -> do
         dup <- cloneStanza stanza
@@ -706,7 +695,7 @@ sendCachedPresence state k = do
     rbp <- atomically $ readTVar (remotesByPeer state)
     jids <- configText ConfigFiles.getBuddies (clientUser client)
     let hosts = map ((\(_,h,_)->h) . splitJID) jids
-    addrs <- resolveAllPeers hosts 
+    addrs <- resolveAllPeers hosts
     let onlines = rbp `Map.intersection` Map.mapKeys PeerKey addrs
         ClientKey laddr = k
     mcon <- atomically $ do ktc <- readTVar (keyToChan state)
@@ -790,9 +779,8 @@ clientSubscriptionRequest state fail k stanza chan = do
     let (mu,h,_) = splitJID to
     to <- return $ unsplitJID (mu,h,Nothing) -- delete resource
     flip (maybe fail) mu $ \u -> do
-    addrs <- resolvePeer h
-    if null addrs then fail else do
     -- add to-address to from's solicited
+    addrs <- resolvePeer h
     addToRosterFile ConfigFiles.modifySolicited (clientUser client) to addrs
     removeFromRosterFile ConfigFiles.modifyBuddies (clientUser client) to addrs
     resolved_subs <- resolvedFromRoster ConfigFiles.getSubscribers (clientUser client)
@@ -800,45 +788,46 @@ clientSubscriptionRequest state fail k stanza chan = do
     -- subscribers: "from"
     -- buddies: "to"
 
-    (ktc,ap) <- atomically $ 
-        liftM2 (,) (readTVar $ keyToChan state)
-                   (readTVar $ associatedPeers state)
-    
-    case stanzaType stanza of
-     PresenceRequestSubscription True -> do
-        hostname <- textHostName
-        let cjid = unsplitJID (Just $ clientUser client, hostname,Nothing)
-        chans <- clientCons state ktc (clientUser client)
-        forM_ chans $ \( Conn { connChan=chan }, client ) -> do
-            -- roster update ask="subscribe"
-            update <- makeRosterUpdate cjid to
-                [ ("ask","subscribe")
-                , if is_subscribed then ("subscription","from")
-                                   else ("subscription","none")
-                ]
-            sendModifiedStanzaToClient update chan
-     _ -> return ()
-
-    let dsts = Map.fromList $ map ((,()) . PeerKey) addrs
-        cdsts = ktc `Map.intersection` dsts
-    forM_ (Map.toList cdsts) $ \(pk,con) -> do
-        -- if already connected, send solicitation ...
-        -- let from = clientJID con client
-        let from = unsplitJID ( Just $ clientUser client
-                              , addrToText $ auxAddr con
-                              , Nothing )
-        mb <- rewriteJIDForPeer to
-        flip (maybe $ return ()) mb $ \(to',addr) -> do
-        dup <- cloneStanza stanza
-        sendModifiedStanzaToPeer (dup { stanzaTo = Just to'
-                                      , stanzaFrom = Just from })
-                                 (connChan con)
-    let addrm = Map.fromList (map (,()) addrs)
-    when (not . Map.null $ addrm Map.\\ ap) $ do
+    case state of
+      PresenceState { server = svVar } -> do
+
+        (ktc,(sv,conns)) <- atomically $
+            liftM2 (,) (readTVar $ keyToChan state)
+                       (takeTMVar svVar)
+
+        case stanzaType stanza of
+         PresenceRequestSubscription True -> do
+            hostname <- textHostName
+            let cjid = unsplitJID (Just $ clientUser client, hostname,Nothing)
+            chans <- clientCons state ktc (clientUser client)
+            forM_ chans $ \( Conn { connChan=chan }, client ) -> do
+                -- roster update ask="subscribe"
+                update <- makeRosterUpdate cjid to
+                    [ ("ask","subscribe")
+                    , if is_subscribed then ("subscription","from")
+                                       else ("subscription","none")
+                    ]
+                sendModifiedStanzaToClient update chan
+         _ -> return ()
+
+        let dsts = Map.fromList $ map ((,()) . PeerKey) addrs
+            cdsts = ktc `Map.intersection` dsts
+        forM_ (Map.toList cdsts) $ \(pk,con) -> do
+            -- if already connected, send solicitation ...
+            -- let from = clientJID con client
+            let from = unsplitJID ( Just $ clientUser client
+                                  , addrToText $ auxAddr con
+                                  , Nothing )
+            mb <- rewriteJIDForPeer to
+            flip (maybe $ return ()) mb $ \(to',addr) -> do
+            dup <- cloneStanza stanza
+            sendModifiedStanzaToPeer (dup { stanzaTo = Just to'
+                                          , stanzaFrom = Just from })
+                                     (connChan con)
+        let addrm = Map.fromList (map (,()) addrs)
         -- Add peer if we are not already associated ...
-        sv <- atomically $ takeTMVar $ server state
-        addPeer sv (head addrs)
-        atomically $ putTMVar (server state) sv
+        Connection.setPolicy conns h Connection.TryingToConnect
+        atomically $ putTMVar svVar (sv,conns)
 
 
 resolvedFromRoster
@@ -900,7 +889,7 @@ peerSubscriptionRequest state fail k stanza chan = do
      else do
 
     -- TODO: if peer-connection is to self, then auto-approve local user.
-    
+
     -- add from-address to to's pending
     addrs <- resolvePeer from_h
 
@@ -995,7 +984,7 @@ clientInformSubscription state fail k stanza = do
                                       , stanzaFrom = Just from })
                                  (connChan con)
         answerProbe state (Just from) pk (connChan con)
-    
+
 peerInformSubscription :: PresenceState
                         -> IO ()
                         -> ConnectionKey
diff --git a/Presence/Server.hs b/Presence/Server.hs
index da0a6973..c38aec2a 100644
--- a/Presence/Server.hs
+++ b/Presence/Server.hs
@@ -768,34 +768,59 @@ warn str = S.hPutStrLn stderr str >> hFlush stderr
 debugNoise :: Monad m => t -> m ()
 debugNoise str = return ()
 
-data TCPStatus = AwaitingRead | AwaitingWrite
+data TCPStatus = Resolving | AwaitingRead | AwaitingWrite
 
-tcpManager :: Show conkey =>
+tcpManager :: ( Show k, Ord k, Ord conkey ) =>
               (conkey -> (SockAddr, ConnectionParameters conkey u, Miliseconds))
-              -> (String -> IO (Maybe conkey))
+              -> (String -> Maybe k)
+              -> (k -> IO (Maybe conkey))
               -> Server conkey u releaseKey x
-              -> Manager TCPStatus conkey
-tcpManager grokKey parseKey sv = Manager
-    { setPolicy = \k -> \case
-        TryingToConnect -> control sv $ case grokKey k of
-                            (saddr,params,ms) -> ConnectWithEndlessRetry saddr params ms
-        OpenToConnect     -> return () -- TODO
-        RefusingToConnect -> return () -- TODO
-    , connections = fmap exportConnection <$> readTVar (conmap sv)
-    , stringToKey = parseKey
+              -> IO (Manager TCPStatus k)
+tcpManager grokKey s2k resolvKey sv = do
+    rmap <- atomically $ newTVar Map.empty -- Map k (Maybe conkey)
+    nullping <- forkPingMachine 0 0
+    return Manager {
+      setPolicy = \k -> \case
+        TryingToConnect -> join $ atomically $ do
+            r <- readTVar rmap
+            case Map.lookup k r of
+                Just {} -> return $ return () -- Connection already in progress.
+                Nothing -> do
+                    modifyTVar' rmap $ Map.insert k Nothing
+                    return $ void $ forkIO $ do
+                    myThreadId >>= flip labelThread ("resolve."++show k)
+                    mconkey <- resolvKey k
+                    case mconkey of
+                        Nothing     -> atomically $ modifyTVar' rmap $ Map.delete k
+                        Just conkey -> do
+                            control sv $ case grokKey conkey of
+                                    (saddr,params,ms) -> ConnectWithEndlessRetry saddr params ms
+        OpenToConnect     -> hPutStrLn stderr "TODO: TCP OpenToConnect"
+        RefusingToConnect -> hPutStrLn stderr "TODO: TCP RefusingToConnect"
+    , connections = do
+        c <- readTVar $ conmap sv
+        fmap (exportConnection nullping c) <$> readTVar rmap
+    , stringToKey = s2k
     , showProgress = \case
+        Resolving     -> "resolving"
         AwaitingRead  -> "awaiting inbound"
         AwaitingWrite -> "awaiting outbound"
     , showKey = show
     }
 
-exportConnection :: ConnectionRecord u -> G.Connection TCPStatus
-exportConnection (ConnectionRecord ckont cstate cdata) = G.Connection
-    { G.connStatus    = return $ case cstate of
-        SaneConnection      {} -> G.Established
-        ConnectionPair      {} -> G.Established
-        ReadOnlyConnection  {} -> G.InProgress AwaitingWrite
-        WriteOnlyConnection {} -> G.InProgress AwaitingRead
+exportConnection :: Ord conkey => PingMachine -> Map conkey (ConnectionRecord u) -> Maybe conkey -> G.Connection TCPStatus
+exportConnection nullping conmap mkey = G.Connection
+    { G.connStatus    = case mkey of
+        Nothing     -> return $ G.Dormant
+        Just conkey -> case Map.lookup conkey conmap of
+            Nothing                                    -> return $ G.InProgress Resolving
+            Just (ConnectionRecord ckont cstate cdata) -> return $ case cstate of
+                SaneConnection      {} -> G.Established
+                ConnectionPair      {} -> G.Established
+                ReadOnlyConnection  {} -> G.InProgress AwaitingWrite
+                WriteOnlyConnection {} -> G.InProgress AwaitingRead
     , G.connPolicy    = return TryingToConnect
-    , G.connPingLogic = connPingTimer cstate
+    , G.connPingLogic = case mkey >>= flip Map.lookup conmap of
+            Nothing                            -> nullping
+            Just (ConnectionRecord _ cstate _) -> connPingTimer cstate
     }
diff --git a/Presence/XMPPServer.hs b/Presence/XMPPServer.hs
index b23bb42e..c81cb9ce 100644
--- a/Presence/XMPPServer.hs
+++ b/Presence/XMPPServer.hs
@@ -1798,11 +1798,11 @@ data XMPPServer
 grokPeer :: XMPPServer -> ConnectionKey -> (SockAddr, ConnectionParameters ConnectionKey SockAddr, Miliseconds)
 grokPeer sv (PeerKey addr) = (addr, _xmpp_peer_params sv, 10000)
 
-xmppConnections :: XMPPServer -> Connection.Manager TCPStatus ConnectionKey
-xmppConnections sv = tcpManager (grokPeer sv) resolvPeer (_xmpp_sv sv)
+xmppConnections :: XMPPServer -> IO (Connection.Manager TCPStatus Text)
+xmppConnections sv = tcpManager (grokPeer sv) (Just . Text.pack) resolvPeer (_xmpp_sv sv)
  where
-    resolvPeer :: String -> IO (Maybe ConnectionKey)
-    resolvPeer str = fmap PeerKey <$> listToMaybe <$> resolvePeer (Text.pack str)
+    resolvPeer :: Text -> IO (Maybe ConnectionKey)
+    resolvPeer str = fmap PeerKey <$> listToMaybe <$> resolvePeer str
 
 xmppServer :: ( MonadResource m
               , MonadIO m