3 files changed, 54 insertions, 56 deletions

diff --git a/Connection/Tcp.hs b/Connection/Tcp.hs
index 35083198..97b78eda 100644
--- a/Connection/Tcp.hs
+++ b/Connection/Tcp.hs
@@ -26,7 +26,7 @@ module Connection.Tcp
 
 import Data.ByteString (ByteString,hGetNonBlocking)
 import qualified Data.ByteString.Char8 as S -- ( hPutStrLn, hPutStr, pack)
-import Data.Conduit ( Source, Sink, Flush )
+import Data.Conduit ( ConduitT, Void, Flush )
 #if MIN_VERSION_containers(0,5,0)
 import qualified Data.Map.Strict as Map
 import           Data.Map.Strict (Map)
@@ -168,7 +168,7 @@ data InOrOut = In | Out
 -- | These events may be read from 'serverEvent' TChannel.
 --
 data ConnectionEvent b
-    = Connection (STM Bool) (Source IO b) (Sink (Flush b) IO ())
+    = Connection (STM Bool) (ConduitT () b IO ()) (ConduitT (Flush b) Void IO ())
     -- ^ A new connection was established
     | ConnectFailure SockAddr
     -- ^ A 'Connect' command failed.
@@ -249,7 +249,7 @@ server ::
     -- forall (m :: * -> *) a u conkey releaseKey.
     (Show conkey, MonadIO m, Ord conkey) =>
     Allocate releaseKey m
-    -> ( IO (Maybe ByteString) -> (ByteString -> IO Bool) -> ( Source IO x, Sink (Flush x) IO () ) )
+    -> ( IO (Maybe ByteString) -> (ByteString -> IO Bool) -> ( ConduitT () x IO (), ConduitT (Flush x) Void IO () ) )
     -> m (Server conkey u releaseKey x)
 server allocate sessionConduits = do
     (key,cmds) <- allocate (atomically newEmptyTMVar)
@@ -438,7 +438,7 @@ socketFamily (SockAddrInet6 _ _ _ _) = AF_INET6
 socketFamily (SockAddrUnix _)        = AF_UNIX
 
 
-conevent :: ( IO (Maybe ByteString) -> (ByteString -> IO Bool) -> ( Source IO x, Sink (Flush x) IO () ) )
+conevent :: ( IO (Maybe ByteString) -> (ByteString -> IO Bool) -> ( ConduitT () x IO (), ConduitT (Flush x) Void IO () ) )
             -> ConnectionState
             -> ConnectionEvent x
 conevent sessionConduits con = Connection pingflag read write
@@ -448,7 +448,7 @@ conevent sessionConduits con = Connection pingflag read write
 
 newConnection :: Ord a
                  => Server a u1 releaseKey x
-                 -> ( IO (Maybe ByteString) -> (ByteString -> IO Bool) -> ( Source IO x, Sink (Flush x) IO () ) )
+                 -> ( IO (Maybe ByteString) -> (ByteString -> IO Bool) -> ( ConduitT () x IO (), ConduitT (Flush x) Void IO () ) )
                  -> ConnectionParameters conkey u
                  -> a
                  -> u1
