asakul
/
pontarius-xmpp
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Merge branch 'master' into upstream

master
Philipp Balzarek 13 years ago
parent
32a18eecea 5ad6c7f7c0
commit
5f1e2f6c3f
10 changed files with 209 additions and 157 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 1
      source/Network/Xmpp.hs
  2. 2
      source/Network/Xmpp/Concurrent.hs
  3. 123
      source/Network/Xmpp/Concurrent/Monad.hs
  4. 43
      source/Network/Xmpp/Concurrent/Threads.hs
  5. 29
      source/Network/Xmpp/Concurrent/Types.hs
  6. 1
      source/Network/Xmpp/IM.hs
  7. 57
      source/Network/Xmpp/Stream.hs
  8. 23
      source/Network/Xmpp/Tls.hs
  9. 29
      source/Network/Xmpp/Types.hs
  10. 18
      tests/Tests.hs

1
source/Network/Xmpp.hs
Unescape View File

@ -27,6 +27,7 @@ module Network.Xmpp
( -- * Session management ( -- * Session management
Session Session
, session , session
, setConnectionClosedHandler
, StreamConfiguration(..) , StreamConfiguration(..)
, SessionConfiguration(..) , SessionConfiguration(..)
, ConnectionDetails(..) , ConnectionDetails(..)

2
source/Network/Xmpp/Concurrent.hs
Unescape View File

@ -117,7 +117,7 @@ newSession stream config = runErrorT $ do
outC <- lift newTChanIO outC <- lift newTChanIO
stanzaChan <- lift newTChanIO stanzaChan <- lift newTChanIO
iqHands <- lift $ newTVarIO (Map.empty, Map.empty) iqHands <- lift $ newTVarIO (Map.empty, Map.empty)
eh <- lift $ newTVarIO $ EventHandlers { connectionClosedHandler = sessionClosedHandler config } eh <- lift $ newTVarIO $ EventHandlers { connectionClosedHandler = onConnectionClosed config }
ros <- liftIO . newTVarIO $ Roster Nothing Map.empty ros <- liftIO . newTVarIO $ Roster Nothing Map.empty
let rosterH = if (enableRoster config) then handleRoster ros let rosterH = if (enableRoster config) then handleRoster ros
else \ _ _ -> return True else \ _ _ -> return True

123
source/Network/Xmpp/Concurrent/Monad.hs
Unescape View File

@ -2,61 +2,62 @@
{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE OverloadedStrings #-}
module Network.Xmpp.Concurrent.Monad where module Network.Xmpp.Concurrent.Monad where
import Network.Xmpp.Types import Control.Applicative ((<$>))
import Control.Concurrent
import Control.Concurrent.STM import Control.Concurrent.STM
import qualified Control.Exception.Lifted as Ex import qualified Control.Exception.Lifted as Ex
import Control.Monad.Reader import Control.Monad.State
import Network.Xmpp.Concurrent.Types import Network.Xmpp.Concurrent.Types
import Network.Xmpp.Stream import Network.Xmpp.Stream
import Network.Xmpp.Types
-- TODO: Wait for presence error? -- TODO: Wait for presence error?
-- -- | Run an XmppConMonad action in isolation. Reader and writer workers will be -- | Run an XmppConMonad action in isolation. Reader and writer workers will be
-- -- temporarily stopped and resumed with the new session details once the action -- temporarily stopped and resumed with the new session details once the action
-- -- returns. The action will run in the calling thread. Any uncaught exceptions -- returns. The action will run in the calling thread. Any uncaught exceptions
-- -- will be interpreted as connection failure. -- will be interpreted as connection failure.
-- withConnection :: XmppConMonad a -> Context -> IO (Either StreamError a) -- withConnection :: XmppConMonad a -> Context -> IO (Either StreamError a)
-- withConnection a session = do withConnection :: (Stream -> IO (b, Stream))
-- wait <- newEmptyTMVarIO -> Session
-- Ex.mask_ $ do -> IO (Either XmppFailure b)
-- -- Suspends the reader until the lock (wait) is released (set to `()'). withConnection a session = do
-- throwTo (readerThread session) $ Interrupt wait wait <- newEmptyTMVarIO
-- -- We acquire the write and stateRef locks, to make sure that this is Ex.mask_ $ do
-- -- the only thread that can write to the stream and to perform a -- Suspends the reader until the lock (wait) is released (set to `()').
-- -- withConnection calculation. Afterwards, we release the lock and throwTo (readerThread session) $ Interrupt wait
-- -- fetches an updated state. -- We acquire the write and stateRef locks, to make sure that this is
-- s <- Ex.catch -- the only thread that can write to the stream and to perform a
-- (atomically $ do -- withConnection calculation. Afterwards, we release the lock and
-- _ <- takeTMVar (writeRef session) -- fetches an updated state.
-- s <- takeTMVar (conStateRef session) s <- Ex.catch
-- putTMVar wait () (atomically $ do
-- return s _ <- takeTMVar (writeRef session)
-- ) s <- takeTMVar (streamRef session)
-- -- If we catch an exception, we have failed to take the MVars above. putTMVar wait ()
-- (\e -> atomically (putTMVar wait ()) >> return s
-- Ex.throwIO (e :: Ex.SomeException) )
-- ) -- If we catch an exception, we have failed to take the MVars above.
-- -- Run the XmppMonad action, save the (possibly updated) states, release (\e -> atomically (putTMVar wait ()) >>
-- -- the locks, and return the result. Ex.throwIO (e :: Ex.SomeException)
-- Ex.catches )
-- (do -- Run the XmppMonad action, save the (possibly updated) states, release
-- (res, s') <- runStateT a s -- the locks, and return the result.
-- atomically $ do Ex.catches
-- putTMVar (writeRef session) (cSend . sCon $ s') (do
-- putTMVar (conStateRef session) s' (res, s') <- a s
-- return $ Right res wl <- withStream' (gets $ streamSend . streamHandle) s'
-- ) atomically $ do
-- -- We treat all Exceptions as fatal. If we catch a StreamError, we putTMVar (writeRef session) wl
-- -- return it. Otherwise, we throw an exception. putTMVar (streamRef session) s'
-- [ Ex.Handler $ \e -> return $ Left (e :: StreamError) return $ Right res
-- , Ex.Handler $ \e -> runStateT xmppKillConnection s )
-- >> Ex.throwIO (e :: Ex.SomeException) -- We treat all Exceptions as fatal. If we catch a StreamError, we
-- ] -- return it. Otherwise, we throw an exception.
[ Ex.Handler $ \e -> return $ Left (e :: XmppFailure)
, Ex.Handler $ \e -> killStream s
>> Ex.throwIO (e :: Ex.SomeException)
]
-- | Executes a function to update the event handlers. -- | Executes a function to update the event handlers.
modifyHandlers :: (EventHandlers -> EventHandlers) -> Session -> IO () modifyHandlers :: (EventHandlers -> EventHandlers) -> Session -> IO ()
@ -70,12 +71,19 @@ modifyHandlers f session = atomically $ modifyTVar_ (eventHandlers session) f
x <- readTVar var x <- readTVar var
writeTVar var (g x) writeTVar var (g x)
-- | Sets the handler to be executed when the server connection is closed. -- | Changes the handler to be executed when the server connection is closed. To
setConnectionClosedHandler_ :: (XmppFailure -> Session -> IO ()) -> Session -> IO () -- avoid race conditions the initial value should be set in the configuration
setConnectionClosedHandler_ eh session = do -- when creating the session
setConnectionClosedHandler :: (XmppFailure -> Session -> IO ()) -> Session -> IO ()
setConnectionClosedHandler eh session = do
modifyHandlers (\s -> s{connectionClosedHandler = modifyHandlers (\s -> s{connectionClosedHandler =
\e -> eh e session}) session \e -> eh e session}) session
runConnectionClosedHandler :: Session -> XmppFailure -> IO ()
runConnectionClosedHandler session e = do
h <- connectionClosedHandler <$> atomically (readTVar $ eventHandlers session)
h e
-- | Run an event handler. -- | Run an event handler.
runHandler :: (EventHandlers -> IO a) -> Session -> IO a runHandler :: (EventHandlers -> IO a) -> Session -> IO a
runHandler h session = h =<< atomically (readTVar $ eventHandlers session) runHandler h session = h =<< atomically (readTVar $ eventHandlers session)
@ -84,16 +92,17 @@ runHandler h session = h =<< atomically (readTVar $ eventHandlers session)
-- | End the current Xmpp session. -- | End the current Xmpp session.
endSession :: Session -> IO () endSession :: Session -> IO ()
endSession session = do -- TODO: This has to be idempotent (is it?) endSession session = do -- TODO: This has to be idempotent (is it?)
closeConnection session _ <- flip withConnection session $ \stream -> do
_ <- closeStreams stream
return ((), stream)
stopThreads session stopThreads session
-- | Close the connection to the server. Closes the stream (by enforcing a -- | Close the connection to the server. Closes the stream (by enforcing a
-- write lock and sending a </stream:stream> element), waits (blocks) for three -- write lock and sending a </stream:stream> element), waits (blocks) for three
-- seconds, and then closes the connection. -- seconds, and then closes the connection.
closeConnection :: Session -> IO () closeConnection :: Session -> IO ()
closeConnection session = Ex.mask_ $ do closeConnection session = do
(_send, connection) <- atomically $ liftM2 (,) _ <-flip withConnection session $ \stream -> do
(takeTMVar $ writeRef session) _ <- closeStreams stream
(takeTMVar $ streamRef session) return ((), stream)
_ <- closeStreams connection runConnectionClosedHandler session StreamEndFailure
return ()

43
source/Network/Xmpp/Concurrent/Threads.hs
Unescape View File

@ -23,19 +23,28 @@ import System.Log.Logger
readWorker :: (Stanza -> IO ()) readWorker :: (Stanza -> IO ())
-> (XmppFailure -> IO ()) -> (XmppFailure -> IO ())
-> TMVar Stream -> TMVar Stream
-> IO () -> IO a
readWorker onStanza onConnectionClosed stateRef = Ex.mask_ go readWorker onStanza onConnectionClosed stateRef = forever . Ex.mask_ $ do
where
go = do s' <- Ex.catches ( do
res <- Ex.catches ( do
-- we don't know whether pull will -- we don't know whether pull will
-- necessarily be interruptible -- necessarily be interruptible
s <- atomically $ do atomically $ do
s@(Stream con) <- readTMVar stateRef s@(Stream con) <- readTMVar stateRef
scs <- streamConnectionState <$> readTMVar con scs <- streamConnectionState <$> readTMVar con
when (scs == Closed) when (stateIsClosed scs)
retry retry
return s return $ Just s
)
[ Ex.Handler $ \(Interrupt t) -> do
void $ handleInterrupts [t]
return Nothing
]
case s' of
Nothing -> return ()
Just s -> do
res <- Ex.catches (do
allowInterrupt allowInterrupt
Just <$> pullStanza s Just <$> pullStanza s
) )
@ -43,17 +52,21 @@ readWorker onStanza onConnectionClosed stateRef = Ex.mask_ go
void $ handleInterrupts [t] void $ handleInterrupts [t]
return Nothing return Nothing
, Ex.Handler $ \(e :: XmppFailure) -> do , Ex.Handler $ \(e :: XmppFailure) -> do
errorM "Pontarius.Xmpp" $ "Read error: "
++ show e
closeStreams s
onConnectionClosed e onConnectionClosed e
errorM "Pontarius.Xmpp" $ "Read error: " ++ show e
return Nothing return Nothing
] ]
case res of case res of
Nothing -> go -- Caught an exception, nothing to do. TODO: Can this happen? Nothing -> return () -- Caught an exception, nothing to
-- do. TODO: Can this happen?
Just (Left e) -> do Just (Left e) -> do
infoM "Pontarius.Xmpp.Reader" $ errorM "Pontarius.Xmpp" $ "Stanza error:" ++ show e
"Connection died: " ++ show e closeStreams s
onConnectionClosed e onConnectionClosed e
Just (Right sta) -> onStanza sta >> go Just (Right sta) -> void $ onStanza sta
where
-- Defining an Control.Exception.allowInterrupt equivalent for GHC 7 -- Defining an Control.Exception.allowInterrupt equivalent for GHC 7
-- compatibility. -- compatibility.
allowInterrupt :: IO () allowInterrupt :: IO ()
@ -67,6 +80,10 @@ readWorker onStanza onConnectionClosed stateRef = Ex.mask_ go
handleInterrupts ts = handleInterrupts ts =
Ex.catch (atomically $ forM ts takeTMVar) Ex.catch (atomically $ forM ts takeTMVar)
(\(Interrupt t) -> handleInterrupts (t:ts)) (\(Interrupt t) -> handleInterrupts (t:ts))
stateIsClosed Closed = True
stateIsClosed Finished = True
stateIsClosed _ = False
-- Two streams: input and output. Threads read from input stream and write to -- Two streams: input and output. Threads read from input stream and write to
-- output stream. -- output stream.

29
source/Network/Xmpp/Concurrent/Types.hs
Unescape View File

@ -7,14 +7,41 @@ import Control.Concurrent
import Control.Concurrent.STM import Control.Concurrent.STM
import qualified Control.Exception.Lifted as Ex import qualified Control.Exception.Lifted as Ex
import qualified Data.ByteString as BS import qualified Data.ByteString as BS
import Data.Default
import qualified Data.Map as Map import qualified Data.Map as Map
import Data.Text (Text) import Data.Text (Text)
import qualified Data.Text as Text
import Data.Typeable import Data.Typeable
import Data.XML.Types (Element) import Data.XML.Types (Element)
import Network.Xmpp.IM.Roster.Types import Network.Xmpp.IM.Roster.Types
import Network.Xmpp.Types import Network.Xmpp.Types
-- | Configuration for the @Session@ object.
data SessionConfiguration = SessionConfiguration
{ -- | Configuration for the @Stream@ object.
sessionStreamConfiguration :: StreamConfiguration
-- | Handler to be run when the session ends (for whatever reason).
, onConnectionClosed :: XmppFailure -> IO ()
-- | Function to generate the stream of stanza identifiers.
, sessionStanzaIDs :: IO (IO StanzaID)
, extraStanzaHandlers :: [StanzaHandler]
, enableRoster :: Bool
}
instance Default SessionConfiguration where
def = SessionConfiguration { sessionStreamConfiguration = def
, onConnectionClosed = \_ -> return ()
, sessionStanzaIDs = do
idRef <- newTVarIO 1
return . atomically $ do
curId <- readTVar idRef
writeTVar idRef (curId + 1 :: Integer)
return . StanzaID . Text.pack . show $ curId
, extraStanzaHandlers = []
, enableRoster = True
}
-- | Handlers to be run when the Xmpp session ends and when the Xmpp connection is -- | Handlers to be run when the Xmpp session ends and when the Xmpp connection is
-- closed. -- closed.
data EventHandlers = EventHandlers data EventHandlers = EventHandlers

1
source/Network/Xmpp/IM.hs
Unescape View File

@ -6,7 +6,6 @@ module Network.Xmpp.IM
, MessageBody(..) , MessageBody(..)
, MessageThread(..) , MessageThread(..)
, MessageSubject(..) , MessageSubject(..)
, InstantMessage (..)
, Subscription(..) , Subscription(..)
, instantMessage , instantMessage
, simpleIM , simpleIM

57
source/Network/Xmpp/Stream.hs
Unescape View File

@ -127,9 +127,10 @@ startStream = runErrorT $ do
(Secured , (Just (jid, _ ))) -> Just jid (Secured , (Just (jid, _ ))) -> Just jid
(Secured , Nothing ) -> Nothing (Secured , Nothing ) -> Nothing
(Closed , _ ) -> Nothing (Closed , _ ) -> Nothing
(Finished , _ ) -> Nothing
case streamAddress st of case streamAddress st of
Nothing -> do Nothing -> do
lift $ lift $ errorM "Pontarius.XMPP" "Server sent no hostname." lift $ lift $ errorM "Pontarius.Xmpp" "Server sent no hostname."
throwError XmppOtherFailure throwError XmppOtherFailure
Just address -> do Just address -> do
pushing pushXmlDecl pushing pushXmlDecl
@ -194,7 +195,7 @@ startStream = runErrorT $ do
void . lift . pushElement . pickleElem xpStreamError void . lift . pushElement . pickleElem xpStreamError
$ StreamErrorInfo sec Nothing el $ StreamErrorInfo sec Nothing el
void . lift $ closeStreams' void . lift $ closeStreams'
liftIO $ errorM "Pontarius.XMPP" $ "closeStreamWithError: " ++ msg liftIO $ errorM "Pontarius.Xmpp" $ "closeStreamWithError: " ++ msg
throwError XmppOtherFailure throwError XmppOtherFailure
checkchildren children = checkchildren children =
let to' = lookup "to" children let to' = lookup "to" children
@ -234,7 +235,7 @@ flattenAttrs attrs = Prelude.map (\(name, cont) ->
-- and calls xmppStartStream. -- and calls xmppStartStream.
restartStream :: StateT StreamState IO (Either XmppFailure ()) restartStream :: StateT StreamState IO (Either XmppFailure ())
restartStream = do restartStream = do
liftIO $ debugM "Pontarius.XMPP" "Restarting stream..." liftIO $ debugM "Pontarius.Xmpp" "Restarting stream..."
raw <- gets (streamReceive . streamHandle) raw <- gets (streamReceive . streamHandle)
let newSource =loopRead raw $= XP.parseBytes def let newSource =loopRead raw $= XP.parseBytes def
buffered <- liftIO . bufferSrc $ newSource buffered <- liftIO . bufferSrc $ newSource
@ -309,7 +310,7 @@ streamS _expectedTo = do -- TODO: check expectedTo
e <- lift $ elements =$ CL.head e <- lift $ elements =$ CL.head
case e of case e of
Nothing -> do Nothing -> do
lift $ lift $ errorM "Pontarius.XMPP" "streamS: Stream ended." lift $ lift $ errorM "Pontarius.Xmpp" "streamS: Stream ended."
throwError XmppOtherFailure throwError XmppOtherFailure
Just r -> streamUnpickleElem xpStreamFeatures r Just r -> streamUnpickleElem xpStreamFeatures r
@ -317,7 +318,7 @@ streamS _expectedTo = do -- TODO: check expectedTo
-- realm. -- realm.
openStream :: HostName -> StreamConfiguration -> IO (Either XmppFailure (Stream)) openStream :: HostName -> StreamConfiguration -> IO (Either XmppFailure (Stream))
openStream realm config = runErrorT $ do openStream realm config = runErrorT $ do
lift $ debugM "Pontarius.XMPP" "Opening stream..." lift $ debugM "Pontarius.Xmpp" "Opening stream..."
stream' <- createStream realm config stream' <- createStream realm config
ErrorT . liftIO $ withStream startStream stream' ErrorT . liftIO $ withStream startStream stream'
return stream' return stream'
@ -330,7 +331,7 @@ closeStreams = withStream closeStreams'
closeStreams' :: StateT StreamState IO (Either XmppFailure [Element]) closeStreams' :: StateT StreamState IO (Either XmppFailure [Element])
closeStreams' = do closeStreams' = do
lift $ debugM "Pontarius.XMPP" "Closing stream..." lift $ debugM "Pontarius.Xmpp" "Closing stream..."
send <- gets (streamSend . streamHandle) send <- gets (streamSend . streamHandle)
cc <- gets (streamClose . streamHandle) cc <- gets (streamClose . streamHandle)
void . liftIO $ send "</stream:stream>" void . liftIO $ send "</stream:stream>"
@ -338,6 +339,7 @@ closeStreams' = do
threadDelay 3000000 -- TODO: Configurable value threadDelay 3000000 -- TODO: Configurable value
void ((Ex.try cc) :: IO (Either Ex.SomeException ())) void ((Ex.try cc) :: IO (Either Ex.SomeException ()))
return () return ()
put xmppNoStream{ streamConnectionState = Finished }
collectElems [] collectElems []
where where
-- Pulls elements from the stream until the stream ends, or an error is -- Pulls elements from the stream until the stream ends, or an error is
@ -361,7 +363,7 @@ wrapIOException action = do
case r of case r of
Right b -> return $ Right b Right b -> return $ Right b
Left e -> do Left e -> do
lift $ warningM "Pontarius.XMPP" $ "wrapIOException: Exception wrapped: " ++ (show e) lift $ warningM "Pontarius.Xmpp" $ "wrapIOException: Exception wrapped: " ++ (show e)
return $ Left $ XmppIOException e return $ Left $ XmppIOException e
pushElement :: Element -> StateT StreamState IO (Either XmppFailure Bool) pushElement :: Element -> StateT StreamState IO (Either XmppFailure Bool)
@ -421,18 +423,18 @@ pullElement = do
e <- runEventsSink (elements =$ await) e <- runEventsSink (elements =$ await)
case e of case e of
Nothing -> do Nothing -> do
lift $ errorM "Pontarius.XMPP" "pullElement: Stream ended." lift $ errorM "Pontarius.Xmpp" "pullElement: Stream ended."
return . Left $ XmppOtherFailure return . Left $ XmppOtherFailure
Just r -> return $ Right r Just r -> return $ Right r
) )
[ ExL.Handler (\StreamEnd -> return $ Left StreamEndFailure) [ ExL.Handler (\StreamEnd -> return $ Left StreamEndFailure)
, ExL.Handler (\(InvalidXmppXml s) -- Invalid XML `Event' encountered, or missing element close tag , ExL.Handler (\(InvalidXmppXml s) -- Invalid XML `Event' encountered, or missing element close tag
-> do -> do
lift $ errorM "Pontarius.XMPP" $ "pullElement: Invalid XML: " ++ (show s) lift $ errorM "Pontarius.Xmpp" $ "pullElement: Invalid XML: " ++ (show s)
return . Left $ XmppOtherFailure) return . Left $ XmppOtherFailure)
, ExL.Handler $ \(e :: InvalidEventStream) , ExL.Handler $ \(e :: InvalidEventStream)
-> do -> do
lift $ errorM "Pontarius.XMPP" $ "pullElement: Invalid event stream: " ++ (show e) lift $ errorM "Pontarius.Xmpp" $ "pullElement: Invalid event stream: " ++ (show e)
return . Left $ XmppOtherFailure return . Left $ XmppOtherFailure
] ]
@ -446,7 +448,7 @@ pullUnpickle p = do
let res = unpickleElem p elem' let res = unpickleElem p elem'
case res of case res of
Left e -> do Left e -> do
lift $ errorM "Pontarius.XMPP" $ "pullUnpickle: Unpickle failed: " ++ (ppUnpickleError e) lift $ errorM "Pontarius.Xmpp" $ "pullUnpickle: Unpickle failed: " ++ (ppUnpickleError e)
return . Left $ XmppOtherFailure return . Left $ XmppOtherFailure
Right r -> return $ Right r Right r -> return $ Right r
@ -470,18 +472,21 @@ catchPush p = ExL.catch
_ -> ExL.throwIO e _ -> ExL.throwIO e
) )
-- Stream state used when there is no connection. zeroHandle :: StreamHandle
xmppNoStream :: StreamState zeroHandle = StreamHandle { streamSend = \_ -> return False
xmppNoStream = StreamState {
streamConnectionState = Closed
, streamHandle = StreamHandle { streamSend = \_ -> return False
, streamReceive = \_ -> do , streamReceive = \_ -> do
errorM "Pontarius.XMPP" "xmppNoStream: No stream on receive." errorM "Pontarius.Xmpp"
ExL.throwIO $ "xmppNoStream: Stream is closed."
XmppOtherFailure ExL.throwIO XmppOtherFailure
, streamFlush = return () , streamFlush = return ()
, streamClose = return () , streamClose = return ()
} }
-- Stream state used when there is no connection.
xmppNoStream :: StreamState
xmppNoStream = StreamState {
streamConnectionState = Closed
, streamHandle = zeroHandle
, streamEventSource = zeroSource , streamEventSource = zeroSource
, streamFeatures = StreamFeatures Nothing [] [] , streamFeatures = StreamFeatures Nothing [] []
, streamAddress = Nothing , streamAddress = Nothing
@ -494,7 +499,7 @@ xmppNoStream = StreamState {
zeroSource :: Source IO output zeroSource :: Source IO output
zeroSource = liftIO $ do zeroSource = liftIO $ do
errorM "Pontarius.Xmpp" "zeroSource" debugM "Pontarius.Xmpp" "zeroSource"
ExL.throwIO XmppOtherFailure ExL.throwIO XmppOtherFailure
createStream :: HostName -> StreamConfiguration -> ErrorT XmppFailure IO (Stream) createStream :: HostName -> StreamConfiguration -> ErrorT XmppFailure IO (Stream)
@ -705,14 +710,14 @@ srvLookup realm resolvSeed = ErrorT $ do
rest <- orderSublist sublist'' rest <- orderSublist sublist''
return $ ((priority, weight, port, domain):rest) return $ ((priority, weight, port, domain):rest)
-- Closes the connection and updates the XmppConMonad Stream state. -- | Close the connection and updates the XmppConMonad Stream state. Does
-- killStream :: Stream -> IO (Either ExL.SomeException ()) -- not send the stream end tag.
killStream :: Stream -> IO (Either XmppFailure ()) killStream :: Stream -> IO (Either XmppFailure ())
killStream = withStream $ do killStream = withStream $ do
cc <- gets (streamClose . streamHandle) cc <- gets (streamClose . streamHandle)
err <- wrapIOException cc err <- wrapIOException cc
-- (ExL.try cc :: IO (Either ExL.SomeException ())) -- (ExL.try cc :: IO (Either ExL.SomeException ()))
put xmppNoStream put xmppNoStream{ streamConnectionState = Finished }
return err return err
-- Sends an IQ request and waits for the response. If the response ID does not -- Sends an IQ request and waits for the response. If the response ID does not
@ -734,13 +739,13 @@ pushIQ iqID to tp lang body stream = runErrorT $ do
Right (IQResultS r) -> do Right (IQResultS r) -> do
unless unless
(iqID == iqResultID r) $ liftIO $ do (iqID == iqResultID r) $ liftIO $ do
liftIO $ errorM "Pontarius.XMPP" $ "pushIQ: ID mismatch (" ++ (show iqID) ++ " /= " ++ (show $ iqResultID r) ++ ")." liftIO $ errorM "Pontarius.Xmpp" $ "pushIQ: ID mismatch (" ++ (show iqID) ++ " /= " ++ (show $ iqResultID r) ++ ")."
liftIO $ ExL.throwIO XmppOtherFailure liftIO $ ExL.throwIO XmppOtherFailure
-- TODO: Log: ("In sendIQ' IDs don't match: " ++ show iqID ++ -- TODO: Log: ("In sendIQ' IDs don't match: " ++ show iqID ++
-- " /= " ++ show (iqResultID r) ++ " .") -- " /= " ++ show (iqResultID r) ++ " .")
return $ Right r return $ Right r
_ -> do _ -> do
liftIO $ errorM "Pontarius.XMPP" $ "pushIQ: Unexpected stanza type." liftIO $ errorM "Pontarius.Xmpp" $ "pushIQ: Unexpected stanza type."
throwError XmppOtherFailure throwError XmppOtherFailure
debugConduit :: (Show o, MonadIO m) => ConduitM o o m b debugConduit :: (Show o, MonadIO m) => ConduitM o o m b
@ -748,7 +753,7 @@ debugConduit = forever $ do
s' <- await s' <- await
case s' of case s' of
Just s -> do Just s -> do
liftIO $ debugM "Pontarius.XMPP" $ "debugConduit: In: " ++ (show s) liftIO $ debugM "Pontarius.Xmpp" $ "debugConduit: In: " ++ (show s)
yield s yield s
Nothing -> return () Nothing -> return ()

23
source/Network/Xmpp/Tls.hs
Unescape View File

@ -51,10 +51,13 @@ tls con = Ex.handle (return . Left . TlsError)
case sState of case sState of
Plain -> return () Plain -> return ()
Closed -> do Closed -> do
liftIO $ errorM "Pontarius.Xmpp" "startTls: The stream is closed." liftIO $ errorM "Pontarius.Xmpp.Tls" "The stream is closed."
throwError XmppNoStream
Finished -> do
liftIO $ errorM "Pontarius.Xmpp.Tls" "The stream is finished."
throwError XmppNoStream throwError XmppNoStream
Secured -> do Secured -> do
liftIO $ errorM "Pontarius.Xmpp" "startTls: The stream is already secured." liftIO $ errorM "Pontarius.Xmpp.Tls" "The stream is already secured."
throwError TlsStreamSecured throwError TlsStreamSecured
features <- lift $ gets streamFeatures features <- lift $ gets streamFeatures
case (tlsBehaviour conf, streamTls features) of case (tlsBehaviour conf, streamTls features) of
@ -67,13 +70,13 @@ tls con = Ex.handle (return . Left . TlsError)
(RefuseTls , Just True) -> throwError XmppOtherFailure (RefuseTls , Just True) -> throwError XmppOtherFailure
(RefuseTls , _ ) -> skipTls (RefuseTls , _ ) -> skipTls
where where
skipTls = liftIO $ infoM "Pontarius.Xmpp" "Skipping TLS negotiation" skipTls = liftIO $ infoM "Pontarius.Xmpp.Tls" "Skipping TLS negotiation"
startTls = do startTls = do
liftIO $ infoM "Pontarius.Xmpp" "Running StartTLS" liftIO $ infoM "Pontarius.Xmpp.Tls" "Running StartTLS"
params <- gets $ tlsParams . streamConfiguration params <- gets $ tlsParams . streamConfiguration
sent <- ErrorT $ pushElement starttlsE sent <- ErrorT $ pushElement starttlsE
unless sent $ do unless sent $ do
liftIO $ errorM "Pontarius.Xmpp" "startTls: Could not sent stanza." liftIO $ errorM "Pontarius.Xmpp.Tls" "Could not sent stanza."
throwError XmppOtherFailure throwError XmppOtherFailure
answer <- lift $ pullElement answer <- lift $ pullElement
case answer of case answer of
@ -84,8 +87,8 @@ tls con = Ex.handle (return . Left . TlsError)
liftIO $ errorM "Pontarius.Xmpp" "startTls: TLS initiation failed." liftIO $ errorM "Pontarius.Xmpp" "startTls: TLS initiation failed."
throwError XmppOtherFailure throwError XmppOtherFailure
Right r -> Right r ->
liftIO $ errorM "Pontarius.Xmpp" $ liftIO $ errorM "Pontarius.Xmpp.Tls" $
"startTls: Unexpected element: " ++ show r "Unexpected element: " ++ show r
hand <- gets streamHandle hand <- gets streamHandle
(_raw, _snk, psh, recv, ctx) <- lift $ tlsinit params (mkBackend hand) (_raw, _snk, psh, recv, ctx) <- lift $ tlsinit params (mkBackend hand)
let newHand = StreamHandle { streamSend = catchPush . psh let newHand = StreamHandle { streamSend = catchPush . psh
@ -94,7 +97,7 @@ tls con = Ex.handle (return . Left . TlsError)
, streamClose = bye ctx >> streamClose hand , streamClose = bye ctx >> streamClose hand
} }
lift $ modify ( \x -> x {streamHandle = newHand}) lift $ modify ( \x -> x {streamHandle = newHand})
liftIO $ infoM "Pontarius.Xmpp" "Stream Secured." liftIO $ infoM "Pontarius.Xmpp.Tls" "Stream Secured."
either (lift . Ex.throwIO) return =<< lift restartStream either (lift . Ex.throwIO) return =<< lift restartStream
modify (\s -> s{streamConnectionState = Secured}) modify (\s -> s{streamConnectionState = Secured})
return () return ()
@ -116,13 +119,13 @@ tlsinit :: (MonadIO m, MonadIO m1) =>
, Context , Context
) )
tlsinit params backend = do tlsinit params backend = do
liftIO $ debugM "Pontarius.Xmpp.TLS" "TLS with debug mode enabled." liftIO $ debugM "Pontarius.Xmpp.Tls" "TLS with debug mode enabled."
gen <- liftIO $ getSystemRandomGen -- TODO: Find better random source? gen <- liftIO $ getSystemRandomGen -- TODO: Find better random source?
con <- client params gen backend con <- client params gen backend
handshake con handshake con
let src = forever $ do let src = forever $ do
dt <- liftIO $ recvData con dt <- liftIO $ recvData con
liftIO $ debugM "Pontarius.Xmpp.TLS" ("In :" ++ BSC8.unpack dt) liftIO $ debugM "Pontarius.Xmpp.Tls" ("In :" ++ BSC8.unpack dt)
yield dt yield dt
let snk = do let snk = do
d <- await d <- await

29
source/Network/Xmpp/Types.hs
Unescape View File

@ -52,7 +52,6 @@ module Network.Xmpp.Types
, parseJid , parseJid
, StreamEnd(..) , StreamEnd(..)
, InvalidXmppXml(..) , InvalidXmppXml(..)
, SessionConfiguration(..)
, TlsBehaviour(..) , TlsBehaviour(..)
, AuthFailure(..) , AuthFailure(..)
) )
@ -828,6 +827,7 @@ data ConnectionState
= Closed -- ^ No stream has been established = Closed -- ^ No stream has been established
| Plain -- ^ Stream established, but not secured via TLS | Plain -- ^ Stream established, but not secured via TLS
| Secured -- ^ Stream established and secured via TLS | Secured -- ^ Stream established and secured via TLS
| Finished -- ^ Stream is closed
deriving (Show, Eq, Typeable) deriving (Show, Eq, Typeable)
-- | Defines operations for sending, receiving, flushing, and closing on a -- | Defines operations for sending, receiving, flushing, and closing on a
@ -1097,7 +1097,7 @@ data InvalidXmppXml = InvalidXmppXml String deriving (Show, Typeable)
instance Exception InvalidXmppXml instance Exception InvalidXmppXml
data ConnectionDetails = UseRealm -- ^ Use realm to resolv host data ConnectionDetails = UseRealm -- ^ Use realm to resolv host
| UseSrv HostName -- ^ Use this hostname for a SRC lookup | UseSrv HostName -- ^ Use this hostname for a SRV lookup
| UseHost HostName PortID -- ^ Use specified host | UseHost HostName PortID -- ^ Use specified host
-- | Configuration settings related to the stream. -- | Configuration settings related to the stream.
@ -1144,31 +1144,6 @@ type StanzaHandler = TChan Stanza -- ^ outgoing stanza
-> Stanza -- ^ stanza to handle -> Stanza -- ^ stanza to handle
-> IO Bool -- ^ True when processing should continue -> IO Bool -- ^ True when processing should continue
-- | Configuration for the @Session@ object.
data SessionConfiguration = SessionConfiguration
{ -- | Configuration for the @Stream@ object.
sessionStreamConfiguration :: StreamConfiguration
-- | Handler to be run when the session ends (for whatever reason).
, sessionClosedHandler :: XmppFailure -> IO ()
-- | Function to generate the stream of stanza identifiers.
, sessionStanzaIDs :: IO (IO StanzaID)
, extraStanzaHandlers :: [StanzaHandler]
, enableRoster :: Bool
}
instance Default SessionConfiguration where
def = SessionConfiguration { sessionStreamConfiguration = def
, sessionClosedHandler = \_ -> return ()
, sessionStanzaIDs = do
idRef <- newTVarIO 1
return . atomically $ do
curId <- readTVar idRef
writeTVar idRef (curId + 1 :: Integer)
return . StanzaID . Text.pack . show $ curId
, extraStanzaHandlers = []
, enableRoster = True
}
-- | How the client should behave in regards to TLS. -- | How the client should behave in regards to TLS.
data TlsBehaviour = RequireTls -- ^ Require the use of TLS; disconnect if it's data TlsBehaviour = RequireTls -- ^ Require the use of TLS; disconnect if it's
-- not offered. -- not offered.

