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

renamed XMPPThread to XMPP 

renamed Thread to Session
split runThreaded in newSession and WithNewSession
master
Philipp Balzarek 14 years ago
parent
f2ab31fe71
commit
70d9b5b47d
9 changed files with 90 additions and 96 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. 16
      src/Network/XMPP.hs
  2. 2
      src/Network/XMPP/Bind.hs
  3. 4
      src/Network/XMPP/Concurrent.hs
  4. 6
      src/Network/XMPP/Concurrent/IQ.hs
  5. 42
      src/Network/XMPP/Concurrent/Monad.hs
  6. 64
      src/Network/XMPP/Concurrent/Threads.hs
  7. 46
      src/Network/XMPP/Concurrent/Types.hs
  8. 4
      src/Network/XMPP/Monad.hs
  9. 2
      src/Network/XMPP/Session.hs

16
src/Network/XMPP.hs
Unescape View File

@ -34,7 +34,7 @@
module Network.XMPP module Network.XMPP
( -- * Session management ( -- * Session management
xmppNewSession withNewSession
, connect , connect
, startTLS , startTLS
, auth , auth
@ -132,7 +132,7 @@ module Network.XMPP
, iqRequestPayload , iqRequestPayload
, iqResultPayload , iqResultPayload
-- * Threads -- * Threads
, XMPPThread , XMPP
, forkXMPP , forkXMPP
-- * Misc -- * Misc
, exampleParams , exampleParams
@ -155,10 +155,6 @@ import Network.XMPP.Types
import Control.Monad.Error import Control.Monad.Error
-- | Create a new, pristine session without an active connection.
xmppNewSession :: XMPPThread a -> IO (a, XMPPConState)
xmppNewSession = withNewSession . runThreaded
-- | Connect to host with given address. -- | Connect to host with given address.
xmppConnect :: HostName -> Text -> XMPPConMonad (Either StreamError ()) xmppConnect :: HostName -> Text -> XMPPConMonad (Either StreamError ())
xmppConnect address hostname = xmppRawConnect address hostname >> xmppStartStream xmppConnect address hostname = xmppRawConnect address hostname >> xmppStartStream
@ -166,18 +162,16 @@ xmppConnect address hostname = xmppRawConnect address hostname >> xmppStartStre
-- | Attempts to secure the connection using TLS. Will return -- | Attempts to secure the connection using TLS. Will return
-- 'TLSNoServerSupport' when the server does not offer TLS or does not -- 'TLSNoServerSupport' when the server does not offer TLS or does not
-- expect it at this time. -- expect it at this time.
startTLS :: TLS.TLSParams -> XMPPThread (Either XMPPTLSError ()) startTLS :: TLS.TLSParams -> XMPP (Either XMPPTLSError ())
startTLS = withConnection . xmppStartTLS startTLS = withConnection . xmppStartTLS
-- | Authenticate to the server with the given username and password -- | Authenticate to the server with the given username and password
-- and bind a resource -- and bind a resource
auth :: Text.Text -- ^ The username auth :: Text.Text -- ^ The username
-> Text.Text -- ^ The password -> Text.Text -- ^ The password
-> Maybe Text -- ^ The desired resource or 'Nothing' to let the server -> Maybe Text -- ^ The desired resource or 'Nothing' to let the server
-- assign one -- assign one
-> XMPPThread (Either SaslError Text.Text) -> XMPP (Either SaslError Text.Text)
auth username passwd resource = runErrorT $ do auth username passwd resource = runErrorT $ do
ErrorT . withConnection $ xmppSASL username passwd ErrorT . withConnection $ xmppSASL username passwd
res <- lift $ xmppBind resource res <- lift $ xmppBind resource
@ -185,5 +179,5 @@ auth username passwd resource = runErrorT $ do
return res return res
-- | Connect to an xmpp server -- | Connect to an xmpp server
connect :: HostName -> Text -> XMPPThread (Either StreamError ()) connect :: HostName -> Text -> XMPP (Either StreamError ())
connect address hostname = withConnection $ xmppConnect address hostname connect address hostname = withConnection $ xmppConnect address hostname

2
src/Network/XMPP/Bind.hs
Unescape View File

@ -40,7 +40,7 @@ jidP = bindP $ xpElemNodes "jid" (xpContent xpPrim)
-- server-generated resource and extract the JID from the non-error -- server-generated resource and extract the JID from the non-error
-- response. -- response.
xmppBind :: Maybe Text -> XMPPThread Text xmppBind :: Maybe Text -> XMPP Text
xmppBind rsrc = do xmppBind rsrc = do
answer <- sendIQ' Nothing Set Nothing (bindBody rsrc) answer <- sendIQ' Nothing Set Nothing (bindBody rsrc)
let (Right IQResult{iqResultPayload = Just b}) = answer -- TODO: Error handling let (Right IQResult{iqResultPayload = Just b}) = answer -- TODO: Error handling

4
src/Network/XMPP/Concurrent.hs
Unescape View File

@ -1,6 +1,6 @@
module Network.XMPP.Concurrent module Network.XMPP.Concurrent
( Thread ( Session
, XMPPThread , XMPP
, module Network.XMPP.Concurrent.Monad , module Network.XMPP.Concurrent.Monad
, module Network.XMPP.Concurrent.Threads , module Network.XMPP.Concurrent.Threads
, module Network.XMPP.Concurrent.IQ , module Network.XMPP.Concurrent.IQ

6
src/Network/XMPP/Concurrent/IQ.hs
Unescape View File

@ -17,7 +17,7 @@ sendIQ :: Maybe JID -- ^ Recipient (to)
-> IQRequestType -- ^ IQ type (Get or Set) -> IQRequestType -- ^ IQ type (Get or Set)
-> Maybe LangTag -- ^ Language tag of the payload (Nothing for default) -> Maybe LangTag -- ^ Language tag of the payload (Nothing for default)
-> Element -- ^ The iq body (there has to be exactly one) -> Element -- ^ The iq body (there has to be exactly one)
-> XMPPThread (TMVar IQResponse) -> XMPP (TMVar IQResponse)
sendIQ to tp lang body = do -- TODO: add timeout sendIQ to tp lang body = do -- TODO: add timeout
newId <- liftIO =<< asks idGenerator newId <- liftIO =<< asks idGenerator
handlers <- asks iqHandlers handlers <- asks iqHandlers
@ -35,14 +35,14 @@ sendIQ' :: Maybe JID
-> IQRequestType -> IQRequestType
-> Maybe LangTag -> Maybe LangTag
-> Element -> Element
-> XMPPThread IQResponse -> XMPP IQResponse
sendIQ' to tp lang body = do sendIQ' to tp lang body = do
ref <- sendIQ to tp lang body ref <- sendIQ to tp lang body
liftIO . atomically $ takeTMVar ref liftIO . atomically $ takeTMVar ref
answerIQ :: (IQRequest, TVar Bool) answerIQ :: (IQRequest, TVar Bool)
-> Either StanzaError (Maybe Element) -> Either StanzaError (Maybe Element)
-> XMPPThread Bool -> XMPP Bool
answerIQ ((IQRequest iqid from _to lang _tp bd), sentRef) answer = do answerIQ ((IQRequest iqid from _to lang _tp bd), sentRef) answer = do
out <- asks outCh out <- asks outCh
let response = case answer of let response = case answer of

42
src/Network/XMPP/Concurrent/Monad.hs
Unescape View File

@ -23,7 +23,7 @@ import Network.XMPP.Monad
-- combination was alread handled -- combination was alread handled
listenIQChan :: IQRequestType -- ^ type of IQs to receive (Get / Set) listenIQChan :: IQRequestType -- ^ type of IQs to receive (Get / Set)
-> Text -- ^ namespace of the child element -> Text -- ^ namespace of the child element
-> XMPPThread (Maybe ( TChan (IQRequest, TVar Bool))) -> XMPP (Maybe ( TChan (IQRequest, TVar Bool)))
listenIQChan tp ns = do listenIQChan tp ns = do
handlers <- asks iqHandlers handlers <- asks iqHandlers
liftIO . atomically $ do liftIO . atomically $ do
@ -39,7 +39,7 @@ listenIQChan tp ns = do
-- | get the inbound stanza channel, duplicates from master if necessary -- | get the inbound stanza channel, duplicates from master if necessary
-- please note that once duplicated it will keep filling up, call -- please note that once duplicated it will keep filling up, call
-- 'dropMessageChan' to allow it to be garbage collected -- 'dropMessageChan' to allow it to be garbage collected
getMessageChan :: XMPPThread (TChan (Either MessageError Message)) getMessageChan :: XMPP (TChan (Either MessageError Message))
getMessageChan = do getMessageChan = do
mChR <- asks messagesRef mChR <- asks messagesRef
mCh <- liftIO $ readIORef mChR mCh <- liftIO $ readIORef mChR
@ -52,7 +52,7 @@ getMessageChan = do
Just mCh' -> return mCh' Just mCh' -> return mCh'
-- | see 'getMessageChan' -- | see 'getMessageChan'
getPresenceChan :: XMPPThread (TChan (Either PresenceError Presence)) getPresenceChan :: XMPP (TChan (Either PresenceError Presence))
getPresenceChan = do getPresenceChan = do
pChR <- asks presenceRef pChR <- asks presenceRef
pCh <- liftIO $ readIORef pChR pCh <- liftIO $ readIORef pChR
@ -66,40 +66,40 @@ getPresenceChan = do
-- | Drop the local end of the inbound stanza channel -- | Drop the local end of the inbound stanza channel
-- from our context so it can be GC-ed -- from our context so it can be GC-ed
dropMessageChan :: XMPPThread () dropMessageChan :: XMPP ()
dropMessageChan = do dropMessageChan = do
r <- asks messagesRef r <- asks messagesRef
liftIO $ writeIORef r Nothing liftIO $ writeIORef r Nothing
-- | see 'dropMessageChan' -- | see 'dropMessageChan'
dropPresenceChan :: XMPPThread () dropPresenceChan :: XMPP ()
dropPresenceChan = do dropPresenceChan = do
r <- asks presenceRef r <- asks presenceRef
liftIO $ writeIORef r Nothing liftIO $ writeIORef r Nothing
-- | Read an element from the inbound stanza channel, acquiring a copy -- | Read an element from the inbound stanza channel, acquiring a copy
-- of the channel as necessary -- of the channel as necessary
pullMessage :: XMPPThread (Either MessageError Message) pullMessage :: XMPP (Either MessageError Message)
pullMessage = do pullMessage = do
c <- getMessageChan c <- getMessageChan
liftIO $ atomically $ readTChan c liftIO $ atomically $ readTChan c
-- | Read an element from the inbound stanza channel, acquiring a copy -- | Read an element from the inbound stanza channel, acquiring a copy
-- of the channel as necessary -- of the channel as necessary
pullPresence :: XMPPThread (Either PresenceError Presence) pullPresence :: XMPP (Either PresenceError Presence)
pullPresence = do pullPresence = do
c <- getPresenceChan c <- getPresenceChan
liftIO $ atomically $ readTChan c liftIO $ atomically $ readTChan c
-- | Send a stanza to the server -- | Send a stanza to the server
sendS :: Stanza -> XMPPThread () sendS :: Stanza -> XMPP ()
sendS a = do sendS a = do
out <- asks outCh out <- asks outCh
liftIO . atomically $ writeTChan out a liftIO . atomically $ writeTChan out a
return () return ()
-- | Fork a new thread -- | Fork a new thread
forkXMPP :: XMPPThread () -> XMPPThread ThreadId forkXMPP :: XMPP () -> XMPP ThreadId
forkXMPP a = do forkXMPP a = do
thread <- ask thread <- ask
mCH' <- liftIO $ newIORef Nothing mCH' <- liftIO $ newIORef Nothing
@ -110,7 +110,7 @@ forkXMPP a = do
filterMessages :: (MessageError -> Bool) filterMessages :: (MessageError -> Bool)
-> (Message -> Bool) -> (Message -> Bool)
-> XMPPThread (Either MessageError Message) -> XMPP (Either MessageError Message)
filterMessages f g = do filterMessages f g = do
s <- pullMessage s <- pullMessage
case s of case s of
@ -119,7 +119,7 @@ filterMessages f g = do
Right m | g m -> return $ Right m Right m | g m -> return $ Right m
| otherwise -> filterMessages f g | otherwise -> filterMessages f g
waitForMessage :: (Message -> Bool) -> XMPPThread Message waitForMessage :: (Message -> Bool) -> XMPP Message
waitForMessage f = do waitForMessage f = do
s <- pullMessage s <- pullMessage
case s of case s of
@ -127,7 +127,7 @@ waitForMessage f = do
Right m | f m -> return m Right m | f m -> return m
| otherwise -> waitForMessage f | otherwise -> waitForMessage f
waitForMessageError :: (MessageError -> Bool) -> XMPPThread MessageError waitForMessageError :: (MessageError -> Bool) -> XMPP MessageError
waitForMessageError f = do waitForMessageError f = do
s <- pullMessage s <- pullMessage
case s of case s of
@ -135,7 +135,7 @@ waitForMessageError f = do
Left m | f m -> return m Left m | f m -> return m
| otherwise -> waitForMessageError f | otherwise -> waitForMessageError f
waitForPresence :: (Presence -> Bool) -> XMPPThread Presence waitForPresence :: (Presence -> Bool) -> XMPP Presence
waitForPresence f = do waitForPresence f = do
s <- pullPresence s <- pullPresence
case s of case s of
@ -149,7 +149,7 @@ waitForPresence f = do
-- The Action will run in the calling thread/ -- The Action will run in the calling thread/
-- NB: This will /not/ catch any exceptions. If you action dies, deadlocks -- NB: This will /not/ catch any exceptions. If you action dies, deadlocks
-- or otherwisely exits abnormaly the XMPP session will be dead. -- or otherwisely exits abnormaly the XMPP session will be dead.
withConnection :: XMPPConMonad a -> XMPPThread a withConnection :: XMPPConMonad a -> XMPP a
withConnection a = do withConnection a = do
readerId <- asks readerThread readerId <- asks readerThread
stateRef <- asks conStateRef stateRef <- asks conStateRef
@ -167,36 +167,36 @@ withConnection a = do
return res return res
-- | Send a presence Stanza -- | Send a presence Stanza
sendPresence :: Presence -> XMPPThread () sendPresence :: Presence -> XMPP ()
sendPresence = sendS . PresenceS sendPresence = sendS . PresenceS
-- | Send a Message Stanza -- | Send a Message Stanza
sendMessage :: Message -> XMPPThread () sendMessage :: Message -> XMPP ()
sendMessage = sendS . MessageS sendMessage = sendS . MessageS
modifyHandlers :: (EventHandlers -> EventHandlers) -> XMPPThread () modifyHandlers :: (EventHandlers -> EventHandlers) -> XMPP ()
modifyHandlers f = do modifyHandlers f = do
eh <- asks eventHandlers eh <- asks eventHandlers
liftIO . atomically $ writeTVar eh . f =<< readTVar eh liftIO . atomically $ writeTVar eh . f =<< readTVar eh
setSessionEndHandler :: XMPPThread () -> XMPPThread () setSessionEndHandler :: XMPP () -> XMPP ()
setSessionEndHandler eh = modifyHandlers (\s -> s{sessionEndHandler = eh}) setSessionEndHandler eh = modifyHandlers (\s -> s{sessionEndHandler = eh})
-- | run an event handler -- | run an event handler
runHandler :: (EventHandlers -> XMPPThread a) -> XMPPThread a runHandler :: (EventHandlers -> XMPP a) -> XMPP a
runHandler h = do runHandler h = do
eh <- liftIO . atomically . readTVar =<< asks eventHandlers eh <- liftIO . atomically . readTVar =<< asks eventHandlers
h eh h eh
-- | End the current xmpp session -- | End the current xmpp session
endSession :: XMPPThread () endSession :: XMPP ()
endSession = do -- TODO: This has to be idempotent (is it?) endSession = do -- TODO: This has to be idempotent (is it?)
withConnection xmppKillConnection withConnection xmppKillConnection
liftIO =<< asks stopThreads liftIO =<< asks stopThreads
runHandler sessionEndHandler runHandler sessionEndHandler
-- | Close the connection to the server -- | Close the connection to the server
closeConnection :: XMPPThread () closeConnection :: XMPP ()
closeConnection = withConnection xmppKillConnection closeConnection = withConnection xmppKillConnection

64
src/Network/XMPP/Concurrent/Threads.hs
Unescape View File

@ -119,28 +119,28 @@ writeWorker stCh writeR = forever $ do
-- returns channel of incoming and outgoing stances, respectively -- returns channel of incoming and outgoing stances, respectively
-- and an Action to stop the Threads and close the connection -- and an Action to stop the Threads and close the connection
startThreads startThreads
:: XMPPConMonad ( TChan (Either MessageError Message) :: IO ( TChan (Either MessageError Message)
, TChan (Either PresenceError Presence) , TChan (Either PresenceError Presence)
, TVar IQHandlers , TVar IQHandlers
, TChan Stanza , TChan Stanza
, IO () , IO ()
, TMVar (BS.ByteString -> IO ()) , TMVar (BS.ByteString -> IO ())
, TMVar XMPPConState , TMVar XMPPConState
, ThreadId , ThreadId
, TVar EventHandlers , TVar EventHandlers
) )
startThreads = do startThreads = do
writeLock <- liftIO . newTMVarIO =<< gets sConPushBS writeLock <- newEmptyTMVarIO
messageC <- liftIO newTChanIO messageC <- newTChanIO
presenceC <- liftIO newTChanIO presenceC <- newTChanIO
outC <- liftIO newTChanIO outC <- newTChanIO
handlers <- liftIO $ newTVarIO ( Map.empty, Map.empty) handlers <- newTVarIO ( Map.empty, Map.empty)
eh <- liftIO $ newTVarIO zeroEventHandlers eh <- newTVarIO zeroEventHandlers
conS <- liftIO . newTMVarIO =<< get conS <- newEmptyTMVarIO
lw <- liftIO . forkIO $ writeWorker outC writeLock lw <- forkIO $ writeWorker outC writeLock
cp <- liftIO . forkIO $ connPersist writeLock cp <- forkIO $ connPersist writeLock
rd <- liftIO . forkIO $ readWorker messageC presenceC handlers conS rd <- forkIO $ readWorker messageC presenceC handlers conS
return (messageC, presenceC, handlers, outC return (messageC, presenceC, handlers, outC
, killConnection writeLock [lw, rd, cp] , killConnection writeLock [lw, rd, cp]
, writeLock, conS ,rd, eh) , writeLock, conS ,rd, eh)
@ -150,24 +150,24 @@ startThreads = do
_ <- forM threads killThread _ <- forM threads killThread
return() return()
-- | Start worker threads and run action. The supplied action will run -- | Creates and initializes a new XMPP session.
-- in the calling thread. use 'forkXMPP' to start another thread. newSession :: IO Session
runThreaded :: XMPPThread a newSession = do
-> XMPPConMonad a
runThreaded a = do
liftIO . putStrLn $ "starting threads"
(mC, pC, hand, outC, stopThreads', writeR, conS, rdr, eh) <- startThreads (mC, pC, hand, outC, stopThreads', writeR, conS, rdr, eh) <- startThreads
liftIO . putStrLn $ "threads running" workermCh <- newIORef $ Nothing
workermCh <- liftIO . newIORef $ Nothing workerpCh <- newIORef $ Nothing
workerpCh <- liftIO . newIORef $ Nothing idRef <- newTVarIO 1
idRef <- liftIO $ newTVarIO 1
let getId = atomically $ do let getId = atomically $ do
curId <- readTVar idRef curId <- readTVar idRef
writeTVar idRef (curId + 1 :: Integer) writeTVar idRef (curId + 1 :: Integer)
return . read. show $ curId return . read. show $ curId
liftIO . putStrLn $ "starting application" return (Session workermCh workerpCh mC pC outC hand writeR rdr getId conS eh stopThreads')
liftIO $ runReaderT a (Thread workermCh workerpCh mC pC outC hand writeR rdr getId conS eh stopThreads')
withNewSession :: XMPP b -> IO b
withNewSession a = newSession >>= runReaderT a
withSession :: Session -> XMPP a -> IO a
withSession = flip runReaderT
-- | Sends a blank space every 30 seconds to keep the connection alive -- | Sends a blank space every 30 seconds to keep the connection alive
connPersist :: TMVar (BS.ByteString -> IO ()) -> IO () connPersist :: TMVar (BS.ByteString -> IO ()) -> IO ()

46
src/Network/XMPP/Concurrent/Types.hs
Unescape View File

@ -23,8 +23,8 @@ type IQHandlers = (Map.Map (IQRequestType, Text) (TChan (IQRequest, TVar Bool))
) )
data EventHandlers = EventHandlers data EventHandlers = EventHandlers
{ sessionEndHandler :: XMPPThread () { sessionEndHandler :: XMPP ()
, connectionClosedHandler :: XMPPThread () , connectionClosedHandler :: XMPP ()
} }
zeroEventHandlers :: EventHandlers zeroEventHandlers :: EventHandlers
@ -33,29 +33,29 @@ zeroEventHandlers = EventHandlers
, connectionClosedHandler = return () , connectionClosedHandler = return ()
} }
data Thread = Thread { messagesRef :: IORef (Maybe ( TChan (Either data Session = Session { messagesRef :: IORef (Maybe ( TChan (Either
MessageError MessageError
Message Message
))) )))
, presenceRef :: IORef (Maybe (TChan (Either , presenceRef :: IORef (Maybe (TChan (Either
PresenceError PresenceError Presence )))
Presence , mShadow :: TChan (Either MessageError
))) Message)
, mShadow :: TChan (Either MessageError -- the original chan
Message) -- the original chan , pShadow :: TChan (Either PresenceError
, pShadow :: TChan (Either PresenceError Presence)
Presence) -- the original chan -- the original chan
, outCh :: TChan Stanza , outCh :: TChan Stanza
, iqHandlers :: TVar IQHandlers , iqHandlers :: TVar IQHandlers
, writeRef :: TMVar (BS.ByteString -> IO () ) , writeRef :: TMVar (BS.ByteString -> IO () )
, readerThread :: ThreadId , readerThread :: ThreadId
, idGenerator :: IO StanzaId , idGenerator :: IO StanzaId
, conStateRef :: TMVar XMPPConState , conStateRef :: TMVar XMPPConState
, eventHandlers :: TVar EventHandlers , eventHandlers :: TVar EventHandlers
, stopThreads :: IO () , stopThreads :: IO ()
} }
type XMPPThread a = ReaderT Thread IO a type XMPP a = ReaderT Session IO a
data Interrupt = Interrupt (TMVar ()) deriving Typeable data Interrupt = Interrupt (TMVar ()) deriving Typeable
instance Show Interrupt where show _ = "<Interrupt>" instance Show Interrupt where show _ = "<Interrupt>"

4
src/Network/XMPP/Monad.hs
Unescape View File

@ -120,8 +120,8 @@ xmppRawConnect host hostname = do
put st put st
withNewSession :: XMPPConMonad a -> IO (a, XMPPConState) xmppNewSession :: XMPPConMonad a -> IO (a, XMPPConState)
withNewSession action = do xmppNewSession action = do
runStateT action xmppZeroConState runStateT action xmppZeroConState
xmppKillConnection :: XMPPConMonad () xmppKillConnection :: XMPPConMonad ()

2
src/Network/XMPP/Session.hs
Unescape View File

@ -32,7 +32,7 @@ xmppSession = do
let IQResultS (IQResult "sess" Nothing Nothing _lang _body) = answer let IQResultS (IQResult "sess" Nothing Nothing _lang _body) = answer
return () return ()
startSession :: XMPPThread () startSession :: XMPP ()
startSession = do startSession = do
answer <- sendIQ' Nothing Set Nothing sessionXML answer <- sendIQ' Nothing Set Nothing sessionXML
case answer of case answer of

Write Preview
Loading…
Cancel
Save