Changed Connection event to provide XML.Event streams.

2 files changed, 64 insertions, 61 deletions

diff --git a/Presence/Server.hs b/Presence/Server.hs
index 5f85c262..46b32b81 100644
--- a/Presence/Server.hs
+++ b/Presence/Server.hs
@@ -24,6 +24,7 @@ module Server where
 
 import Data.ByteString (ByteString,hGetNonBlocking)
 import qualified Data.ByteString.Char8 as S -- ( hPutStrLn, hPutStr, pack)
+import Data.Conduit ( Source, Sink, Flush )
 #if MIN_VERSION_containers(0,5,0)
 import qualified Data.Map.Strict as Map
 import           Data.Map.Strict (Map)
@@ -160,9 +161,7 @@ data InOrOut = In | Out
 -- | These events may be read from 'serverEvent' TChannel.
 --
 data ConnectionEvent b
-    = Got b
-    -- ^ Arrival of data from a socket
-    | Connection (STM Bool) (IO (Maybe ByteString)) (ByteString -> IO Bool)
+    = Connection (STM Bool) (Source IO b) (Sink (Flush b) IO ())
     -- ^ A new connection was established
     | ConnectFailure SockAddr
     -- ^ A 'Connect' command failed.
@@ -192,16 +191,16 @@ data ConnectionRecord u
 
 -- | This object accepts commands and signals events and maintains
 -- the list of currently listening ports and established connections.
-data Server a u releaseKey
+data Server a u releaseKey b
  = Server { serverCommand    :: TMVar (ServerInstruction a u)
-          , serverEvent      :: TChan ((a,u), ConnectionEvent ByteString)
+          , serverEvent      :: TChan ((a,u), ConnectionEvent b)
           , serverReleaseKey :: releaseKey
           , conmap           :: TVar (Map a (ConnectionRecord u))
           , listenmap        :: TVar (Map PortNumber ServerHandle)
           , retrymap         :: TVar (Map SockAddr (TVar Bool,InterruptibleDelay))
           }
 
-control :: Server a u releaseKey -> ServerInstruction a u -> IO ()
+control :: Server a u releaseKey b -> ServerInstruction a u -> IO ()
 control sv = atomically . putTMVar (serverCommand sv)
 
 type Allocate releaseKey m = forall b. IO b -> (b -> IO ()) -> m (releaseKey, b)
@@ -226,8 +225,12 @@ noCleanUp io _ = ( (,) () ) `liftM` liftIO io
 -- >   let loop = do
 -- >          (_,event) <- atomically $ readTChan (serverEvent sv)
 -- >          case event of
--- >            Got bytes -> putStrLn $ "got: " ++ show bytes
--- >            _         -> return ()
+-- >            Connection getPingFlag readData writeData -> do
+-- >              forkIO $ do
+-- >                fix $ \readLoop -> do
+-- >                    readData >>= mapM $ \bytes ->
+-- >                        putStrLn $ "got: " ++ show bytes
+-- >                        readLoop
 -- >          case event of EOF -> return ()
 -- >                        _   -> loop
 -- >   liftIO loop
@@ -238,8 +241,10 @@ noCleanUp io _ = ( (,) () ) `liftM` liftIO io
 server ::
     -- forall (m :: * -> *) a u conkey releaseKey.
     (Show conkey, MonadIO m, Ord conkey) =>
-    Allocate releaseKey m -> m (Server conkey u releaseKey)
-server allocate = do
+    Allocate releaseKey m
+    -> ( IO (Maybe ByteString) -> (ByteString -> IO Bool) -> ( Source IO x, Sink (Flush x) IO () ) )
+    -> m (Server conkey u releaseKey x)
+server allocate sessionConduits = do
     (key,cmds) <- allocate (atomically newEmptyTMVar)
                            (atomically . flip putTMVar Quit)
     server <- liftIO . atomically $ do
@@ -288,6 +293,8 @@ server allocate = do
                  `fmap` atomically (readTVar $ listenmap server)
     when (not listening) $ do
 
+    hPutStrLn stderr $ "Started listening on "++show port
+
     let family  = AF_INET6
         address = case family of
             AF_INET  -> SockAddrInet port iNADDR_ANY
@@ -297,13 +304,14 @@ server allocate = do
         { serverWarn    = hPutStrLn stderr
         , serverSession = \sock _ h -> do
             (conkey,u) <- makeConnKey params sock
-            _ <- newConnection server params conkey u h In
+            _ <- newConnection server sessionConduits params conkey u h In
             return ()
         }
 
     atomically $ listenmap server `modifyTVar'` Map.insert port sserv
 
   doit server (Ignore port) = do
+    hPutStrLn stderr $ "Stopping listen on "++show port
     mb <- atomically $ do
         map <- readTVar $ listenmap server
         modifyTVar' (listenmap server) $ Map.delete port
