move from exceptions as StreamHandle error indicator to XmppFailure error values.

This changes the return type of at least the following functions from IO Bool to IO (Either XmppFailure ())
  * sendIQ

  * sendIQ'

  * answerIQ

  * sendMessage

  * sendPresence

source/Network/Xmpp/Concurrent.hs

@@ -104,12 +104,12 @@ handleIQ iqHands out sta as = do
                                   False -> do
                                       didSend <- out response
                                       case didSend of
-                                          True -> do
+                                          Right () -> do
                                               atomically $ putTMVar sentRef True
-                                              return $ Just True
-                                          False -> do
+                                              return $ Just (Right ())
+                                          er@Left{} -> do
                                               atomically $ putTMVar sentRef False
-                                              return $ Just False
+                                              return $ Just er
                   writeTChan ch $ IQRequestTicket answerT iq as
                   return Nothing
         maybe (return ()) (void . out) res

source/Network/Xmpp/Concurrent/Basic.hs

@@ -11,12 +11,12 @@ import           Network.Xmpp.Stream
 import           Network.Xmpp.Types
 import           Network.Xmpp.Utilities
 
-semWrite :: WriteSemaphore -> BS.ByteString -> IO Bool
+semWrite :: WriteSemaphore -> BS.ByteString -> IO (Either XmppFailure ())
 semWrite sem bs = Ex.bracket (atomically $ takeTMVar sem)
                           (atomically . putTMVar sem)
                           ($ bs)
 
-writeStanza :: WriteSemaphore -> Stanza -> IO Bool
+writeStanza :: WriteSemaphore -> Stanza -> IO (Either XmppFailure ())
 writeStanza sem a = do
     let outData = renderElement $ nsHack (pickleElem xpStanza a)
     semWrite sem outData
@@ -24,11 +24,11 @@ writeStanza sem a = do
 
 -- | Send a stanza to the server without running plugins. (The stanza is sent as
 -- is)
-sendRawStanza :: Stanza -> Session -> IO Bool
+sendRawStanza :: Stanza -> Session -> IO (Either XmppFailure ())
 sendRawStanza a session = writeStanza (writeSemaphore session) a
 
 -- | Send a stanza to the server, managed by plugins
-sendStanza :: Stanza -> Session -> IO Bool
+sendStanza :: Stanza -> Session -> IO (Either XmppFailure ())
 sendStanza = flip sendStanza'
 
 -- | Get the channel of incoming stanzas.

source/Network/Xmpp/Concurrent/IQ.hs

@@ -1,9 +1,10 @@
 {-# OPTIONS_HADDOCK hide #-}
 module Network.Xmpp.Concurrent.IQ where
 
+import           Control.Applicative ((<$>))
 import           Control.Concurrent (forkIO)
-import           Control.Concurrent.Thread.Delay (delay)
 import           Control.Concurrent.STM
+import           Control.Concurrent.Thread.Delay (delay)
 
 import           Control.Monad
 
@@ -15,9 +16,10 @@ import           Network.Xmpp.Concurrent.Basic
 import           Network.Xmpp.Concurrent.Types
 import           Network.Xmpp.Types
 
--- | Sends an IQ, returns Just a 'TMVar' that will be filled with the first
+-- | Sends an IQ, returns Right 'TMVar' that will be filled with the first
 -- inbound IQ with a matching ID that has type @result@ or @error@ or Nothing if
--- the stanza could not be sent
+-- the stanza could not be sent.
+-- Returns Left 'XmppFailure' when sending the stanza failed
 sendIQ :: Maybe Integer -- ^ Timeout . When the timeout is reached the response
                         -- TMVar will be filled with 'IQResponseTimeout' and the
                         -- id is removed from the list of IQ handlers. 'Nothing'
@@ -28,7 +30,7 @@ sendIQ :: Maybe Integer -- ^ Timeout . When the timeout is reached the response
                          -- default)
        -> Element -- ^ The IQ body (there has to be exactly one)
        -> Session
-       -> IO (Maybe (TMVar ( Maybe (Annotated IQResponse))))
+       -> IO (Either XmppFailure (TMVar (Maybe (Annotated IQResponse))))
 sendIQ timeOut to tp lang body session = do -- TODO: Add timeout
     newId <- idGenerator session
     ref <- atomically $ do
