pontarius-xmpp/src/Network/XMPP/Concurrent/Monad.hs

module Network.XMPP.Concurrent.Monad where

import           Network.XMPP.Types

import           Control.Concurrent
import           Control.Concurrent.STM
import qualified Control.Exception.Lifted as Ex
import           Control.Monad.IO.Class
import           Control.Monad.Trans.Reader
import           Control.Monad.Trans.State

import           Data.IORef
import qualified Data.Map as Map
import           Data.Text(Text)

import           Network.XMPP.Concurrent.Types

-- | Register a new IQ listener. IQ requests matching the type and namespace will
-- be put in the channel.
listenIQChan :: IQRequestType  -- ^ type of IQs to receive (Get / Set)
                -> Text -- ^ namespace of the child element
                -> XMPPThread (Bool, TChan (IQRequest, TVar Bool))
listenIQChan tp ns = do
  handlers <- asks iqHandlers
  liftIO . atomically $ do
    (byNS, byID) <- readTVar handlers
    iqCh <- newTChan
    let (present, byNS') = Map.insertLookupWithKey' (\_ new _ -> new)
                                                     (tp,ns) iqCh byNS
    writeTVar handlers (byNS', byID)
    return $ case present of
               Nothing -> (True, iqCh)
               Just iqCh' -> (False, iqCh')

-- | get the inbound stanza channel, duplicates from master if necessary
-- please note that once duplicated it will keep filling up, call
-- 'dropMessageChan' to allow it to be garbage collected
getMessageChan :: XMPPThread (TChan (Either MessageError Message))
getMessageChan = do
  mChR <- asks messagesRef
  mCh <- liftIO $ readIORef mChR
  case mCh of
    Nothing -> do
      shadow <- asks mShadow
      mCh' <- liftIO $ atomically $ dupTChan shadow
      liftIO $ writeIORef mChR (Just mCh')
      return mCh'
    Just mCh' -> return mCh'

-- | see 'getMessageChan'
getPresenceChan :: XMPPThread (TChan (Either PresenceError Presence))
getPresenceChan = do
  pChR <- asks presenceRef
  pCh <- liftIO $ readIORef pChR
  case pCh of
    Nothing -> do
      shadow <- asks pShadow
      pCh' <- liftIO $ atomically $ dupTChan shadow
      liftIO $ writeIORef pChR (Just pCh')
      return pCh'
    Just pCh' -> return pCh'

-- | Drop the local end of the inbound stanza channel
-- from our context so it can be GC-ed
dropMessageChan :: XMPPThread ()
dropMessageChan = do
  r <- asks messagesRef
  liftIO $ writeIORef r Nothing

-- | see 'dropMessageChan'
dropPresenceChan :: XMPPThread ()
dropPresenceChan = do
  r <- asks presenceRef
  liftIO $ writeIORef r Nothing

-- | Read an element from the inbound stanza channel, acquiring a copy
-- of the channel as necessary
pullMessage :: XMPPThread (Either MessageError Message)
pullMessage = do
  c <- getMessageChan
  liftIO $ atomically $ readTChan c

-- | Read an element from the inbound stanza channel, acquiring a copy
-- of the channel as necessary
pullPresence :: XMPPThread (Either PresenceError Presence)
pullPresence = do
  c <- getPresenceChan
  liftIO $ atomically $ readTChan c

-- | Send a stanza to the server
sendS :: Stanza -> XMPPThread ()
sendS a = do
  out <- asks outCh
  liftIO . atomically $ writeTChan out a
  return ()

-- | Fork a new thread
forkXMPP :: XMPPThread () -> XMPPThread ThreadId
forkXMPP a = do
  thread <- ask
  mCH' <- liftIO $ newIORef Nothing
  pCH' <- liftIO $ newIORef Nothing
  liftIO $ forkIO $ runReaderT a (thread {messagesRef = mCH'
                                         ,presenceRef = pCH'
                                         })

filterMessages :: (MessageError -> Bool)
               -> (Message -> Bool)
               -> XMPPThread (Either MessageError Message)
filterMessages f g = do
  s <- pullMessage
  case s of
    Left  e | f e -> return $ Left e
            | otherwise -> filterMessages f g
    Right m | g m -> return $ Right m
            | otherwise -> filterMessages f g

waitForMessage :: (Message -> Bool) -> XMPPThread Message
waitForMessage f = do
  s <- pullMessage
  case s of
    Left _ -> waitForMessage f
    Right m | f m -> return m
            | otherwise -> waitForMessage f

waitForMessageError :: (MessageError -> Bool) -> XMPPThread MessageError
waitForMessageError f = do
  s <- pullMessage
  case s of
    Right _ -> waitForMessageError f
    Left  m | f m -> return m
            | otherwise -> waitForMessageError f

waitForPresence :: (Presence -> Bool) -> XMPPThread Presence
waitForPresence f = do
  s <- pullPresence
  case s of
    Left _ -> waitForPresence f
    Right m | f m -> return m
            | otherwise -> waitForPresence f

-- | Run an XMPPMonad action in isolation.
-- Reader and writer workers will be temporarily stopped
-- and resumed with the new session details once the action returns.
-- The Action will run in the calling thread/
-- NB: This will /not/ catch any exceptions. If you action dies, deadlocks
-- or otherwisely exits abnormaly the XMPP session will be dead.
withConnection :: XMPPConMonad a -> XMPPThread a
withConnection a = do
  readerId <- asks readerThread
  stateRef <- asks conStateRef
  write <- asks writeRef
  wait <- liftIO $ newEmptyTMVarIO
  liftIO . throwTo readerId $ Interrupt wait
  s <- liftIO . atomically $ do
    putTMVar wait ()
    takeTMVar write
    takeTMVar stateRef
  (res, s') <- liftIO $ runStateT a s
  liftIO . atomically $ do
    putTMVar write (sConPushBS s')
    putTMVar stateRef s'
  return res

sendPresence :: Presence -> XMPPThread ()
sendPresence = sendS . PresenceS

sendMessage :: Message -> XMPPThread ()
sendMessage = sendS . MessageS