Added Chunk/Flush logic to outgoing connections in xmppServer

2 files changed, 69 insertions, 18 deletions

diff --git a/Presence/EventUtil.hs b/Presence/EventUtil.hs
new file mode 100644
index 00000000..e62f8afc
--- /dev/null
+++ b/Presence/EventUtil.hs
@@ -0,0 +1,23 @@
+{-# LANGUAGE OverloadedStrings #-}
+module EventUtil where
+
+import Control.Monad
+import Data.XML.Types as XML
+
+getStreamName (EventBeginElement name _) = name
+
+isEventBeginElement (EventBeginElement {}) = True
+isEventBeginElement _ = False
+
+isEventEndElement (EventEndElement {}) = True
+isEventEndElement _ = False
+
+-- Note: This function ignores name space qualification
+elementAttrs expected (EventBeginElement name attrs)
+    | nameLocalName name==expected 
+    = return attrs
+elementAttrs _ _ = mzero
+
+streamP name = Name name (Just "http://etherx.jabber.org/streams") (Just "stream")
+
+attr   name value = (name,[ContentText value])
diff --git a/xmppServer.hs b/xmppServer.hs
index e8864706..cbd30bda 100644
--- a/xmppServer.hs
+++ b/xmppServer.hs
@@ -18,6 +18,7 @@ import qualified Data.ByteString.Char8 as Strict8
 import Data.Conduit
 import qualified Data.Conduit.List as CL
 import qualified Data.Conduit.Binary as CB
+import Data.Conduit.Blaze (builderToByteStringFlush)
 
 import qualified Text.XML.Stream.Render as XML
 import qualified Text.XML.Stream.Parse as XML
@@ -28,6 +29,7 @@ import Data.Monoid ( (<>) )
 import qualified Control.Concurrent.STM.UpdateStream as Slotted
 import ControlMaybe
 import NestingXML
+import EventUtil
 import Server
 
 
@@ -37,20 +39,24 @@ wlogb s = Strict8.putStrLn s
 
 control sv = atomically . putTMVar (serverCommand sv)
 
--- Note: This function ignores name space qualification
-elementAttrs expected (EventBeginElement name attrs)
-    | nameLocalName name==expected 
-    = return attrs
-elementAttrs _ _ = mzero
+discardFlush :: Monad m => ConduitM (Flush a) a m ()
+discardFlush = awaitForever $ \x -> do
+    let unchunk (Chunk a) = a
+        ischunk (Chunk _) = True
+        ischunk _         = False
+    when (ischunk x) $ yield (unchunk x)
 
-getStreamName (EventBeginElement name _) = name
 
 xmlStream :: ReadCommand -> WriteCommand -> ( Source IO XML.Event
-                                            , Sink XML.Event IO () )
+                                            , Sink (Flush XML.Event) IO () )
 xmlStream conread conwrite = (xsrc,xsnk)
  where
     xsrc = src $= XML.parseBytes XML.def
-    xsnk = XML.renderBytes XML.def =$ snk
+    xsnk = -- XML.renderBytes XML.def =$ snk
+        XML.renderBuilderFlush XML.def 
+        =$= builderToByteStringFlush 
+        =$= discardFlush 
+        =$ snk
 
     src = do
         v <- lift conread
@@ -117,16 +123,25 @@ chanContents ch = do
               x
 
 
-isEventBeginElement (EventBeginElement {}) = True
-isEventBeginElement _ = False
+greetPeer =
+    [ EventBeginDocument
+    , EventBeginElement (streamP "stream")
+        [ attr "xmlns"   "jabber:server"
+        , attr "version" "1.0"
+        ]
+    ]
+
+goodbyePeer = 
+    [ EventEndElement (streamP "stream")
+    , EventEndDocument
+    ]
+
 
-isEventEndElement (EventEndElement {}) = True
-isEventEndElement _ = False
 
 forkConnection :: ConnectionKey
                 -> FlagCommand
                 -> Source IO XML.Event
-                -> Sink XML.Event IO ()
+                -> Sink (Flush XML.Event) IO ()
                 -> TChan Stanza
                 ->  IO (TChan Stanza)
 forkConnection k pingflag src snk stanzas = do
@@ -139,16 +154,29 @@ forkConnection k pingflag src snk stanzas = do
     slots <- atomically $ Slotted.new isEventBeginElement isEventEndElement
     let _ = slots :: Slotted.UpdateStream () XML.Event
     let slot_src = do
-            what <- lift . atomically $ orElse
-                (Slotted.pull slots >>= \x -> return $ do
-                    yield x
-                    slot_src)
-                (readTMVar rdone >> return (return ()))
+            CL.sourceList greetPeer =$= CL.map Chunk
+            yield Flush
+            needsFlush <- lift . atomically $ newTVar False
+            what <- lift . atomically $ foldr1 orElse
+                [Slotted.pull slots >>= \x -> do
+                    writeTVar needsFlush True
+                    return $ do
+                    yield (Chunk x)
+                    slot_src
+                ,do Slotted.isEmpty slots >>= check
+                    readTVar needsFlush >>= check
+                    writeTVar needsFlush False
+                    return $ do
+                    yield Flush
+                    slot_src
+                ,readTMVar rdone >> return (return ())
+                ]
             what
     forkIO $ do slot_src $$ snk
                 wlog $ "end post-queue fork: " ++ show k
     output <- atomically newTChan
     forkIO $ do
+        -- mapM_ (atomically . Slotted.push slots Nothing) greetPeer
         fix $ \loop -> do
             what <- atomically $ foldr1 orElse
                 [readTChan output >>= \stanza -> return $ do