diff --git a/Presence/Nesting.hs b/Presence/Nesting.hs
index 720237fd..cf47c9fc 100644
--- a/Presence/Nesting.hs
+++ b/Presence/Nesting.hs
@@ -1,19 +1,19 @@
+{-# LANGUAGE FlexibleContexts  #-}
 {-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE FlexibleContexts #-}
 module Nesting where
 
+import Control.Monad.State.Strict
 import Data.Conduit
 import Data.Conduit.Lift
-import Data.XML.Types
-import qualified Data.Text as S
-import Control.Monad.State.Strict
 import qualified Data.List as List
+import qualified Data.Text as S
+import Data.XML.Types
 
 type Lang = S.Text
 
 data StrictList a = a :! !(StrictList a) | StrictNil
 
-data XMLState = XMLState { 
+data XMLState = XMLState {
         nestingLevel :: Int,
         langStack :: StrictList (Int,Lang)
 }
@@ -21,8 +21,8 @@ data XMLState = XMLState {
 type NestingXML o m a = ConduitM Event o (StateT XMLState m) a
 
 doNestingXML :: Monad m => NestingXML o m r -> ConduitM Event o m r
-doNestingXML m = 
-    evalStateC (XMLState 0 StrictNil) (trackNesting =$= m)
+doNestingXML m =
+    evalStateC (XMLState 0 StrictNil) (trackNesting .| m)
 
 nesting :: Monad m => NestingXML o m Int
 nesting = lift $ (return . nestingLevel) =<< get
@@ -33,18 +33,18 @@ xmlLang = fmap (fmap snd . top . langStack) (lift get)
     top ( a :! _as ) = Just a
     top _           = Nothing
 
-trackNesting :: Monad m => Conduit Event (StateT XMLState m) Event
+trackNesting :: Monad m => ConduitM Event Event (StateT XMLState m) ()
 trackNesting = awaitForever doit
  where
   doit xml = do
     XMLState lvl langs <- lift get
     lift . put $ case xml of
-              EventBeginElement _ attrs -> 
+              EventBeginElement _ attrs ->
                 case lookupLang attrs of
                   Nothing   -> XMLState (lvl+1) langs
                   Just lang -> XMLState (lvl+1) ( (lvl+1,lang) :! langs)
-              EventEndElement _     -> 
-                case langs of 
+              EventEndElement _     ->
+                case langs of
                   (llvl,_) :! ls | llvl==lvl -> XMLState (lvl-1) ls
                   _              | otherwise -> XMLState (lvl-1) langs
               _  -> XMLState lvl langs
@@ -82,7 +82,5 @@ nextElement = do
             Just (EventBeginElement _ _) -> return xml
             Just _ -> do
                 lvl' <- nesting
-                if (lvl'>=lvl) then loop 
+                if (lvl'>=lvl) then loop
                                else return Nothing
-
-
diff --git a/Presence/XMPPServer.hs b/Presence/XMPPServer.hs
index 24cfd055..2345cb67 100644
--- a/Presence/XMPPServer.hs
+++ b/Presence/XMPPServer.hs
@@ -297,11 +297,11 @@ wlog = dput XJabber
 wlogb :: ByteString -> IO ()
 wlogb = wlog . Strict8.unpack
 
-flushPassThrough :: Monad m => Conduit a m b -> Conduit (Flush a) m (Flush b)
-flushPassThrough c = getZipConduit $ ZipConduit (onlyChunks =$= mapOutput Chunk c) <* ZipConduit onlyFlushes
+flushPassThrough :: Monad m => ConduitT a b m () -> ConduitT (Flush a) (Flush b) m ()
+flushPassThrough c = getZipConduit $ ZipConduit (onlyChunks .| mapOutput Chunk c) <* ZipConduit onlyFlushes
   where
-    onlyChunks  :: Monad m => Conduit (Flush a) m a
-    onlyFlushes :: Monad m => Conduit (Flush a) m (Flush b)
+    onlyChunks  :: Monad m => ConduitT (Flush a) a m ()
+    onlyFlushes :: Monad m => ConduitT (Flush a) (Flush b) m ()
     onlyChunks           = awaitForever yieldChunk
     onlyFlushes          = awaitForever yieldFlush
     yieldFlush Flush     = yield Flush
@@ -309,17 +309,17 @@ flushPassThrough c = getZipConduit $ ZipConduit (onlyChunks =$= mapOutput Chunk
     yieldChunk (Chunk x) = yield x
     yieldChunk _         = return ()
 
-xmlStream :: ReadCommand -> WriteCommand -> ( Source IO XML.Event
-                                            , Sink (Flush XML.Event) IO () )
+xmlStream :: ReadCommand -> WriteCommand -> ( ConduitT () XML.Event IO ()
+                                            , ConduitT (Flush XML.Event) Void IO () )
 xmlStream conread conwrite = (xsrc,xsnk)
  where
-    xsrc = src $= XML.parseBytes XML.def
-    xsnk :: Sink (Flush Event) IO ()
+    xsrc = src .| XML.parseBytes XML.def
+    xsnk :: ConduitT (Flush Event) Void IO ()
     xsnk = -- XML.renderBytes XML.def =$ snk
         flushPassThrough (XML.renderBuilder XML.def)
-        =$= builderToByteStringFlush
-        =$= discardFlush
-        =$ snk
+        .| builderToByteStringFlush
+        .| discardFlush
+        .| snk
       where
         discardFlush :: Monad m => ConduitM (Flush a) a m ()
         discardFlush = awaitForever yieldChunk
@@ -365,8 +365,8 @@ copyToChannel f chan closer_stack = awaitForever copy
 prettyPrint :: ByteString -> ConduitM Event Void IO ()
 prettyPrint prefix =
         XML.renderBytes (XML.def { XML.rsPretty=True })
-        =$= CB.lines
-        =$ CL.mapM_ (wlogb . (prefix <>))
+        .| CB.lines
+        .| CL.mapM_ (wlogb . (prefix <>))
 
 swapNamespace :: Monad m => Text -> Text -> ConduitM Event Event m ()
 swapNamespace old new = awaitForever (yield . swapit old new)
@@ -403,14 +403,14 @@ fixHeaders Stanza { stanzaType=typ, stanzaTo=mto, stanzaFrom=mfrom } = do
         delAttrib akey as = filter ((/=akey) . fst) as
 
 conduitToChan
-  :: Conduit () IO Event
+  :: ConduitT () Event IO ()
      -> IO (LockedChan Event, TVar (Maybe [Event]), TMVar a)
 conduitToChan c = do
     chan <- atomically newLockedChan
     clsrs <- atomically $ newTVar (Just [])
     quitvar <- atomically $ newEmptyTMVar
     forkIO $ do
-        c =$= copyToChannel id chan clsrs $$ awaitForever (const $ return ())
+        runConduit $ c .| copyToChannel id chan clsrs .| awaitForever (const $ return ())
         atomically $ writeTVar clsrs Nothing
     return (chan,clsrs,quitvar)
 
@@ -419,7 +419,7 @@ conduitToStanza
      -> Maybe Text -- ^ id
      -> Maybe Text -- ^ from
      -> Maybe Text -- ^ to
-     -> Conduit () IO Event
+     -> ConduitT () Event IO ()
      -> IO Stanza
 conduitToStanza stype mid from to c = do
     (chan,clsrs,quitvar) <- conduitToChan c
@@ -476,7 +476,7 @@ sendModifiedStanzaToPeer stanza chan = do
  where
     old = "jabber:client"
     new = "jabber:server"
-    c = stanzaToConduit stanza =$= swapNamespace old new =$= fixHeaders stanza
+    c = stanzaToConduit stanza .| swapNamespace old new .| fixHeaders stanza
     processedType (Error cond tag) = Error cond (swapit old new tag)
     processedType x = x
 
@@ -498,7 +498,7 @@ sendModifiedStanzaToClient stanza chan = do
  where
     old = "jabber:server"
     new = "jabber:client"
-    c = stanzaToConduit stanza =$= swapNamespace old new =$= fixHeaders stanza
+    c = stanzaToConduit stanza .| swapNamespace old new .| fixHeaders stanza
     processedType (Error cond tag) = Error cond (swapit old new tag)
     processedType x = x
 
@@ -939,7 +939,7 @@ xmppInbound cdta xmpp (namespace,tellmyname,tellyourname,mkorigin) pingflag stan
         liftIO . atomically $ do
             writeLChan chan stanzaTag
             modifyTVar' clsrs (fmap (closerFor stanzaTag:))
-        copyToChannel id chan clsrs =$= do
+        copyToChannel id chan clsrs .| do
           let mid   = lookupAttrib "id"   $ tagAttrs stanzaTag
               mfrom = lookupAttrib "from" $ tagAttrs stanzaTag
               mto   = lookupAttrib "to"   $ tagAttrs stanzaTag
@@ -1218,7 +1218,7 @@ slotsToSource ::
     -> TVar (Maybe (StanzaWrap XML.Event))
     -> TVar Bool
     -> TMVar ()
-    -> Source IO (Flush XML.Event)
+    -> ConduitT () (Flush XML.Event) IO ()
 slotsToSource slots nesting lastStanza needsFlush rdone =
     fix $ \slot_src -> join . lift . atomically $ foldr1 orElse
                 [Slotted.pull slots >>= \x -> do
@@ -1255,11 +1255,11 @@ forkConnection :: Server PeerAddress ConnectionData releaseKey XML.Event
                 -> PeerAddress -- SockAddr (XXX(what?): remote for peer, local for client)
                 -> ConnectionData
                 -> FlagCommand
-                -> Source IO XML.Event
-                -> Sink (Flush XML.Event) IO ()
+                -> ConduitT () XML.Event IO ()
+                -> ConduitT (Flush XML.Event) Void IO ()
                 -> TChan Stanza
                 -> MVar ()
-                ->  IO (TChan Stanza)
+                -> IO (TChan Stanza)
 forkConnection sv xmpp saddr cdta pingflag src snk stanzas pp_mvar = do
     let auxAddr = cdAddr cdta
         clientOrServer@(namespace,tellmyname,telltheirname,_) = case auxAddr of
@@ -1283,7 +1283,7 @@ forkConnection sv xmpp saddr cdta pingflag src snk stanzas pp_mvar = do
     nesting <- atomically $ newTVar 0
     let _ = slots :: Slotted.UpdateStream XMPPState (Either (StanzaWrap XML.Event) XML.Event)
     let greet_src = do
-            CL.sourceList (greet' namespace me) =$= CL.map Chunk
+            CL.sourceList (greet' namespace me) .| CL.map Chunk
             yield Flush
         slot_src = slotsToSource slots nesting lastStanza needsFlush rdone
     -- client.PeerAddress {peerAddress = [::1]:5222}
@@ -1302,7 +1302,7 @@ forkConnection sv xmpp saddr cdta pingflag src snk stanzas pp_mvar = do
                 -- Read all slots-queued XML events or stanzas and yield them
                 -- upstream.  This should continue until the connection is
                 -- closed.
-                (greet_src >> slot_src) $$ snk
+                runConduit $ (greet_src >> slot_src) .| snk
 
                 -- Connection is now closed.  Here we handle any unsent stanzas.
                 last <- atomically $ readTVar lastStanza
@@ -1360,7 +1360,7 @@ forkConnection sv xmpp saddr cdta pingflag src snk stanzas pp_mvar = do
                                     Left  _ -> "P"
                         wlog ""
                         liftIO $ takeMVar pp_mvar
-                        stanzaToConduit dup $$ prettyPrint typ
+                        runConduit $ stanzaToConduit dup .| prettyPrint typ
                         liftIO $ putMVar pp_mvar ()
                     -- wlog $ "hacks: "++show (stanzaId stanza)
                     case stanzaType stanza of
@@ -1371,10 +1371,10 @@ forkConnection sv xmpp saddr cdta pingflag src snk stanzas pp_mvar = do
                             -- wlog $ "cache id thread: " ++ show x
                             atomically $ modifyTVar' msgids (take 3 . (x:))
                         _ -> return ()
-                    stanzaToConduit stanza =$= wrapStanzaConduit stanza
-                        $$ awaitForever
+                    runConduit $ stanzaToConduit stanza .| wrapStanzaConduit stanza
+                        .| awaitForever
                             -- TODO: PresenceStatus stanzas should be pushed to appropriate slots
-                            $ liftIO . atomically . Slotted.push slots Nothing
+                            (liftIO . atomically . Slotted.push slots Nothing)
                     case stanzaType stanza of
                         Error err tag | tagName tag=="{jabber:client}message" -> do
                             wlog $ "handling Error hacks"
@@ -1388,9 +1388,9 @@ forkConnection sv xmpp saddr cdta pingflag src snk stanzas pp_mvar = do
                             when b $ do
                             let sim = simulateChatError err (stanzaFrom stanza)
                             wlog $ "sending simulated chat for error message."
-                            CL.sourceList sim =$= wrapStanzaConduit stanza -- not quite right, but whatever
-                                $$ awaitForever
-                                    $ liftIO . atomically . Slotted.push slots Nothing
+                            runConduit $ CL.sourceList sim .| wrapStanzaConduit stanza -- not quite right, but whatever
+                                .| awaitForever
+                                    (liftIO . atomically . Slotted.push slots Nothing)
                         Error e _ -> do
                             wlog $ "no hacks for error: " ++ show e
                         _ -> return ()
@@ -1406,9 +1406,9 @@ forkConnection sv xmpp saddr cdta pingflag src snk stanzas pp_mvar = do
                     mapM_ (atomically . Slotted.push slots (Just $ PingSlot))
                           ping
                     wlog ""
-                    CL.sourceList ping0 $$ prettyPrint $ case auxAddr of
+                    runConduit $ CL.sourceList ping0 .| prettyPrint (case auxAddr of
                         Right _ -> "C<-Ping"
-                        Left  _ -> "P<-Ping "
+                        Left  _ -> "P<-Ping ")
                     loop
                 ,readTMVar rdone >> return (return ())
                 ]
@@ -1417,7 +1417,7 @@ forkConnection sv xmpp saddr cdta pingflag src snk stanzas pp_mvar = do
     forkIO $ do
         myThreadId >>= flip labelThread (lbl "xmpp-reader.")
         -- src $$ awaitForever (lift . putStrLn . takeWhile (/=' ') . show)
-        src $$ xmppInbound cdta xmpp clientOrServer pingflag stanzas output rdone
+        runConduit $ src .| xmppInbound cdta xmpp clientOrServer pingflag stanzas output rdone
         atomically $ putTMVar rdone ()
         wlog $ "end reader fork: " ++ lbl ""
     return output
@@ -1764,7 +1764,7 @@ monitor sv params xmpp = do
                             let rsc = unsplitJID (n,hostname,r) where (n,_,r) = splitJID rsc0
                             let reply = iq_bind_reply (stanzaId stanza) rsc
                             -- sendReply quitVar SetResource reply replyto
-                            let requestVersion :: Producer IO XML.Event
+                            let requestVersion :: ConduitT i XML.Event IO ()
                                 requestVersion = do
                                   yield $ EventBeginElement "{jabber:client}iq"
                                               [ attr "to" rsc
@@ -1872,7 +1872,7 @@ monitor sv params xmpp = do
                     PeerOrigin {}   -> "P"
         wlog ""
         liftIO $ takeMVar pp_mvar
-        stanzaToConduit dup $$ prettyPrint typ
+        runConduit $ stanzaToConduit dup .| prettyPrint typ
         liftIO $ putMVar pp_mvar ()
       ]
     action