@@ -342,7 +350,7 @@ server allocate = do
         connect sock addr
         (conkey,u) <- makeConnKey params (restrictSocket sock)
         h <- socketToHandle sock ReadWriteMode
-        newConnection server params conkey u h Out
+        newConnection server sessionConduits params conkey u h Out
         return ()
 
   doit server (ConnectWithEndlessRetry addr params interval) = do
@@ -380,7 +388,7 @@ server allocate = do
             connect sock addr
             (conkey,u) <- makeConnKey params (restrictSocket sock)
             h <- socketToHandle sock ReadWriteMode
-            threads <- newConnection server params conkey u h Out
+            threads <- newConnection server sessionConduits params conkey u h Out
             atomically $ do threadsWait threads
                             readTVar retryVar
         fin_utc <- getCurrentTime
@@ -428,30 +436,29 @@ socketFamily (SockAddrInet6 _ _ _ _) = AF_INET6
 socketFamily (SockAddrUnix _)        = AF_UNIX
 
 
-conevent :: ConnectionState -> ConnectionEvent b
-conevent con = Connection pingflag read write
+conevent :: ( IO (Maybe ByteString) -> (ByteString -> IO Bool) -> ( Source IO x, Sink (Flush x) IO () ) )
+            -> ConnectionState
+            -> ConnectionEvent x
+conevent sessionConduits con = Connection pingflag read write
  where
-    pingflag = swapTVar (pingFlag (connPingTimer con)) False
-    read =  connRead con
-    write = connWrite con
+    pingflag     = swapTVar (pingFlag (connPingTimer con)) False
+    (read,write) = sessionConduits (connRead con) (connWrite con)
 
 newConnection :: Ord a
-                 => Server a u1 releaseKey
+                 => Server a u1 releaseKey x
+                 -> ( IO (Maybe ByteString) -> (ByteString -> IO Bool) -> ( Source IO x, Sink (Flush x) IO () ) )
                  -> ConnectionParameters conkey u
                  -> a
                  -> u1
                  -> Handle
                  -> InOrOut
                  -> IO ConnectionThreads
-newConnection server params conkey u h inout = do
+newConnection server sessionConduits params conkey u h inout = do
     hSetBuffering h NoBuffering
-    let (forward,idle_ms,timeout_ms) =
+    let (idle_ms,timeout_ms) =
             case (inout,duplex params) of
-                (Out,False) -> ( const $ return ()
-                              , 0
-                              , 0 )
-                _ -> ( announce . ((conkey,u),) . Got
-                     , pingInterval params
+                (Out,False) -> ( 0, 0 )
+                _ -> ( pingInterval params
                      , timeout params )
 
     new <- do pinglogic <- forkPingMachine idle_ms timeout_ms
@@ -473,7 +480,7 @@ newConnection server params conkey u h inout = do
             (newCon,e) <- return $
               if duplex params
                 then let newcon = SaneConnection new
-                     in ( newcon, ((conkey,u), conevent newcon) )
+                     in ( newcon, ((conkey,u), conevent sessionConduits newcon) )
                 else ( case inout of
                         In  -> ReadOnlyConnection new
                         Out -> WriteOnlyConnection new
@@ -496,22 +503,6 @@ newConnection server params conkey u h inout = do
                 kont <- updateConMap conkey u new what
                 putTMVar started ()
                 return kont
-#ifdef TEST
-    -- enable this for 'Got' events
-    forkIO $ do -- inout==In || duplex params then forkIO $ do
-        warn $ "waiting gots thread: " <> bshow (conkey,inout)
-        atomically $ takeTMVar started
-        -- pingBump pinglogic -- start the ping timer
-        fix $ \loop -> do
-        -- warn $ "read thread: " <> bshow (conkey,inout)
-        mb <- threadsRead new
-        -- pingBump pinglogic
-        warn $ "got: " <> bshow (mb,(conkey,inout))
-        maybe (return ())
-              (atomically . forward >=> const loop)
-              mb
-        warn $ "quit-gots: " <> bshow (conkey,inout)
-#endif
     return new
  where
 
@@ -562,17 +553,17 @@ newConnection server params conkey u h inout = do
             announce ((conkey,u),EOF)
             connClose replaced
             let newcon = SaneConnection new
-            announce $ ((conkey,u),conevent newcon)
+            announce $ ((conkey,u),conevent sessionConduits newcon)
             return $ newcon
         else
          case replaced of
             WriteOnlyConnection w | inout==In ->
                  do let newcon = ConnectionPair new w
-                    announce ((conkey,u),conevent newcon)
+                    announce ((conkey,u),conevent sessionConduits newcon)
                     return newcon
             ReadOnlyConnection r  | inout==Out ->
                  do let newcon = ConnectionPair r new
-                    announce ((conkey,u),conevent newcon)
+                    announce ((conkey,u),conevent sessionConduits newcon)
                     return newcon
             _ -> do -- connFlush todo
                     announce ((conkey,u0), EOF)
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 }