18
tests/Tests.hs
Unescape View File

@ -175,7 +175,6 @@ runMain debug number multi = do
sendPresence presenceOnline context sendPresence presenceOnline context
thread1 <- forkIO $ autoAccept =<< dupSession context thread1 <- forkIO $ autoAccept =<< dupSession context
thread2 <- forkIO $ iqResponder =<< dupSession context thread2 <- forkIO $ iqResponder =<< dupSession context
thread2 <- forkIO $ showPresence =<< dupSession context
when active $ do when active $ do
liftIO $ threadDelay 1000000 -- Wait for the other thread to go online liftIO $ threadDelay 1000000 -- Wait for the other thread to go online
-- discoTest debug' -- discoTest debug'
@ -199,3 +198,20 @@ run i multi = do
main = do main = do
updateGlobalLogger "Pontarius.Xmpp" $ setLevel DEBUG updateGlobalLogger "Pontarius.Xmpp" $ setLevel DEBUG
run 0 True run 0 True
connectionClosedTest = do
updateGlobalLogger "Pontarius.Xmpp" $ setLevel DEBUG
let debug' = infoM "Pontarius.Xmpp"
debug' "running"
let we = testUser1
Right context <- session (Text.unpack $ domainpart we)
(Just ([scramSha1 (fromJust $ localpart we) Nothing "pwd"], resourcepart we))
config {onConnectionClosed = \e -> do
debug' $ "closed: " ++ show e
}
sendPresence presenceOnline context
forkIO $ threadDelay 3000000 >> void (closeConnection context)
forever $ threadDelay 1000000
return ()

Write Preview
Loading…
Cancel
Save