Changed Connection event to provide XML.Event streams.

1 files changed, 27 insertions, 15 deletions

diff --git a/Presence/XMPPServer.hs b/Presence/XMPPServer.hs
index 362cc8a8..b63f04c3 100644
--- a/Presence/XMPPServer.hs
+++ b/Presence/XMPPServer.hs
@@ -46,7 +46,12 @@ import Control.Monad.Trans (lift)
 import Control.Monad.IO.Class (MonadIO, liftIO)
 import Control.Monad.Fix (fix)
 import Control.Monad
-import Control.Concurrent (forkIO)
+#ifdef THREAD_DEBUG
+import Control.Concurrent.Lifted.Instrument (forkIO,myThreadId,labelThread)
+#else
+import Control.Concurrent.Lifted (forkIO,myThreadId)
+import GHC.Conc                  (labelThread)
+#endif
 import Control.Concurrent.STM
 -- import Control.Concurrent.STM.TChan
 import Network.SocketLike
@@ -80,6 +85,7 @@ import qualified System.Random
 import Data.Void (Void)
 import System.Endian (toBE32)
 import Control.Applicative
+import System.IO
 
 peerport :: PortNumber
 peerport = 5269
@@ -873,7 +879,7 @@ makePong namespace mid to from =
        ]
 
 
-xmppInbound :: Server ConnectionKey SockAddr ReleaseKey
+xmppInbound :: Server ConnectionKey SockAddr ReleaseKey XML.Event
                 -> XMPPServerParameters
                 -> ConnectionKey
                 -> SockAddr
@@ -1175,7 +1181,7 @@ presenceStanza stanza_type type_attr me jid =
                 ]
         , EventEndElement "{jabber:server}presence" ]
 
-forkConnection :: Server ConnectionKey SockAddr ReleaseKey
+forkConnection :: Server ConnectionKey SockAddr ReleaseKey XML.Event
                 -> XMPPServerParameters
                 -> ConnectionKey
                 -> SockAddr
@@ -1235,7 +1241,8 @@ forkConnection sv xmpp k laddr pingflag src snk stanzas = do
                 ,readTMVar rdone >> return (return ())
                 ]
             what
-    forkIO $ do (greet_src >> slot_src) $$ snk
+    forkIO $ do myThreadId >>= flip labelThread ("post-queue."++show k)
+                (greet_src >> slot_src) $$ snk
                 last <- atomically $ readTVar lastStanza
                 es <- while (atomically . fmap not $ Slotted.isEmpty slots)
                             (atomically . Slotted.pull $ slots)
@@ -1259,11 +1266,13 @@ forkConnection sv xmpp k laddr pingflag src snk stanzas = do
                 -- TODO: Probably some stanzas should be queued or saved for re-connect.
                 mapM_ fail $ filter notError (maybeToList last ++ es')
                 wlog $ "end post-queue fork: " ++ show k
+    
     output <- atomically newTChan
     hacks <- atomically $ newTVar Map.empty
     msgids <- atomically $ newTVar []
     forkIO $ do
         -- mapM_ (atomically . Slotted.push slots Nothing) greetPeer
+        myThreadId >>= flip labelThread ("pre-queue."++show k)
         verbosity <- xmppVerbosity xmpp
         fix $ \loop -> do
             what <- atomically $ foldr1 orElse
@@ -1338,6 +1347,7 @@ forkConnection sv xmpp k laddr pingflag src snk stanzas = do
             what
         wlog $ "end pre-queue fork: " ++ show k
     forkIO $ do
+        myThreadId >>= flip labelThread ("reader."++show k)
         -- src $$ awaitForever (lift . putStrLn . takeWhile (/=' ') . show)
         src $$ xmppInbound sv xmpp k laddr pingflag stanzas output rdone
         atomically $ putTMVar rdone ()
@@ -1437,7 +1447,7 @@ sendRoster query xmpp replyto = do
     -}
 
 
-socketFromKey :: Server ConnectionKey SockAddr ReleaseKey -> ConnectionKey -> IO SockAddr
+socketFromKey :: Server ConnectionKey SockAddr ReleaseKey XML.Event -> ConnectionKey -> IO SockAddr
 socketFromKey sv k = do
     map <- atomically $ readTVar (conmap sv)
     let mcd = Map.lookup k map
@@ -1607,7 +1617,7 @@ makeErrorStanza stanza = do
            ]
 
 monitor ::
-    Server ConnectionKey SockAddr ReleaseKey
+    Server ConnectionKey SockAddr ReleaseKey XML.Event
     -> ConnectionParameters ConnectionKey SockAddr
     -> XMPPServerParameters
     -> IO b
@@ -1619,12 +1629,10 @@ monitor sv params xmpp = do
     action <- atomically $ foldr1 orElse
       [ readTChan chan >>= \((k,u),e) -> return $ do 
         case e of
-            Connection pingflag conread conwrite -> do
-                 wlog $ tomsg k "Connection"
-                 let (xsrc,xsnk) = xmlStream conread conwrite
-                 outs <- forkConnection sv xmpp k u pingflag xsrc xsnk stanzas
-                 xmppNewConnection xmpp k u outs
-                 return ()
+            Connection pingflag xsrc xsnk -> do
+                wlog $ tomsg k "Connection"
+                outs <- forkConnection sv xmpp k u pingflag xsrc xsnk stanzas
+                xmppNewConnection xmpp k u outs
             ConnectFailure addr -> return () -- wlog $ tomsg k "ConnectFailure"
             EOF        -> do wlog $ tomsg k "EOF"
                              xmppEOF xmpp k
@@ -1775,7 +1783,7 @@ monitor sv params xmpp = do
             _ = str :: String
 
 data XMPPServer
-    = XMPPServer { _xmpp_sv :: Server ConnectionKey SockAddr ReleaseKey
+    = XMPPServer { _xmpp_sv :: Server ConnectionKey SockAddr ReleaseKey XML.Event
                  , _xmpp_peer_params :: ConnectionParameters ConnectionKey SockAddr
                  }
 
@@ -1787,7 +1795,7 @@ xmppServer :: ( MonadResource m
               , MonadIO m
               ) => XMPPServerParameters -> m XMPPServer
 xmppServer xmpp = do
-    sv <- server allocate
+    sv <- server allocate xmlStream
     -- some fuzz helps avoid simultaneity
     pingfuzz <- liftIO $ do
         gen <- System.Random.getStdGen
@@ -1803,8 +1811,12 @@ xmppServer xmpp = do
                           , timeout = 0
                           }
     liftIO $ do
-        forkIO $ monitor sv peer_params xmpp
+        forkIO $ do
+            myThreadId >>= flip labelThread ("XMPP.monitor")
+            monitor sv peer_params xmpp
+        hPutStrLn stderr $ "Starting peer listen"
         control sv (Listen peerport peer_params)
+        hPutStrLn stderr $ "Starting client listen"
         control sv (Listen clientport client_params)
     return XMPPServer { _xmpp_sv = sv, _xmpp_peer_params = peer_params }