@@ -38,15 +40,15 @@ sendIQ timeOut to tp lang body session = do -- TODO: Add timeout
           -- TODO: Check for id collisions (shouldn't happen?)
         return resRef
     res <- sendStanza (IQRequestS $ IQRequest newId Nothing to lang tp body) session
-    if res
-        then do
+    case res of
+        Right () -> do
             case timeOut of
                 Nothing -> return ()
                 Just t -> void . forkIO $ do
                           delay t
                           doTimeOut (iqHandlers session) newId ref
-            return $ Just ref
-        else return Nothing
+            return $ Right ref
+        Left e -> return $ Left e
   where
     doTimeOut handlers iqid var = atomically $ do
       p <- tryPutTMVar var Nothing
@@ -56,18 +58,28 @@ sendIQ timeOut to tp lang body session = do -- TODO: Add timeout
       return ()
 
 -- | Like 'sendIQ', but waits for the answer IQ.
-sendIQ' :: Maybe Integer
+sendIQA' :: Maybe Integer
         -> Maybe Jid
         -> IQRequestType
         -> Maybe LangTag
         -> Element
         -> Session
         -> IO (Either IQSendError (Annotated IQResponse))
-sendIQ' timeout to tp lang body session = do
+sendIQA' timeout to tp lang body session = do
     ref <- sendIQ timeout to tp lang body session
-    maybe (return $ Left IQSendError) (fmap (maybe (Left IQTimeOut) Right)
+    either (return . Left . IQSendError) (fmap (maybe (Left IQTimeOut) Right)
                                      . atomically . takeTMVar) ref
 
+-- | Like 'sendIQ', but waits for the answer IQ. Discards plugin Annotations
+sendIQ' :: Maybe Integer
+        -> Maybe Jid
+        -> IQRequestType
+        -> Maybe LangTag
+        -> Element
+        -> Session
+        -> IO (Either IQSendError IQResponse)
+sendIQ' timeout to tp lang body session = fmap fst <$> sendIQA' timeout to tp lang body session
+
 -- | Retrieves an IQ listener channel. If the namespace/'IQRequestType' is not
 -- already handled, a new 'TChan' is created and returned as a 'Right' value.
 -- Otherwise, the already existing channel will be returned wrapped in a 'Left'
@@ -119,5 +131,5 @@ dropIQChan tp ns session = do
 -- (False is returned in that case)
 answerIQ :: IQRequestTicket
          -> Either StanzaError (Maybe Element)
-         -> IO (Maybe Bool)
+         -> IO (Maybe (Either XmppFailure ()))
 answerIQ ticket = answerTicket ticket

source/Network/Xmpp/Concurrent/Message.hs

@@ -1,6 +1,7 @@
 {-# OPTIONS_HADDOCK hide #-}
 module Network.Xmpp.Concurrent.Message where
 
+import Control.Applicative((<$>))
 import Network.Xmpp.Concurrent.Types
 import Control.Concurrent.STM
 import Network.Xmpp.Types
@@ -16,46 +17,61 @@ pullMessage session = do
         MessageErrorS e -> return $ Left  (e, as)
         _ -> pullMessage session
 
+-- | Get the next received message with plugin Annotations
+getMessageA :: Session -> IO (Annotated Message)
+getMessageA = waitForMessageA (const True)
+
 -- | Get the next received message
-getMessage :: Session -> IO (Annotated Message)
-getMessage = waitForMessage (const True)
+getMessage :: Session -> IO Message
+getMessage s = fst <$> getMessageA s
 
 -- | Pulls a (non-error) message and returns it if the given predicate returns
 -- @True@.
-waitForMessage :: (Annotated Message -> Bool) -> Session -> IO (Annotated Message)
-waitForMessage f session = do
+waitForMessageA :: (Annotated Message -> Bool) -> Session -> IO (Annotated Message)
+waitForMessageA f session = do
     s <- pullMessage session
     case s of
-        Left _ -> waitForMessage f session
+        Left _ -> waitForMessageA f session
         Right m | f m -> return m
-                | otherwise -> waitForMessage f session
+                | otherwise -> waitForMessageA f session
+
+waitForMessage :: (Message -> Bool) -> Session -> IO Message
+waitForMessage f s  = fst <$> waitForMessageA (f . fst) s
 
 -- | Pulls an error message and returns it if the given predicate returns @True@.
-waitForMessageError :: (Annotated MessageError -> Bool)
+waitForMessageErrorA :: (Annotated MessageError -> Bool)
                     -> Session
                     -> IO (Annotated MessageError)
-waitForMessageError f session = do
+waitForMessageErrorA f session = do
     s <- pullMessage session
     case s of
-        Right _ -> waitForMessageError f session
+        Right _ -> waitForMessageErrorA f session
         Left  m | f m -> return m
-                | otherwise -> waitForMessageError f session
+                | otherwise -> waitForMessageErrorA f session
 
+waitForMessageError :: (MessageError -> Bool) -> Session -> IO MessageError
+waitForMessageError f s  = fst <$> waitForMessageErrorA (f . fst) s
 
 -- | Pulls a message and returns it if the given predicate returns @True@.
-filterMessages :: (Annotated MessageError -> Bool)
+filterMessagesA :: (Annotated MessageError -> Bool)
                -> (Annotated Message -> Bool)
                -> Session -> IO (Either (Annotated MessageError)
                                         (Annotated Message))
-filterMessages f g session = do
+filterMessagesA f g session = do
     s <- pullMessage session
     case s of
         Left  e | f e -> return $ Left e
-                | otherwise -> filterMessages f g session
+                | otherwise -> filterMessagesA f g session
         Right m | g m -> return $ Right m
-                | otherwise -> filterMessages f g session
+                | otherwise -> filterMessagesA f g session
+
+filterMessages :: (MessageError -> Bool)
+               -> (Message -> Bool)
+               -> Session
+               -> IO (Either (Annotated MessageError) (Annotated Message))
+filterMessages f g s = filterMessagesA (f . fst) (g . fst) s
 
 -- | Send a message stanza. Returns @False@ when the 'Message' could not be
 -- sent.
-sendMessage :: Message -> Session -> IO Bool
+sendMessage :: Message -> Session -> IO (Either XmppFailure ())
 sendMessage m session = sendStanza (MessageS m) session

source/Network/Xmpp/Concurrent/Presence.hs

@@ -1,6 +1,7 @@
 {-# OPTIONS_HADDOCK hide #-}
 module Network.Xmpp.Concurrent.Presence where
 
+import Control.Applicative ((<$>))
 import Control.Concurrent.STM
 import Network.Xmpp.Types
 import Network.Xmpp.Concurrent.Types
@@ -8,27 +9,33 @@ import Network.Xmpp.Concurrent.Basic
 
 -- | Read an element from the inbound stanza channel, discardes any non-Presence
 -- stanzas from the channel
-pullPresence :: Session -> IO (Either (Annotated PresenceError)
+pullPresenceA :: Session -> IO (Either (Annotated PresenceError)
                                       (Annotated Presence))
-pullPresence session = do
+pullPresenceA session = do
     (stanza, as) <- atomically . readTChan $ stanzaCh session
     case stanza of
         PresenceS p -> return $ Right (p, as)
         PresenceErrorS e -> return $ Left (e, as)
-        _ -> pullPresence session
+        _ -> pullPresenceA session
+
+pullPresence :: Session -> IO (Either PresenceError Presence)
+pullPresence s = either (Left . fst) (Right . fst) <$> pullPresenceA s
 
 -- | Pulls a (non-error) presence and returns it if the given predicate returns
 -- @True@.
-waitForPresence :: (Annotated Presence -> Bool)
+waitForPresenceA :: (Annotated Presence -> Bool)
                 -> Session
                 -> IO (Annotated Presence)
-waitForPresence f session = do
-    s <- pullPresence session
+waitForPresenceA f session = do
+    s <- pullPresenceA session
     case s of
-        Left _ -> waitForPresence f session
+        Left _ -> waitForPresenceA f session
         Right m | f m -> return m
-                | otherwise -> waitForPresence f session
+                | otherwise -> waitForPresenceA f session
+
+waitForPresence :: (Presence -> Bool) -> Session -> IO Presence
+waitForPresence f s = fst <$> waitForPresenceA (f . fst) s
 
 -- | Send a presence stanza.
-sendPresence :: Presence -> Session -> IO Bool
+sendPresence :: Presence -> Session -> IO (Either XmppFailure ())
 sendPresence p session = sendStanza (PresenceS p) session

source/Network/Xmpp/Concurrent/Threads.hs

@@ -45,26 +45,24 @@ readWorker onStanza onCClosed stateRef = forever . Ex.mask_ $ do
         Just s -> do
             res <- Ex.catches (do
                              allowInterrupt
-                             Just <$> pullStanza s
-                              )
-                       [ Ex.Handler $ \(Interrupt t) -> do
-                              void $ handleInterrupts [t]
-                              return Nothing
-                       , Ex.Handler $ \(e :: XmppFailure) -> do
+                             res <- pullStanza s
+                             case res of
+                                 Left e -> do
                                      errorM "Pontarius.Xmpp" $ "Read error: "
                                          ++ show e
                                      _ <- closeStreams s
                                      onCClosed e
                                      return Nothing
+                                 Right r -> return $ Just r
+                              )
+                       [ Ex.Handler $ \(Interrupt t) -> do
+                              void $ handleInterrupts [t]
+                              return Nothing
                        ]
             case res of
                 Nothing -> return () -- Caught an exception, nothing to
                                      -- do. TODO: Can this happen?
-                Just (Left e) -> do
-                    errorM "Pontarius.Xmpp" $ "Stanza error:" ++ show e
-                    _ <- closeStreams s
-                    onCClosed e
-                Just (Right sta) -> void $ onStanza sta
+                Just sta -> void $ onStanza sta
   where
     -- Defining an Control.Exception.allowInterrupt equivalent for GHC 7
     -- compatibility.
@@ -89,7 +87,7 @@ readWorker onStanza onCClosed stateRef = forever . Ex.mask_ $ do
 -- | Runs thread in XmppState monad. Returns channel of incoming and outgoing
 -- stances, respectively, and an Action to stop the Threads and close the
 -- connection.
-startThreadsWith :: TMVar (BS.ByteString -> IO Bool)
+startThreadsWith :: TMVar (BS.ByteString -> IO (Either XmppFailure ()))
                  -> (Stanza -> IO ())
                  -> TMVar EventHandlers
                  -> Stream
@@ -110,7 +108,7 @@ startThreadsWith writeSem stanzaHandler eh con = do
     killConnection threads = liftIO $ do
         _ <- atomically $ do
             _ <- takeTMVar writeSem
-            putTMVar writeSem $ \_ -> return False
+            putTMVar writeSem $ \_ -> return $ Left XmppNoStream
         _ <- forM threads killThread
         return ()
     -- Call the connection closed handlers.
@@ -122,7 +120,7 @@ startThreadsWith writeSem stanzaHandler eh con = do
 
 -- Acquires the write lock, pushes a space, and releases the lock.
 -- | Sends a blank space every 30 seconds to keep the connection alive.
-connPersist :: TMVar (BS.ByteString -> IO Bool) -> IO ()
+connPersist :: TMVar (BS.ByteString -> IO a) -> IO ()
 connPersist sem = forever $ do
     pushBS <- atomically $ takeTMVar sem
     _ <- pushBS " "

source/Network/Xmpp/Concurrent/Types.hs

@@ -20,7 +20,7 @@ import           Network.Xmpp.IM.Roster.Types
 import           Network.Xmpp.Sasl.Types
 import           Network.Xmpp.Types
 
-type StanzaHandler =  (Stanza -> IO Bool) -- ^ outgoing stanza
+type StanzaHandler =  (Stanza -> IO (Either XmppFailure ()) ) -- ^ outgoing stanza
                    -> Stanza       -- ^ stanza to handle
                    -> [Annotation] -- ^ annotations added by previous handlers
                    -> IO [(Stanza, [Annotation])]  -- ^ modified stanzas and
@@ -40,14 +40,15 @@ getAnnotation = foldr (\(Annotation a) b -> maybe b Just $ cast a) Nothing . snd
 data Plugin' = Plugin' { inHandler :: Stanza
                                       -> [Annotation]
                                       -> IO [(Stanza, [Annotation])]
-                       , outHandler :: Stanza -> IO Bool
+                       , outHandler :: Stanza -> IO (Either XmppFailure ())
                           -- | In order to allow plugins to tie the knot (Plugin
                           -- / Session) we pass the plugin the completed Session
                           -- once it exists
                        , onSessionUp :: Session -> IO ()
                        }
 
-type Plugin = (Stanza -> IO Bool) -> ErrorT XmppFailure IO Plugin'
+type Plugin = (Stanza -> IO (Either XmppFailure ()))
+              -> ErrorT XmppFailure IO Plugin'
 
 -- | Configuration for the @Session@ object.
 data SessionConfiguration = SessionConfiguration
@@ -86,7 +87,7 @@ instance Show Interrupt where show _ = "<Interrupt>"
 
 instance Ex.Exception Interrupt
 
-type WriteSemaphore = TMVar (BS.ByteString -> IO Bool)
+type WriteSemaphore = TMVar (BS.ByteString -> IO (Either XmppFailure ()))
 
 -- | The Session object represents a single session with an XMPP server. You can
 -- use 'session' to establish a session
@@ -106,7 +107,7 @@ data Session = Session
     , stopThreads :: IO ()
     , rosterRef :: TVar Roster
     , conf :: SessionConfiguration
-    , sendStanza' :: Stanza -> IO Bool
+    , sendStanza' :: Stanza -> IO (Either XmppFailure ())
     , sRealm :: HostName
     , sSaslCredentials :: Maybe (ConnectionState -> [SaslHandler] , Maybe Text)
     , reconnectWait :: TVar Int
@@ -122,17 +123,19 @@ type IQHandlers = ( Map.Map (IQRequestType, Text) (TChan IQRequestTicket)
 -- | Contains whether or not a reply has been sent, and the IQ request body to
 -- reply to.
 data IQRequestTicket = IQRequestTicket
-    { answerTicket :: Either StanzaError (Maybe Element) -> IO (Maybe Bool)
+    { answerTicket :: Either StanzaError (Maybe Element)
+                      -> IO (Maybe (Either XmppFailure ()))
                       -- ^ Return Nothing when the IQ request was already
-                      -- answered before, Just True when it was sucessfully
-                      -- answered and Just False when the answer was attempted,
-                      -- but failed (e.g. there is a connection failure)
+                      -- answered before, Just (Right ()) when it was
+                      -- sucessfully answered and Just (Left error) when the
+                      -- answer was attempted, but failed (e.g. there is a
+                      -- connection failure)
     , iqRequestBody :: IQRequest
       -- | Annotations set by plugins in receive
     , iqRequestAnnotations :: [Annotation]
     }
 
 -- | Error that can occur during sendIQ'
-data IQSendError = IQSendError -- There was an error sending the IQ stanza
+data IQSendError = IQSendError XmppFailure -- There was an error sending the IQ stanza
                  | IQTimeOut -- No answer was received during the allotted time
                    deriving (Show, Eq)

source/Network/Xmpp/IM/Roster.hs

@@ -36,7 +36,7 @@ timeout = Just 3000000 -- 3 seconds
 rosterPush :: Item -> Session -> IO (Either IQSendError (Annotated IQResponse))
 rosterPush item session = do
     let el = pickleElem xpQuery (Query Nothing [fromItem item])
-    sendIQ' timeout Nothing Set Nothing el session
+    sendIQA' timeout Nothing Set Nothing el session
 
 -- | Add or update an item to the roster.
 --
@@ -55,7 +55,7 @@ rosterAdd j n gs session = do
                                             , qiSubscription = Nothing
                                             , qiGroups = nub gs
                                             }])
-    sendIQ' timeout Nothing Set Nothing el session
+    sendIQA' timeout Nothing Set Nothing el session
 
 -- | Remove an item from the roster. Return True when the item is sucessfully
 -- removed or if it wasn't in the roster to begin with.
@@ -132,16 +132,16 @@ retrieveRoster mbOldRoster sess = do
         Left e -> do
             errorM "Pontarius.Xmpp.Roster" $ "getRoster: " ++ show e
             return Nothing
-        Right (IQResponseResult (IQResult{iqResultPayload = Just ros}), _)
+        Right (IQResponseResult IQResult{iqResultPayload = Just ros})
             -> case unpickleElem xpQuery ros of
             Left _e -> do
                 errorM "Pontarius.Xmpp.Roster" "getRoster: invalid query element"
                 return Nothing
             Right ros' -> return . Just $ toRoster ros'
-        Right (IQResponseResult (IQResult{iqResultPayload = Nothing}), _) -> do
+        Right (IQResponseResult IQResult{iqResultPayload = Nothing}) -> do
             return mbOldRoster
                 -- sever indicated that no roster updates are necessary
-        Right (IQResponseError e, _) -> do
+        Right (IQResponseError e) -> do
             errorM "Pontarius.Xmpp.Roster" $ "getRoster: server returned error"
                    ++ show e
             return Nothing

source/Network/Xmpp/Sasl/Common.hs

@@ -139,8 +139,7 @@ saslInit mechanism payload = do
     r <- lift . pushElement . saslInitE mechanism $
         Text.decodeUtf8 . B64.encode <$> payload
     case r of
-        Right True -> return ()
-        Right False -> throwError $ AuthStreamFailure XmppNoStream
+        Right () -> return ()
         Left e  -> throwError $ AuthStreamFailure e
 
 -- | Pull the next element.
@@ -205,8 +204,7 @@ respond m = do
     r <- lift . pushElement . saslResponseE . fmap (Text.decodeUtf8 . B64.encode) $ m
     case r of
         Left e -> throwError $ AuthStreamFailure e
-        Right False -> throwError $ AuthStreamFailure XmppNoStream
-        Right True -> return ()
+        Right () -> return ()
 
 -- | Run the appropriate stringprep profiles on the credentials.
 -- May fail with 'AuthStringPrepFailure'

source/Network/Xmpp/Stream.hs

@@ -5,6 +5,7 @@
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE TupleSections #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE FlexibleContexts #-}
 
 module Network.Xmpp.Stream where
 
@@ -16,7 +17,6 @@ import qualified Control.Exception.Lifted as ExL
 import           Control.Monad
 import           Control.Monad.Error
 import           Control.Monad.State.Strict
-import           Control.Monad.Trans.Resource as R
 import           Data.ByteString (ByteString)
 import qualified Data.ByteString as BS
 import qualified Data.ByteString.Char8 as BSC8
@@ -44,7 +44,6 @@ import           System.IO
 import           System.Log.Logger
 import           System.Random (randomRIO)
 import           Text.XML.Stream.Parse as XP
-import           Text.XML.Unresolved(InvalidEventStream(..))
 
 import           Network.Xmpp.Utilities
 
@@ -65,17 +64,6 @@ lmb :: [t] -> Maybe [t]
 lmb [] = Nothing
 lmb x = Just x
 
-pushing  :: MonadIO m =>
-               m (Either XmppFailure Bool)
-            -> ErrorT XmppFailure m ()
-pushing m = do
-    res <- ErrorT m
-    case res of
-        True -> return ()
-        False -> do
-            liftIO $ debugM "Pontarius.Xmpp" "Failed to send data."
-            throwError XmppOtherFailure
-
 -- Unpickles and returns a stream element.
 streamUnpickleElem :: PU [Node] a
                    -> Element
@@ -89,7 +77,7 @@ streamUnpickleElem p x = do
 
 -- This is the conduit sink that handles the stream XML events. We extend it
 -- with ErrorT capabilities.
-type StreamSink a = ErrorT XmppFailure (ConduitM Event Void IO) a
+type StreamSink a = ConduitM Event Void (ErrorT XmppFailure IO) a
 
 -- Discards all events before the first EventBeginElement.
 throwOutJunk :: Monad m => ConduitM Event a m ()
@@ -103,8 +91,8 @@ throwOutJunk = do
 -- Returns an (empty) Element from a stream of XML events.
 openElementFromEvents :: StreamSink Element
 openElementFromEvents = do
-    lift throwOutJunk
-    hd <- lift CL.head
+    throwOutJunk
+    hd <- await
     case hd of
         Just (EventBeginElement name attrs) -> return $ Element name attrs []
         _ -> do
@@ -134,15 +122,15 @@ startStream = runErrorT $ do
             lift $ lift $ errorM "Pontarius.Xmpp" "Server sent no hostname."
             throwError XmppOtherFailure
         Just address -> do
-            pushing pushXmlDecl
-            pushing . pushOpenElement . streamNSHack $
+            ErrorT $ pushXmlDecl
+            ErrorT . pushOpenElement . streamNSHack $
                 pickleElem xpStream ( "1.0"
                                     , expectedTo
                                     , Just (Jid Nothing address Nothing)
                                     , Nothing
                                     , preferredLang $ streamConfiguration st
                                     )
-    response <- ErrorT $ runEventsSink $ runErrorT $ streamS expectedTo
+    response <- ErrorT $ runEventsSink $ streamS expectedTo
     case response of
       Right (ver, from, to, sid, lt, features)
         | (Text.unpack ver) /= "1.0" ->
@@ -244,11 +232,15 @@ restartStream = do
     startStream
 
 
-sourceStreamHandle :: MonadIO m => StreamHandle -> ConduitM i ByteString m ()
+sourceStreamHandle :: (MonadIO m, MonadError XmppFailure m)
+                      => StreamHandle -> ConduitM i ByteString m ()
 sourceStreamHandle s = loopRead $ streamReceive s
   where
     loopRead rd = do
-        bs <- liftIO (rd 4096)
+        bs' <- liftIO (rd 4096)
+        bs <- case bs' of
+            Left e -> throwError e
+            Right r -> return r
         if BS.null bs
             then return ()
             else do
@@ -260,25 +252,31 @@ sourceStreamHandle s = loopRead $ streamReceive s
 -- We buffer sources because we don't want to lose data when multiple
 -- xml-entities are sent with the same packet and we don't want to eternally
 -- block the StreamState while waiting for data to arrive
-bufferSrc :: MonadIO m => Source IO o -> IO (ConduitM i o m ())
+bufferSrc :: Source (ErrorT XmppFailure IO) o
+          -> IO (ConduitM i o (ErrorT XmppFailure IO) ())
 bufferSrc src = do
     ref <- newTMVarIO $ DCI.ResumableSource src (return ())
     let go = do
-            dt <- liftIO $ Ex.bracketOnError (atomically $ takeTMVar ref)
-                                          (\_ -> atomically . putTMVar ref $
-                                                 DCI.ResumableSource zeroSource
-                                                                     (return ())
-                                          )
+            dt <- liftIO $ Ex.bracketOnError
+                      (atomically $ takeTMVar ref)
+                      (\_ -> atomically . putTMVar ref $ zeroResumableSource)
                       (\s -> do
-                                                (s', dt) <- s $$++ CL.head
+                            res <- runErrorT (s $$++ await)
+                            case res of
+                                Left e -> do
+                                    atomically $ putTMVar ref zeroResumableSource
+                                    return $ Left e
+                                Right (s',b) -> do
                                     atomically $ putTMVar ref s'
-                                                return dt
+                                    return $ Right b
                       )
             case dt of
-                Nothing -> return ()
-                Just d -> yield d >> go
+                Left e -> throwError e
+                Right Nothing -> return ()
+                Right (Just d) -> yield d >> go
     return go
-
+  where
+    zeroResumableSource = DCI.ResumableSource zeroSource (return ())
 
 -- Reads the (partial) stream:stream and the server features from the stream.
 -- Returns the (unvalidated) stream attributes, the unparsed element, or
@@ -302,7 +300,7 @@ streamS _expectedTo = do -- TODO: check expectedTo
   where
     xmppStreamHeader :: StreamSink (Either Element (Text, Maybe Jid, Maybe Jid, Maybe Text.Text, Maybe LangTag))
     xmppStreamHeader = do
-        lift throwOutJunk
+        throwOutJunk
         -- Get the stream:stream element (or whatever it is) from the server,
         -- and validate what we get.
         el <- openElementFromEvents -- May throw `XmppOtherFailure' if an
@@ -312,7 +310,7 @@ streamS _expectedTo = do -- TODO: check expectedTo
             Right r -> return $ Right r
     xmppStreamFeatures :: StreamSink StreamFeatures
     xmppStreamFeatures = do
-        e <- lift $ elements =$ CL.head
+        e <- elements =$ await
         case e of
             Nothing -> do
                 lift $ lift $ errorM "Pontarius.Xmpp" "streamS: Stream ended."
@@ -367,21 +365,22 @@ debugOut :: MonadIO m => ByteString -> m ()
 debugOut outData = liftIO $ debugM "Pontarius.Xmpp"
              ("Out: " ++ (Text.unpack . Text.decodeUtf8 $ outData))
 
-wrapIOException :: IO a -> StateT StreamState IO (Either XmppFailure a)
+wrapIOException :: MonadIO m =>
+                   IO a -> m (Either XmppFailure a)
 wrapIOException action = do
     r <- liftIO $ tryIOError action
     case r of
         Right b -> return $ Right b
         Left e -> do
-            lift $ warningM "Pontarius.Xmpp" $ "wrapIOException: Exception wrapped: " ++ (show e)
+            liftIO $ warningM "Pontarius.Xmpp" $ "wrapIOException: Exception wrapped: " ++ (show e)
             return $ Left $ XmppIOException e
 
-pushElement :: Element -> StateT StreamState IO (Either XmppFailure Bool)
+pushElement :: Element -> StateT StreamState IO (Either XmppFailure ())
 pushElement x = do
     send <- gets (streamSend . streamHandle)
     let outData = renderElement $ nsHack x
     debugOut outData
-    wrapIOException $ send  outData
+    lift $ send outData
   where
     -- HACK: We remove the "jabber:client" namespace because it is set as
     -- default in the stream. This is to make isode's M-LINK server happy and
@@ -400,53 +399,46 @@ nsHack e@(Element{elementName = n})
     mapNSHack nd = nd
 
 -- | Encode and send stanza
-pushStanza :: Stanza -> Stream -> IO (Either XmppFailure Bool)
+pushStanza :: Stanza -> Stream -> IO (Either XmppFailure ())
 pushStanza s = withStream' . pushElement $ pickleElem xpStanza s
 
 -- XML documents and XMPP streams SHOULD be preceeded by an XML declaration.
 -- UTF-8 is the only supported XMPP encoding. The standalone document
 -- declaration (matching "SDDecl" in the XML standard) MUST NOT be included in
 -- XMPP streams. RFC 6120 defines XMPP only in terms of XML 1.0.
-pushXmlDecl :: StateT StreamState IO (Either XmppFailure Bool)
+pushXmlDecl :: StateT StreamState IO (Either XmppFailure ())
 pushXmlDecl = do
     con <- gets streamHandle
-    wrapIOException $ (streamSend con) "<?xml version=\"1.0\" encoding=\"UTF-8\" ?>"
+    lift $ streamSend con "<?xml version=\"1.0\" encoding=\"UTF-8\" ?>"
 
-pushOpenElement :: Element -> StateT StreamState IO (Either XmppFailure Bool)
+pushOpenElement :: Element -> StateT StreamState IO (Either XmppFailure ())
 pushOpenElement e = do
     send <- gets (streamSend . streamHandle)
     let outData = renderOpenElement e
     debugOut outData
-    wrapIOException $ send outData
+    lift $ send outData
 
 -- `Connect-and-resumes' the given sink to the stream source, and pulls a
 -- `b' value.
-runEventsSink :: Sink Event IO b -> StateT StreamState IO b
+runEventsSink :: Sink Event (ErrorT XmppFailure IO) b
+              -> StateT StreamState IO (Either XmppFailure b)
 runEventsSink snk = do -- TODO: Wrap exceptions?
     src <- gets streamEventSource
-    r <- liftIO $ src $$ snk
-    return  r
+    lift . runErrorT $ src $$ snk
 
 pullElement :: StateT StreamState IO (Either XmppFailure Element)
 pullElement = do
-    ExL.catches (do
     e <- runEventsSink (elements =$ await)
     case e of
-            Nothing -> do
-                lift $ errorM "Pontarius.Xmpp" "pullElement: Stream ended."
-                return . Left $ XmppOtherFailure
-            Just r -> return $ Right r
-        )
-        [ ExL.Handler (\StreamEnd -> return $ Left StreamEndFailure)
-        , ExL.Handler (\(InvalidXmppXml s) -- Invalid XML `Event' encountered, or missing element close tag
-                     -> do
-                            lift $ errorM "Pontarius.Xmpp" $ "pullElement: Invalid XML: " ++ (show s)
-                            return . Left $ XmppOtherFailure)
-        , ExL.Handler $ \(e :: InvalidEventStream)
-                     -> do
-                            lift $ errorM "Pontarius.Xmpp" $ "pullElement: Invalid event stream: " ++ (show e)
+        Left l -> do
+            liftIO .  errorM "Pontarius.Xmpp" $
+                  "Error while retrieving XML element: " ++ show l
+            return $ Left l
+
+        Right Nothing -> do
+            liftIO $ errorM "Pontarius.Xmpp" "pullElement: Stream ended."
             return . Left $ XmppOtherFailure
-        ]
+        Right (Just r) -> return $ Right r
 
 -- Pulls an element and unpickles it.
 pullUnpickle :: PU [Node] a -> StateT StreamState IO (Either XmppFailure a)
@@ -473,21 +465,21 @@ pullStanza = withStream' $ do
 
 -- Performs the given IO operation, catches any errors and re-throws everything
 -- except 'ResourceVanished' and IllegalOperation, in which case it will return False instead
-catchPush :: IO () -> IO Bool
+catchPush :: IO () -> IO (Either XmppFailure ())
 catchPush p = ExL.catch
-    (p >> return True)
+    (p >> return (Right ()))
     (\e -> case GIE.ioe_type e of
-         GIE.ResourceVanished -> return False
-         GIE.IllegalOperation -> return False
+         GIE.ResourceVanished -> return . Left $ XmppIOException e
+         GIE.IllegalOperation -> return . Left $ XmppIOException e
          _ -> ExL.throwIO e
     )
 
 zeroHandle :: StreamHandle
-zeroHandle = StreamHandle { streamSend = \_ -> return False
+zeroHandle = StreamHandle { streamSend = \_ -> return (Left XmppNoStream)
                           , streamReceive = \_ -> do
                                  errorM "Pontarius.Xmpp"
                                         "xmppNoStream: Stream is closed."
-                                 ExL.throwIO XmppOtherFailure
+                                 return $ Left XmppNoStream
                           , streamFlush = return ()
                           , streamClose = return ()
                           }
@@ -507,14 +499,16 @@ xmppNoStream = StreamState {
     , streamConfiguration = def
     }
 
-zeroSource :: Source IO output
-zeroSource = liftIO $ do
-    debugM "Pontarius.Xmpp" "zeroSource"
-    ExL.throwIO XmppOtherFailure
+zeroSource :: Source (ErrorT XmppFailure IO) a
+zeroSource = do
+    liftIO $ debugM "Pontarius.Xmpp" "zeroSource"
+    throwError XmppNoStream
 
 handleToStreamHandle :: Handle -> StreamHandle
-handleToStreamHandle h = StreamHandle { streamSend = \d -> catchPush $ BS.hPut h d
-                                      , streamReceive = \n -> BS.hGetSome h n
+handleToStreamHandle h = StreamHandle { streamSend = \d ->
+                                         wrapIOException $ BS.hPut h d
+                                      , streamReceive = \n ->
+                                         wrapIOException $ BS.hGetSome h n
                                       , streamFlush = hFlush h
                                       , streamClose = hClose h
                                       }
@@ -547,9 +541,9 @@ createStream realm config = do
             lift $ debugM "Pontarius.Xmpp" "Did not acquire handle."
             throwError TcpConnectionFailure
   where
-    logConduit :: Conduit ByteString IO ByteString
+    logConduit :: MonadIO m => Conduit ByteString m ByteString
     logConduit = CL.mapM $ \d -> do
-        debugM "Pontarius.Xmpp" $ "In: " ++ (BSC8.unpack d) ++
+        liftIO . debugM "Pontarius.Xmpp" $ "In: " ++ (BSC8.unpack d) ++
             "."
         return d
 
@@ -780,7 +774,7 @@ pushIQ :: Text
        -> Stream
        -> IO (Either XmppFailure (Either IQError IQResult))
 pushIQ iqID to tp lang body stream = runErrorT $ do
-    pushing $ pushStanza
+    ErrorT $ pushStanza
         (IQRequestS $ IQRequest iqID Nothing to lang tp body) stream
     res <- lift $ pullStanza stream
     case res of
@@ -807,7 +801,7 @@ debugConduit = forever $ do
             yield s
         Nothing -> return ()
 
-elements :: R.MonadThrow m => Conduit Event m Element
+elements :: MonadError XmppFailure m => Conduit Event m Element
 elements = do
         x <- await
         case x of
@@ -816,11 +810,11 @@ elements = do
                                                  elements
             -- This might be an XML error if the end element tag is not
             -- "</stream>". TODO: We might want to check this at a later time
-            Just (EventEndElement _) -> lift $ R.monadThrow StreamEnd
+            Just (EventEndElement _) -> throwError StreamEndFailure
             Just (EventContent (ContentText ct)) | Text.all isSpace ct ->
                 elements
             Nothing -> return ()
-            _ -> lift $ R.monadThrow $ InvalidXmppXml $ "not an element: " ++ show x
+            _ -> throwError $ XmppInvalidXml $ "not an element: " ++ show x
   where
     many' f =
         go id
@@ -834,8 +828,7 @@ elements = do
         (y, ns) <- many' goN
         if y == Just (EventEndElement n)
             then return $ Element n as $ compressNodes ns
-            else lift $ R.monadThrow $ InvalidXmppXml $
-                                         "Missing close tag: " ++ show n
+            else throwError . XmppInvalidXml $ "Missing close tag: " ++ show n
     goN = do
         x <- await
         case x of

source/Network/Xmpp/Tls.hs

@@ -34,7 +34,10 @@ mkBackend con = Backend { backendSend = \bs -> void (streamSend con bs)
     bufferReceive recv n = BS.concat `liftM` (go n)
       where
         go m = do
-            bs <- recv m
+            mbBs <- recv m
+            bs <- case mbBs of
+                Left e -> Ex.throwIO e
+                Right r -> return r
             case BS.length bs of
                 0 -> return []
                 l -> if l < m
@@ -46,7 +49,9 @@ starttlsE = Element "{urn:ietf:params:xml:ns:xmpp-tls}starttls" [] []
 
 -- | Checks for TLS support and run starttls procedure if applicable
 tls :: Stream -> IO (Either XmppFailure ())
-tls con = Ex.handle (return . Left . TlsError)
+tls con = fmap join -- We can have Left values both from exceptions and the
+                    -- error monad. Join unifies them into one error layer
+          . wrapExceptions
           . flip withStream con
           . runErrorT $ do
     conf <- gets $ streamConfiguration
@@ -77,10 +82,7 @@ tls con = Ex.handle (return . Left . TlsError)
     startTls = do
         liftIO $ infoM "Pontarius.Xmpp.Tls" "Running StartTLS"
         params <- gets $ tlsParams . streamConfiguration
-        sent <- ErrorT $ pushElement starttlsE
-        unless sent $ do
-            liftIO $ errorM "Pontarius.Xmpp.Tls" "Could not sent stanza."
-            throwError XmppOtherFailure
+        ErrorT $ pushElement starttlsE
         answer <- lift $ pullElement
         case answer of
             Left e -> throwError e
@@ -95,7 +97,7 @@ tls con = Ex.handle (return . Left . TlsError)
         hand <- gets streamHandle
         (_raw, _snk, psh, recv, ctx) <- lift $ tlsinit params (mkBackend hand)
         let newHand = StreamHandle { streamSend = catchPush . psh
-                                   , streamReceive = recv
+                                   , streamReceive = wrapExceptions . recv
                                    , streamFlush = contextFlush ctx
                                    , streamClose = bye ctx >> streamClose hand
                                    }
@@ -173,7 +175,19 @@ connectTls config params host = do
     let hand = handleToStreamHandle h
     (_raw, _snk, psh, recv, ctx) <- tlsinit params $ mkBackend hand
     return $ StreamHandle { streamSend = catchPush . psh
-                          , streamReceive = recv
+                          , streamReceive = wrapExceptions . recv
                           , streamFlush = contextFlush ctx
                           , streamClose = bye ctx >> streamClose hand
                           }
+
+wrapExceptions :: IO a -> IO (Either XmppFailure a)
+wrapExceptions f = Ex.catches (liftM Right $ f)
+                 [ Ex.Handler $ return . Left . XmppIOException
+                 , Ex.Handler $ wrap . XmppTlsError
+                 , Ex.Handler $ wrap . XmppTlsConnectionNotEstablished
+                 , Ex.Handler $ wrap . XmppTlsTerminated
+                 , Ex.Handler $ wrap . XmppTlsHandshakeFailed
+                 , Ex.Handler $ return . Left
+                 ]
+  where
+    wrap = return . Left . TlsError

source/Network/Xmpp/Types.hs

@@ -38,6 +38,7 @@ module Network.Xmpp.Types
     , StanzaErrorCondition(..)
     , StanzaErrorType(..)
     , XmppFailure(..)
+    , XmppTlsError(..)
     , StreamErrorCondition(..)
     , Version(..)
     , StreamHandle(..)
@@ -62,8 +63,6 @@ module Network.Xmpp.Types
     , domainpart
     , resourcepart
     , parseJid
-    , StreamEnd(..)
-    , InvalidXmppXml(..)
     , TlsBehaviour(..)
     , AuthFailure(..)
     )
@@ -486,6 +485,12 @@ data StreamErrorInfo = StreamErrorInfo
     , errorXml       :: !(Maybe Element)
     } deriving (Show, Eq)
 
+data XmppTlsError = XmppTlsError TLSError
+                  | XmppTlsConnectionNotEstablished ConnectionNotEstablished
+                  | XmppTlsTerminated Terminated
+                  | XmppTlsHandshakeFailed HandshakeFailed
+                    deriving (Show, Eq, Typeable)
+
 -- | Signals an XMPP stream error or another unpredicted stream-related
 -- situation. This error is fatal, and closes the XMPP stream.
 data XmppFailure = StreamErrorFailure StreamErrorInfo -- ^ An error XML stream
@@ -507,7 +512,7 @@ data XmppFailure = StreamErrorFailure StreamErrorInfo -- ^ An error XML stream
                                         -- failed.
                  | XmppIllegalTcpDetails -- ^ The TCP details provided did not
                                          -- validate.
-                 | TlsError TLSError -- ^ An error occurred in the
+                 | TlsError XmppTlsError -- ^ An error occurred in the
                                      -- TLS layer
                  | TlsNoServerSupport -- ^ The server does not support
                                       -- the use of TLS
@@ -522,6 +527,7 @@ data XmppFailure = StreamErrorFailure StreamErrorInfo -- ^ An error XML stream
                                     -- the log.
                  | XmppIOException IOException -- ^ An 'IOException'
                                                -- occurred
+                 | XmppInvalidXml String -- ^ Received data is not valid XML
                  deriving (Show, Eq, Typeable)
 
 instance Exception XmppFailure
@@ -649,9 +655,10 @@ data ConnectionState
 -- | Defines operations for sending, receiving, flushing, and closing on a
 -- stream.
 data StreamHandle =
-    StreamHandle { streamSend :: BS.ByteString -> IO Bool -- ^ Sends may not
+    StreamHandle { streamSend :: BS.ByteString
+                                 -> IO (Either XmppFailure ()) -- ^ Sends may not
                                                           -- interleave
-                 , streamReceive :: Int -> IO BS.ByteString
+                 , streamReceive :: Int -> IO (Either XmppFailure BS.ByteString)
                    -- This is to hold the state of the XML parser (otherwise we
                    -- will receive EventBeginDocument events and forget about
                    -- name prefixes). (TODO: Clarify)
@@ -665,7 +672,7 @@ data StreamState = StreamState
       -- | Functions to send, receive, flush, and close the stream
     , streamHandle :: StreamHandle
       -- | Event conduit source, and its associated finalizer
-    , streamEventSource :: Source IO Event
+    , streamEventSource :: Source (ErrorT XmppFailure IO) Event
       -- | Stream features advertised by the server
     , streamFeatures :: !StreamFeatures -- TODO: Maybe?
       -- | The hostname or IP specified for the connection
@@ -993,13 +1000,6 @@ resourceprepProfile = SP.Profile { SP.maps = [SP.b1]
                                  , SP.shouldCheckBidi = True
                                  }
 
-data StreamEnd = StreamEnd deriving (Typeable, Show)
-instance Exception StreamEnd
-
-data InvalidXmppXml = InvalidXmppXml String deriving (Show, Typeable)
-
-instance Exception InvalidXmppXml
-
 data ConnectionDetails = UseRealm -- ^ Use realm to resolv host
                        | UseSrv HostName -- ^ Use this hostname for a SRV lookup
                        | UseHost HostName PortID -- ^ Use specified host

source/Network/Xmpp/Xep/ServiceDiscovery.hs

@@ -79,7 +79,7 @@ handleInfoRequest
   :: [Identity]
      -> [Text.Text]
      -> Map.Map Text.Text ([Identity], [Text.Text])
-     -> (Stanza -> IO Bool)
+     -> (Stanza -> IO (Either XmppFailure ()) )
      -> Stanza
      -> [Annotation]
      -> IO [Annotated Stanza]
@@ -126,7 +126,7 @@ queryItems timeout to' node session' = do
 
 
 handleItemsRequest :: (Maybe Text.Text -> IO (Maybe [Item]))
-                   -> (Stanza -> IO Bool)
+                   -> (Stanza -> IO (Either XmppFailure ()))
                    -> Stanza
                    -> [Annotation]
                    -> IO [Annotated Stanza]