Attempt to control buffer sends with renderBytes...

2 files changed, 22 insertions, 9 deletions

diff --git a/Presence/ServerC.hs b/Presence/ServerC.hs
index 36e2d7bf..ae0bf154 100644
--- a/Presence/ServerC.hs
+++ b/Presence/ServerC.hs
@@ -18,6 +18,7 @@ import Data.ByteString.Char8
     )
 import qualified Data.ByteString.Char8 as S
     ( hPutStr
+    , hPutStrLn
     )
 import System.IO 
     ( IOMode(..)
@@ -115,18 +116,20 @@ doServer (HCons family port) g = runServer port (runConn g)
 
 packets :: MonadIO m => Handle -> Source m S.ByteString
 packets h = do
-    packet <- lift $ liftIO $ getPacket h
+    packet <- liftIO $ getPacket h
     yield packet
-    isEof <- lift $ liftIO $ hIsEOF h
+    isEof <- liftIO $ hIsEOF h
     when (not isEof) (packets h)
  where
     getPacket h = do { hWaitForInput h (-1) ; hGetNonBlocking h 1024 }
 
 outgoing :: MonadIO m => Handle -> Sink S.ByteString m ()
 outgoing h = do
+    liftIO . L.putStrLn $ "outgoing: waiting"
     mpacket <- await
+    liftIO . L.putStrLn $ "outgoing: got packet " <++> bshow mpacket
     maybe (return ()) 
-          (\r -> (lift . liftIO . S.hPutStr h $ r) >> outgoing h) 
+          (\r -> (liftIO . S.hPutStrLn h $ r) >> outgoing h) 
           mpacket
     
 
diff --git a/Presence/XMPP.hs b/Presence/XMPP.hs
index 8508c3b4..f81af7c0 100644
--- a/Presence/XMPP.hs
+++ b/Presence/XMPP.hs
@@ -29,6 +29,7 @@ import Control.Concurrent (forkIO,killThread)
 import Control.Concurrent.Async
 import Control.Exception (handle,SomeException(..),finally)
 import Control.Monad.IO.Class
+import Control.Monad.Trans.Class
 import Control.Monad.Trans.Maybe
 import Todo
 import Control.Monad as Monad
@@ -136,9 +137,12 @@ gatherElement opentag empty = gatherElement' (empty `mplus` return opentag) 1
         if (cnt>0) then gatherElement' ts' cnt'
                    else return ts' 
 
+voidMaybeT body = (>> return ()) . runMaybeT $ body
+fixMaybeT f = (>> return ()) . runMaybeT . fix $ f
+
 fromClient :: (MonadIO m, XMPPSession session) => 
     session -> TChan Commands -> Sink XML.Event m ()
-fromClient session cmdChan = (>>return ()) . runMaybeT $ do
+fromClient session cmdChan = voidMaybeT $ do
     let log  = liftIO . L.putStrLn . ("client-in: " <++>)
         send = liftIO . atomically . writeTChan cmdChan . Send
     mawait >>= guard . (==EventBeginDocument)
@@ -172,17 +176,17 @@ prettyPrint prefix xs =
         S.putStrLn prefix
         CL.sourceList xs $= renderBytes (def { rsPretty=True }) $$ CL.mapM_ S.putStr
 
-toClient :: MonadIO m => TChan Presence -> TChan Commands -> Source m XML.Event
+toClient :: MonadIO m => TChan Presence -> TChan Commands -> Source m [XML.Event]
 toClient pchan cmdChan = fix $ \loop -> do
     event <- liftIO . atomically $ 
-              orElse (fmap Left $ readTChan pchan)
+              orElse (fmap Left  $ readTChan pchan)
                      (fmap Right $ readTChan cmdChan)
     case event of
       Right QuitThread -> return ()
-      Right (Send xs)  -> mapM_ yield xs >> prettyPrint "client-out: " xs >> loop
+      Right (Send xs)  -> yield xs >> loop -- prettyPrint "client-out: " xs >> loop
       Left  presence   -> do 
         xs <- liftIO $ xmlifyPresenceForClient presence
-        Monad.mapM_ yield xs
+        yield xs
         loop
 
 handleClient
@@ -198,7 +202,7 @@ handleClient st src snk = do
     pchan <- subscribe session Nothing
     cmdChan <- atomically newTChan
 
-    writer <- async ( toClient pchan cmdChan $$ renderBytes def =$ snk )
+    writer <- async ( toClient pchan cmdChan $$ renderChunks =$ snk )
     finally         ( src $= parseBytes def $$ fromClient session cmdChan )
         $ do
             atomically $ writeTChan cmdChan QuitThread
@@ -226,3 +230,9 @@ seekRemotePeers config chan = do
     putStrLn "unimplemented: seekRemotePeers"
     -- TODO
     return ()
+
+renderChunks :: (MonadUnsafeIO m, MonadIO m) => ConduitM [Event] ByteString m ()
+renderChunks = fixMaybeT $  \loop -> do
+    xs <- mawait
+    lift . when (not . null $ xs) $ ( CL.sourceList xs $= renderBytes def $$ CL.mapM_ yield )
+    loop