copyToChan keeps a stack of closers

1 files changed, 36 insertions, 13 deletions

diff --git a/Presence/XMPPServer.hs b/Presence/XMPPServer.hs
index 95110301..a9e7e336 100644
--- a/Presence/XMPPServer.hs
+++ b/Presence/XMPPServer.hs
@@ -42,6 +42,10 @@ clientport = 5222
 
 
 -- TODO: http://xmpp.org/rfcs/rfc6120.html#rules-remote-error
+-- client connection
+-- socat script to send stanza fragment
+-- copyToChannel can keep a stack of closers to append to finish-off a stanza
+-- the TMVar () from forkConnection can be passed and with a stanza to detect interruption
 
 addrToText :: SockAddr -> Text
 addrToText (addr@(SockAddrInet _ _)) = Text.pack $ stripColon (show addr)
@@ -92,10 +96,19 @@ data Stanza
     | PongStanza { -- stanzaId :: Maybe Text
                    stanzaChan :: TChan (Maybe XML.Event) }
 
-copyToChannel f chan = awaitForever copy
+copyToChannel f chan closer_stack = awaitForever copy
  where
     copy x = do 
         liftIO . atomically $ writeTChan chan (f x)
+        case x of
+            EventBeginDocument {} -> do
+                let clsr = closerFor x
+                liftIO . atomically $
+                    modifyTVar' closer_stack (clsr:)
+            EventEndDocument {} -> do
+                liftIO . atomically $
+                    modifyTVar' closer_stack (drop 1)
+            _ -> return ()
         yield x
 
 
@@ -123,19 +136,23 @@ xmppInbound :: ConnectionKey -> FlagCommand
                 -> Source IO XML.Event
                 -> TChan Stanza
                 -> TChan Stanza
+                -> TMVar ()
                 -> Sink XML.Event IO ()
-xmppInbound k pingflag src stanzas output = doNestingXML $ do
+xmppInbound k pingflag src stanzas output donevar = doNestingXML $ do
     withXML $ \begindoc -> do
     when (begindoc==EventBeginDocument) $ do
     whenJust nextElement $ \xml -> do
     withJust (elementAttrs "stream" xml) $ \stream_attrs -> do
     fix $ \loop -> do
     -- liftIO . wlog $ "waiting for stanza."
-    chan <- liftIO $ atomically newTChan
+    (chan,clsrs) <- liftIO . atomically $
+        liftM2 (,) newTChan (newTVar [])
     whenJust nextElement $ \stanza -> do
         stanza_lvl <- nesting
-        ioWriteChan chan (Just stanza)
-        copyToChannel Just chan =$= do
+        liftIO . atomically $ do
+            writeTChan chan (Just stanza)
+            modifyTVar' clsrs (closerFor stanza:)
+        copyToChannel Just chan clsrs =$= do
           dispatch <- 
             case () of
                 _ | stanza `isServerIQOf` "get" -> grockStanzaIQGet stanza
@@ -146,10 +163,13 @@ xmppInbound k pingflag src stanzas output = doNestingXML $ do
                 let to = "todo"
                     from = "todo"
                 let pong = peerPong (stanzaId d) to from
+                liftIO $ wlog "got ping, sending pong..."
                 pongChan <- liftIO $ atomically newTChan
                 ioWriteChan output (PongStanza pongChan)
-                mapM_ (ioWriteChan pongChan . Just) pong
-                ioWriteChan pongChan Nothing
+                void . liftIO . forkIO $ do
+                    mapM_ (ioWriteChan pongChan . Just) pong
+                    ioWriteChan pongChan Nothing
+                    liftIO $ wlog "finished pong stanza"
               disp -> ioWriteChan stanzas disp
           awaitCloser stanza_lvl
         ioWriteChan chan Nothing
@@ -279,8 +299,11 @@ forkConnection k pingflag src snk stanzas = do
                         from = "todo" -- Look it up from Server object
                                       -- or pass it with Connection event.
                         mid = Just "ping"
+                        ping = peerPing mid to from
                     mapM_ (atomically . Slotted.push slots (Just $ PingSlot))
-                          (peerPing mid to from)
+                          ping
+                    wlog ""
+                    prettyPrint "P<- " ping
                     loop
                 ,readTMVar rdone >> return (return ())
                 ]
@@ -288,7 +311,7 @@ forkConnection k pingflag src snk stanzas = do
         wlog $ "end pre-queue fork: " ++ show k
     forkIO $ do
         -- src $$ awaitForever (lift . putStrLn . takeWhile (/=' ') . show)
-        src $$ xmppInbound k pingflag src stanzas output
+        src $$ xmppInbound k pingflag src stanzas output rdone
         atomically $ putTMVar rdone ()
         wlog $ "end reader fork: " ++ show k
     return output
@@ -337,12 +360,12 @@ monitor sv params = do
                 wlog $ tomsg k "ReadOnly"
                 control sv (Connect (callBackAddress k) params)
             HalfConnection Out -> wlog $ tomsg k "WriteOnly"
-            RequiresPing -> wlog $ tomsg k "RequiresPing"
+            RequiresPing -> return () -- wlog $ tomsg k "RequiresPing"
             _ -> return ()
       , readTChan stanzas >>= \stanza -> return $ do
         xs <- readUntilNothing (stanzaChan stanza)
         wlog ""
-        prettyPrint "STANZA: " xs
+        prettyPrint "P-> " xs
       ]
     action
     loop
@@ -356,8 +379,8 @@ xmppServer :: (MonadResource m, MonadIO m) => m (Server ConnectionKey,Connection
 xmppServer = do
     sv <- server
     let peer_params = (connectionDefaults peerKey)
-                          { pingInterval = 10000
-                          , timeout = 10000
+                          { pingInterval = 5000
+                          , timeout = 1000
                           , duplex = False }
         client_params = connectionDefaults clientKey
     liftIO $ do