{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE CPP #-}
{-# LANGUAGE ViewPatterns #-}
module XMPP
( module XMPPTypes
, module SocketLike
, listenForXmppClients
, listenForRemotePeers
, newServerConnections
, seekRemotePeers
, quitListening
, OutBoundMessage(..)
, sendMessage
) where
import ServerC
import XMPPTypes
import SocketLike
import ByteStringOperators
import ControlMaybe
import Data.Maybe (catMaybes)
import Data.HList
import Network.Socket
( Family
, connect
, socketToHandle
, sClose
, Socket(..)
, socket
, SocketType(..)
)
import Network.BSD
( PortNumber
, getHostName
, hostName
, hostAliases
, getProtocolNumber
)
import System.IO
( BufferMode(..)
, IOMode(..)
, hSetBuffering
)
import Control.Concurrent.STM
import Data.Conduit
import qualified Data.Conduit.List as CL
import qualified Data.Conduit.Binary as CB
import Data.ByteString (ByteString)
import qualified Data.ByteString.Char8 as S (pack,putStr,putStrLn,append)
import qualified Data.ByteString.Lazy.Char8 as L
( putStrLn
, fromChunks
, unlines
, hPutStrLn
)
import Control.Concurrent (forkIO,killThread)
import Control.Concurrent.Async
import Control.Exception
( handle
-- , SomeException(..)
, finally
, bracketOnError )
import GHC.IO.Exception (IOException(..))
import System.IO.Error (isDoesNotExistError)
import Control.Monad.IO.Class
import Control.Monad.Trans.Class
import Control.Monad.Trans.Maybe
import Todo
import Control.Monad as Monad
import Text.XML.Stream.Parse (parseBytes,content)
import Text.XML.Stream.Render
import Data.XML.Types as XML
import qualified Data.Text as S (takeWhile)
import Data.Text.Encoding as S (decodeUtf8,encodeUtf8)
import Data.Text.Lazy.Encoding as L (decodeUtf8,encodeUtf8)
import Data.Text.Lazy (toStrict)
import Data.Monoid
import qualified Data.Sequence as Seq
import Data.Foldable (toList)
#ifdef RENDERFLUSH
import Data.Conduit.Blaze
#endif
import Data.List (find)
import qualified Text.Show.ByteString as L
import NestingXML
import Data.Set as Set (Set,(\\))
import qualified Data.Set as Set
import qualified Data.Map as Map
import Data.Map as Map (Map)
import GHC.Conc
( threadStatus
, ThreadStatus(..)
, ThreadId
)
textToByteString x = L.fromChunks [S.encodeUtf8 x]
data Commands =
Send [XML.Event]
| BoundToResource
| InterestedInRoster
| QuitThread
deriving Prelude.Show
xmlifyPresenceForClient :: Presence -> IO [XML.Event]
xmlifyPresenceForClient (Presence jid stat) = do
let n = name jid
rsc = resource jid
names <- getNamesForPeer (peer jid)
let tostr p = L.decodeUtf8 $ n <$++> "@" ++> L.fromChunks [p] <++?> "/" <++$> rsc
jidstrs = fmap (toStrict . tostr) names
return (concatMap presenceEvents jidstrs)
where
presenceEvents jidstr =
[ EventBeginElement "{jabber:client}presence" (("from",[ContentText jidstr]):typ stat)
, EventBeginElement "{jabber:client}show" []
, EventContent (ContentText . shw $ stat)
, EventEndElement "{jabber:client}show"
, EventEndElement "{jabber:client}presence"
]
typ Offline = [("type",[ContentText "unavailable"])]
typ _ = []
shw Available = "chat"
shw Away = "away"
shw Offline = "away" -- Is this right?
prefix ## name = Name name Nothing (Just prefix)
streamP name = Name name (Just "http://etherx.jabber.org/streams") (Just "stream")
greet host =
[ EventBeginDocument
, EventBeginElement (streamP "stream")
[("from",[ContentText host])
,("id",[ContentText "someid"])
,("xmlns",[ContentText "jabber:client"])
,("xmlns:stream",[ContentText "http://etherx.jabber.org/streams"])
,("version",[ContentText "1.0"])
]
, EventBeginElement (streamP "features") []
, EventBeginElement "{urn:ietf:params:xml:ns:xmpp-bind}bind" []
, EventEndElement "{urn:ietf:params:xml:ns:xmpp-bind}bind"
{-
-- , " "
, " "
-- , " DIGEST-MD5"
, " PLAIN"
, " "
-}
, EventEndElement (streamP "features")
]
-- type Consumer i m r = forall o. ConduitM i o m r
mawait :: Monad m => MaybeT (ConduitM i o m) i
mawait = MaybeT await
-- Note: This function ignores name space qualification
elementAttrs expected (EventBeginElement name attrs)
| nameLocalName name==expected
= return attrs
elementAttrs _ _ = mzero
eventIsBeginElement (EventBeginElement _ _) = True
eventIsBeginElement _ = False
eventIsEndElement (EventEndElement _) = True
eventIsEndElement _ = False
filterMapElement::
(Monad m, MonadPlus mp) =>
(Event -> mp a) -> Event -> mp a -> MaybeT (ConduitM Event o m) (mp a)
filterMapElement ret opentag empty = loop (empty `mplus` ret opentag) 1
where
loop ts 0 = return ts
loop ts cnt = do
tag <- mawait
let ts' = mplus ts (ret tag)
case () of
_ | eventIsEndElement tag -> loop ts' (cnt-1)
_ | eventIsBeginElement tag -> loop ts' (cnt+1)
_ -> loop ts' cnt
gatherElement ::
(Monad m, MonadPlus mp) =>
Event -> mp Event -> NestingXML o m (mp Event)
gatherElement opentag empty = loop (empty `mplus` return opentag) 1
where
loop ts 0 = return ts
loop ts cnt = do
maybeXML (return ts) $ \tag -> do
let ts' = mplus ts (return tag)
case () of
_ | eventIsEndElement tag -> loop ts' (cnt-1)
_ | eventIsBeginElement tag -> loop ts' (cnt+1)
_ -> loop ts' cnt
{-
sourceStanza :: Monad m => Event -> ConduitM Event Event m ()
sourceStanza opentag = yield opentag >> loop 1
where
loop 0 = return ()
loop cnt = do
e <- await
let go tag cnt = yield tag >> loop cnt
case e of
Just tag | eventIsEndElement tag -> go tag (cnt-1)
Just tag | eventIsBeginElement tag -> go tag (cnt+1)
Just tag -> go tag cnt
Nothing -> return ()
-}
voidMaybeT body = (>> return ()) . runMaybeT $ body
fixMaybeT f = (>> return ()) . runMaybeT . fix $ f
iq_bind_reply id jid =
[ EventBeginElement "{jabber:client}iq" [("type",[ContentText "result"]),("id",[ContentText id])]
, EventBeginElement "{urn:ietf:params:xml:ns:xmpp-bind}bind"
[("xmlns",[ContentText "urn:ietf:params:xml:ns:xmpp-bind"])]
, EventBeginElement "{urn:ietf:params:xml:ns:xmpp-bind}jid" []
, EventContent (ContentText jid)
, EventEndElement "{urn:ietf:params:xml:ns:xmpp-bind}jid"
, EventEndElement "{urn:ietf:params:xml:ns:xmpp-bind}bind"
, EventEndElement "{jabber:client}iq"
]
uncontent cs = head $ map getText cs
where
getText (ContentText x) = x
getText (ContentEntity x ) = x
tagAttrs (EventBeginElement _ xs) = xs
tagAttrs _ = []
tagName (EventBeginElement n _) = n
tagName _ = ""
handleIQSetBind session cmdChan stanza_id = do
mchild <- nextElement
rsc <- case mchild of
Just child -> do
let unhandledBind = do
liftIO $ putStrLn $ "unhandled-bind: "++show child
return ""
case tagName child of
"{urn:ietf:params:xml:ns:xmpp-bind}resource"
-> do
rsc <- lift content
return . textToByteString $ rsc
_ -> unhandledBind
Nothing -> do
liftIO $ putStrLn $ "empty bind request!"
return ""
liftIO $ do
L.putStrLn $ "iq-set-bind-resource " <++> rsc
setResource session rsc
jid <- getJID session
atomically $ do
writeTChan cmdChan (Send $ iq_bind_reply stanza_id (toStrict $ L.decodeUtf8 $ L.show jid) )
writeTChan cmdChan BoundToResource
forCachedPresence session $ \presence -> do
xs <- xmlifyPresenceForClient presence
atomically . writeTChan cmdChan . Send $ xs
iq_session_reply host stanza_id =
[ EventBeginElement "{jabber:client}iq"
[("id",[ContentText stanza_id])
,("from",[ContentText host])
,("type",[ContentText "result"])
]
, EventEndElement "{jabber:client}iq"
]
handleIQSetSession session cmdChan stanza_id = do
host <- liftIO $ do
jid <- getJID session
names <- getNamesForPeer (peer jid)
return (S.decodeUtf8 . head $ names)
liftIO . atomically . writeTChan cmdChan . Send $ iq_session_reply host stanza_id
handleIQSet session cmdChan tag = do
withJust (lookupAttrib "id" (tagAttrs tag)) $ \stanza_id -> do
whenJust nextElement $ \child -> do
let unhandledSet = liftIO $ putStrLn ("iq-set: "++show (stanza_id,child))
case tagName child of
"{urn:ietf:params:xml:ns:xmpp-bind}bind"
-> handleIQSetBind session cmdChan stanza_id
"{urn:ietf:params:xml:ns:xmpp-session}session"
-> handleIQSetSession session cmdChan stanza_id
_ -> unhandledSet
matchAttrib name value attrs =
case find ( (==name) . fst) attrs of
Just (_,[ContentText x]) | x==value -> True
Just (_,[ContentEntity x]) | x==value -> True
_ -> False
lookupAttrib name attrs =
case find ( (==name) . fst) attrs of
Just (_,[ContentText x]) -> Just x
Just (_,[ContentEntity x]) -> Just x
_ -> Nothing
iqTypeSet = "set"
iqTypeGet = "get"
iqTypeResult = "result"
iqTypeError = "error"
isIQOf (EventBeginElement name attrs) testType
| name=="{jabber:client}iq"
&& matchAttrib "type" testType attrs
= True
isIQOf _ _ = False
iq_service_unavailable host iq_id mjid req =
[ EventBeginElement "{jabber:client}iq"
[("type",[ContentText "error"])
,("id",[ContentText iq_id])
-- , TODO: set "from" if isJust mjid
]
, EventBeginElement req []
, EventEndElement req
, EventBeginElement "{jabber:client}error" [("type",[ContentText "cancel"])]
, EventBeginElement "{urn:ietf:params:xml:ns:xmpp-stanzas}service-unavailable" []
, EventEndElement "{urn:ietf:params:xml:ns:xmpp-stanzas}service-unavailable"
, EventEndElement "{jabber:client}error"
, EventEndElement "{jabber:client}iq"
]
attr name value = (name,[ContentText value])
attrbs name value = (name,[ContentText (toStrict . L.decodeUtf8 $ value)])
getRoster session iqid = do
let getlist f = do
bs <- f session
-- js <- mapM parseHostNameJID bs
return (Set.fromList bs) -- js)
buddies <- getlist getMyBuddies
subscribers <- getlist getMySubscribers
subnone <- getlist getMyOthers
solicited <- getlist getMySolicited
let subto = buddies \\ subscribers
let subfrom = subscribers \\ buddies
let subboth = Set.intersection buddies subscribers
-- solicited -> ask='subscribe'
jid <- getJID session
let dest = toStrict . L.decodeUtf8 . bshow $ jid
let items= (xmlify solicited "to" subto)
++(xmlify solicited "from" subfrom)
++(xmlify solicited "both" subboth)
++(xmlify solicited "none" subnone)
openiq = [EventBeginElement "{jabber:client}iq"
[ attr "id" iqid
, attr "to" dest
, attr "type" "result" ]
,EventBeginElement "{jabber:iq:roster}query"
[] -- todo: ver?
]
closeiq = [EventEndElement "{jabber:iq:roster}query"
,EventEndElement "{jabber:client}iq"]
return $ openiq ++ items ++ closeiq
where
xmlify solicited stype set = flip concatMap (Set.toList set)
$ \jid ->
[ EventBeginElement "item"
([ attr "jid" (toStrict . L.decodeUtf8 $ jid)
, attr "subscription" stype
]++if Set.member jid solicited
then [attr "ask" "subscribe"]
else [] )
, EventEndElement "item"
]
handleIQGet session cmdChan tag = do
withJust (lookupAttrib "id" (tagAttrs tag)) $ \stanza_id -> do
whenJust nextElement $ \child -> do
host <- liftIO $ do
jid <- getJID session
names <- getNamesForPeer (peer jid)
return (S.decodeUtf8 . head $ names)
let unhandledGet req = do
liftIO $ putStrLn ("iq-get: "++show (stanza_id,child))
liftIO . atomically . writeTChan cmdChan . Send $ iq_service_unavailable host stanza_id Nothing req
case tagName child of
-- "{http://jabber.org/protocol/disco#items}query" -> liftIO $ putStrLn "iq-get-query-items"
"{urn:xmpp:ping}ping" -> liftIO $ do
let mjid = lookupAttrib "from" (tagAttrs tag)
let pong = [ EventBeginElement "{jabber:client}iq"
$ (case mjid of
Just jid -> (attr "to" jid :)
_ -> id )
[ attr "type" "result"
, attr "id" stanza_id
, attr "from" host
]
, EventEndElement "{jabber:client}iq"
]
atomically . writeTChan cmdChan . Send $ pong
"{jabber:iq:roster}query" -> liftIO $ do
putStrLn $ "REQUESTED ROSTER " ++ show tag
-- REQUESTED ROSTER EventBeginElement
-- (Name {nameLocalName = "iq", nameNamespace = Just "jabber:client", namePrefix = Nothing})
-- [(Name { nameLocalName = "id"
-- , nameNamespace = Nothing
-- , namePrefix = Nothing},
-- [ContentText "32a337c2-7b22-45b6-9d21-15ded0d079ec"])
-- ,(Name {nameLocalName = "type", nameNamespace = Nothing, namePrefix = Nothing},
-- [ContentText "get"])]
roster <- getRoster session stanza_id
atomically $ do
writeTChan cmdChan InterestedInRoster
writeTChan cmdChan . Send $ roster
sendPending session
req -> unhandledGet req
fromClient :: (MonadThrow m,MonadIO m, JabberClientSession session) =>
session -> TChan Commands -> Sink XML.Event m ()
fromClient session cmdChan = doNestingXML $ do
let log = liftIO . L.putStrLn . ("(C) " <++>)
send = liftIO . atomically . writeTChan cmdChan . Send
withXML $ \begindoc -> do
when (begindoc==EventBeginDocument) $ do
log "begin-doc"
withXML $ \xml -> do
withJust (elementAttrs "stream" xml) $ \stream_attrs -> do
log $ "stream atributes: " <++> bshow stream_attrs
host <- liftIO $ do
jid <- getJID session
names <- getNamesForPeer (peer jid)
return (S.decodeUtf8 . head $ names)
send $ greet host
fix $ \loop -> do
log "waiting for stanza."
whenJust nextElement $ \stanza -> do
stanza_lvl <- nesting
liftIO . putStrLn $ "stanza: "++show stanza
let unhandledStanza = do
xs <- gatherElement stanza Seq.empty
prettyPrint "unhandled-C: " (toList xs)
case () of
_ | stanza `isIQOf` iqTypeSet -> handleIQSet session cmdChan stanza
_ | stanza `isIQOf` iqTypeGet -> handleIQGet session cmdChan stanza
_ | stanza `isClientPresenceOf` presenceTypeSubscribe
-> clientRequestsSubscription session cmdChan stanza
_ | stanza `isClientPresenceOf` presenceTypeSubscribed
-> clientApprovesSubscription session stanza
_ | otherwise -> unhandledStanza
awaitCloser stanza_lvl
loop
log $ "end of stream"
withXML $ \xml -> do
log $ "end-of-document: " <++> bshow xml
prettyPrint prefix xs =
liftIO $ do
CL.sourceList xs $= renderBytes (def { rsPretty=True }) =$= CB.lines $$ CL.mapM_ (S.putStrLn . (prefix `S.append`))
rosterPush to contact attrs = do
let n = name to
rsc = resource to
names <- getNamesForPeer (peer to)
let tostr p = L.decodeUtf8 $ n <$++> "@" ++> L.fromChunks [p] <++?> "/" <++$> rsc
jidstrs = fmap (toStrict . tostr) names
tojid = head jidstrs
return
[ EventBeginElement "{jabber:client}iq"
[ attr "to" tojid
, attr "id" "someid"
, attr "type" "set"
]
, EventBeginElement "{jabber:iq:roster}query" []
, EventBeginElement "{jabber:iq:roster}item" (attr "jid" contact : attrs)
, EventEndElement "{jabber:iq:roster}item"
, EventEndElement "{jabber:iq:roster}query"
, EventEndElement "{jabber:client}iq"
]
data EventsForClient = CmdChan Commands
| PChan Presence
| RChan RosterEvent
toClient :: (MonadIO m, JabberClientSession session ) =>
session -> TChan Presence -> TChan Commands -> TChan RosterEvent -> Source m [XML.Event]
toClient session pchan cmdChan rchan = toClient' False False
where
toClient' isBound isInterested = do
let loop = toClient' isBound isInterested
send xs = yield xs >> prettyPrint ">C: " xs
event <- liftIO . atomically $
foldr1 orElse [fmap CmdChan $ readTChan cmdChan
,fmap RChan $ readTChan rchan
,fmap PChan $ readTChan pchan
]
case event of
CmdChan QuitThread -> return ()
CmdChan (Send xs) -> send xs >> loop
CmdChan BoundToResource -> toClient' True isInterested
CmdChan InterestedInRoster -> do
liftIO . putStrLn $ "Roster: interested"
toClient' isBound True
-- CmdChan cmd -> liftIO (putStrLn $ "unhandled event: "++show cmd) >> loop
RChan (RequestedSubscription who contact) -> do
jid <- liftIO $ getJID session
when (isInterested && Just who==name jid) $ do
r <- liftIO $ rosterPush jid (toStrict . L.decodeUtf8 $ contact) [attr "ask" "subscribe"]
send r
loop
RChan (NewBuddy who contact) -> do
liftIO . putStrLn $ "Roster push: NewBuddy "++show (isInterested,who,contact)
(jid,me) <- liftIO $ do
jid <- getJID session
me <- asHostNameJID jid
return (jid,me)
withJust me $ \me -> do
when (isInterested && Just who==name jid) $ do
send [ EventBeginElement "{jabber:client}presence"
[ attrbs "from" contact
, attrbs "to" me
, attr "type" "subscribed"
]
, EventEndElement "{jabber:client}presence" ]
let f True = "from"
f False = "to"
subscription <- fmap f (liftIO $ isSubscribed session contact)
r <- liftIO . handleIO (\e -> putStrLn ("Roster NewBuddy error: "++show e) >> return []) $ do
rosterPush jid
(toStrict . L.decodeUtf8 $ contact)
[attr "subscription" subscription]
send r
loop
RChan (PendingBuddy who contact) -> do
liftIO . putStrLn $ "Roster: Pending buddy "++show (isInterested,who,contact)
(jid,me) <- liftIO $ do
jid <- getJID session
me <- asHostNameJID jid
return (jid,me)
withJust me $ \me -> do
when (isInterested && Just who==name jid) $ do
send [ EventBeginElement "{jabber:client}presence"
[ attrbs "from" contact
, attrbs "to" me
, attr "type" "subscribe"
]
, EventEndElement "{jabber:client}presence" ]
loop
PChan presence -> do
when isBound $ do
xs <- liftIO $ xmlifyPresenceForClient presence
send xs
loop
handleClient
:: (SocketLike sock, HHead l (XMPPClass session),
JabberClientSession session) =>
HCons sock (HCons t l) -> Source IO ByteString -> Sink ByteString IO () -> IO ()
handleClient st src snk = do
let HCons sock (HCons _ st') = st
session_factory = hHead st'
pname <- getPeerName sock
session <- newSession session_factory sock
Prelude.putStrLn $ "PEER NAME: "++Prelude.show pname
pchan <- subscribe session Nothing
rchan <- subscribeToRoster session
cmdChan <- atomically newTChan
#ifdef RENDERFLUSH
writer <- async ( toClient session pchan cmdChan rchan
$$ flushList
=$= renderBuilderFlush def
=$= builderToByteStringFlush
=$= discardFlush
=$ snk )
#else
writer <- async ( toClient session pchan cmdChan rchan $$ renderChunks =$ snk )
#endif
finally ( src $= parseBytes def $$ fromClient session cmdChan )
$ do
atomically $ writeTChan cmdChan QuitThread
wait writer
closeSession session
listenForXmppClients ::
(HList l, HHead l (XMPPClass session), HExtend e1 l2 l1,
HExtend e l1 (HCons PortNumber l), JabberClientSession session) =>
Family -> e1 -> e -> l2 -> IO ServerHandle
listenForXmppClients addr_family session_factory port st = do
doServer (addr_family .*. port .*. session_factory .*. st) handleClient
#ifdef RENDERFLUSH
flushList :: Monad m => ConduitM [a] (Flush a) m ()
flushList = fixMaybeT $ \loop -> do
xs <- mawait
lift ( CL.sourceList xs $$ CL.mapM_ (yield . Chunk) )
lift ( yield Flush )
loop
discardFlush :: Monad m => ConduitM (Flush a) a m ()
discardFlush = fixMaybeT $ \loop -> do
x <- mawait
let unchunk (Chunk a) = a
ischunk (Chunk _) = True
ischunk _ = False
lift . when (ischunk x) $ yield (unchunk x)
loop
#else
renderChunks :: (MonadUnsafeIO m, MonadIO m) => ConduitM [Event] ByteString m ()
renderChunks = fixMaybeT $ \loop -> do
xs <- mawait
lift . when (not . null $ xs) $ ( CL.sourceList xs $= renderBytes def $$ CL.mapM_ yield )
loop
#endif
listenForRemotePeers
:: (HList l, HHead l (XMPPPeerClass session),
HExtend e l1 (HCons PortNumber l), HExtend e1 l2 l1,
JabberPeerSession session) =>
Family -> e1 -> e -> l2 -> IO ServerHandle
listenForRemotePeers addrfamily session_factory port st = do
doServer (addrfamily .*. port .*. session_factory .*. st) handlePeer
handlePeer
:: (SocketLike sock, HHead l (XMPPPeerClass session),
JabberPeerSession session) =>
HCons sock (HCons t1 l) -> Source IO ByteString -> t -> IO ()
handlePeer st src snk = do
let HCons sock (HCons _ st') = st
session_factory = hHead st'
name <- fmap bshow $ getPeerName sock
L.putStrLn $ "(P) connected " <++> name
session <- newPeerSession session_factory sock
finally ( src $= parseBytes def $$ fromPeer session )
$ do
L.putStrLn $ "(P) disconnected " <++> name
closePeerSession session
handlePeerPresence session stanza False = do
-- Offline
withJust (lookupAttrib "from" (tagAttrs stanza)) $ \jid -> do
peer_jid <- liftIO $ parseAddressJID (textToByteString jid)
liftIO $ announcePresence session (Presence peer_jid Offline)
handlePeerPresence session stanza True = do
-- online (Available or Away)
let log = liftIO . L.putStrLn . ("(P) " <++>)
withJust (lookupAttrib "from" (tagAttrs stanza)) $ \jid -> do
pjid <- liftIO $ parseAddressJID (textToByteString jid)
-- stat <- show element content
let parseChildren stat = do
child <- nextElement
case child of
Just tag | tagName tag=="{jabber:server}show"
-> fmap toStat (lift content)
Just tag | otherwise -> parseChildren stat
Nothing -> return stat
toStat "away" = Away
toStat "xa" = Away -- TODO: xa
toStat "dnd" = Away -- TODO: dnd
toStat "chat" = Available
stat' <- parseChildren Available
liftIO $ announcePresence session (Presence pjid stat')
log $ bshow (Presence pjid stat')
matchAttribMaybe name (Just value) attrs =
case find ( (==name) . fst) attrs of
Just (_,[ContentText x]) | x==value -> True
Just (_,[ContentEntity x]) | x==value -> True
_ -> False
matchAttribMaybe name Nothing attrs
| find ( (==name) . fst) attrs==Nothing
= True
matchAttribMaybe name Nothing attrs
| otherwise
= False
presenceTypeOffline = Just "unavailable"
presenceTypeOnline = Nothing
presenceTypeProbe = Just "probe"
presenceTypeSubscribe = Just "subscribe"
presenceTypeSubscribed = Just "subscribed"
isPresenceOf (EventBeginElement name attrs) testType
| name=="{jabber:server}presence"
&& matchAttribMaybe "type" testType attrs
= True
isPresenceOf _ _ = False
isClientPresenceOf (EventBeginElement name attrs) testType
| name=="{jabber:client}presence"
&& matchAttribMaybe "type" testType attrs
= True
isClientPresenceOf _ _ = False
handlePresenceProbe session stanza = do
withJust (lookupAttrib "to" (tagAttrs stanza)) $ \to -> do
-- withJust (lookupAttrib "from" (tagAttrs stanza)) $ \from -> do
jid <- liftIO $ parseAddressJID $ textToByteString to
withJust (name jid) $ \user -> do
liftIO $ L.putStrLn $ "RECEIVED PROBE "<++>bshow (peerAddress session,to)
liftIO $ do
subs <- getSubscribers (peerSessionFactory session) user
liftIO $ L.putStrLn $ "subscribers for "<++>bshow user<++>": " <++>bshow subs
forM_ subs $ \jidstr -> do
handleIO_ (return ()) $ do
L.putStrLn $ "parsing " <++>jidstr
sub <- parseHostNameJID jidstr
putStrLn $ "comparing " ++show (peer sub , peerAddress session)
when (peer sub == discardPort (peerAddress session)) $ do
ps <- userStatus session user
mapM_ (announcePresence session) ps
return ()
subscribeToPresence subscribers peer_jid user = do
pjid <- parseAddressJID peer_jid
if Set.member pjid subscribers
then return ()
else return ()
bare (JID n host _) = JID n host Nothing
presenceErrorRemoteNotFound iqid from to = return
[ EventBeginElement "{stream:client}presence"
( case iqid of { Nothing -> id; Just iqid -> ( attr "id" iqid :) }
$ [ attr "from" to
, attr "type" "error"
] )
, EventBeginElement "{stream:client}error"
[ attr "type" "modify"]
, EventBeginElement "{urn:ietf:params:xml:ns:xmpp-stanzas}remote-server-not-found"
[]
, EventEndElement "{urn:ietf:params:xml:ns:xmpp-stanzas}remote-server-not-found"
, EventEndElement "{stream:client}error"
, EventEndElement "{stream:client}presence"
]
presenceSubscribed from = return
[ EventBeginElement "{stream:client}presence"
[ attr "from" from
, attr "type" "subscribed"
]
, EventEndElement "{stream:client}presence"
]
clientRequestsSubscription session cmdChan stanza = do
liftIO $ do
putStrLn $ "CLIENT PRESENCE SUBSCRIBE " ++ show stanza
withJust (lookupAttrib "to" (tagAttrs stanza)) $ \to_str0 -> do
let to_str = S.takeWhile (/='/') to_str0
from = lookupAttrib "from" (tagAttrs stanza)
iqid = lookupAttrib "id" (tagAttrs stanza)
let handleError e | isDoesNotExistError e = do
putStrLn $ "remote-server-not-found"
r <- presenceErrorRemoteNotFound iqid from to_str
atomically $ writeTChan cmdChan (Send r)
handleError e = do
putStrLn $ "ERROR: "++ show e
handleIO handleError $ do
let to_str' = textToByteString to_str
to_jid <- fmap bare $ parseHostNameJID to_str'
if (is_remote . peer) to_jid
then do
addSolicited session to_str' to_jid
putStrLn $ "added to solicited: " ++ show to_jid
-- TODO: notify peer.
else do
-- addLocalSubscriber session to_str
-- self <- getJID session
r <- presenceSubscribed to_str -- self
atomically $ writeTChan cmdChan (Send r)
return ()
stanzaFromTo ::
JabberPeerSession session =>
session -> Event -> IO (Maybe (JID, JID))
stanzaFromTo session stanza =
let lookup key = fmap textToByteString (lookupAttrib key (tagAttrs stanza))
parse jidstr = handleIO_ (return Nothing) (fmap Just $ parseAddressJID jidstr)
in case liftM2 (,) (lookup "from") (lookup "to") of
Nothing -> return Nothing
Just (from,to) -> do
mfrom <- parse from
mto <- parse to
case liftM2 (,) mfrom mto of
Nothing -> return Nothing
Just (from,to) -> do
let fromjid = JID (name from) (peerAddress session) Nothing
return $ Just (fromjid,to)
peerRequestsSubsription session stanza = do
liftIO $ putStrLn $ "PEER PRESENCE SUBSCRIBE " ++ show stanza
whenJust (liftIO . handleIO (\e -> putStrLn ("peerRequestsSubsription: "++show e) >> return Nothing)
$ stanzaFromTo session stanza) $ \(fromjid,tojid) -> do
withJust (name tojid) $ \user -> do
subs <- liftIO $ do
subs <- getSubscribers (peerSessionFactory session) user
msubs <- flip mapM subs $ \str -> do
handleIO_ (return Nothing)
(fmap Just $ parseHostNameJID str)
return (catMaybes msubs)
if elem fromjid subs
then do
liftIO . L.putStrLn $ bshow fromjid <++> " already subscribed to " <++> user
-- if already subscribed, reply
liftIO $ do
sendPeerMessage session (Approval tojid fromjid)
ps <- userStatus session user
mapM_ (announcePresence session) ps
else
liftIO $ processRequest session user fromjid
clientApprovesSubscription session stanza = do
liftIO $ putStrLn $ "CLIENT APPROVES SUBSCRIPTION"
withJust (lookupAttrib "to" (tagAttrs stanza)) $ \to_str -> do
liftIO $ approveSubscriber session (textToByteString to_str)
peerApprovesSubscription session stanza = do
liftIO $ putStrLn $ "PEER APPROVES SUBSCRIPTION"
whenJust (liftIO . handleIO (\e -> putStrLn ("peerApprovesSubscription: "++show e) >> return Nothing)
$ stanzaFromTo session stanza) $ \(fromjid,tojid) -> do
withJust (name tojid) $ \user -> do
liftIO $ processApproval session user fromjid
fromPeer :: (MonadThrow m,MonadIO m, JabberPeerSession session) =>
session -> Sink XML.Event m ()
fromPeer session = doNestingXML $ do
let log = liftIO . L.putStrLn . ("(P) " <++>)
withXML $ \begindoc -> do
when (begindoc==EventBeginDocument) $ do
log "begin-doc"
withXML $ \xml -> do
withJust (elementAttrs "stream" xml) $ \stream_attrs -> do
log $ "stream atributes: " <++> bshow stream_attrs
fix $ \loop -> do
log "waiting for stanza."
whenJust nextElement $ \stanza -> do
stanza_lvl <- nesting
let unhandledStanza = do
xs <- gatherElement stanza Seq.empty
prettyPrint "P: " (toList xs)
case () of
_ | stanza `isPresenceOf` presenceTypeOnline
-> handlePeerPresence session stanza True
_ | stanza `isPresenceOf` presenceTypeOffline
-> handlePeerPresence session stanza False
_ | stanza `isPresenceOf` presenceTypeProbe
-> handlePresenceProbe session stanza
_ | stanza `isPresenceOf` presenceTypeSubscribe
-> peerRequestsSubsription session stanza
_ | stanza `isPresenceOf` presenceTypeSubscribed
-> peerApprovesSubscription session stanza
_ -> unhandledStanza
awaitCloser stanza_lvl
loop
log $ "end of stream"
withXML $ \xml -> do
log $ "end-of-document: " <++> bshow xml
newServerConnections = newTVar Map.empty
data CachedMessages = CachedMessages
{ presences :: Map JID JabberShow
, probes :: Map JID (Set (Bool,JID)) -- False means solicitation rather than probe
, approvals :: Map JID (Set JID)
}
newCache = CachedMessages Map.empty Map.empty Map.empty
connect_to_server chan peer = (>> return ()) . runMaybeT $ do
let port = 5269 :: Int
-- We'll cache Presence notifications until the socket
-- is ready.
cached <- liftIO $ newIORef newCache
let mmInsert val key mm = Map.alter f key mm
where
f Nothing = Just $ Set.singleton val
f (Just set) = Just $ Set.insert val set
let cacheCmd (OutBoundPresence (Presence jid Offline)) cached = do
cache <- readIORef cached
writeIORef cached (cache { presences=Map.delete jid . presences $ cache })
cacheCmd (OutBoundPresence p@(Presence jid st)) cached = do
cache <- readIORef cached
writeIORef cached (cache { presences=Map.insert jid st . presences $ cache })
cacheCmd (PresenceProbe from to) cached = do
cache <- readIORef cached
let probes' = mmInsert (True,from) to $ probes cache
writeIORef cached (cache { probes=probes' })
cacheCmd (Solicitation from to) cached = do
cache <- readIORef cached
let probes' = mmInsert (False,from) to $ probes cache
writeIORef cached (cache { probes=probes' })
cacheCmd (Approval from to) cached = do
cache <- readIORef cached
let approvals' = mmInsert from to $ approvals cache
writeIORef cached (cache { approvals=approvals' })
fix $ \sendmsgs -> do
connected <- liftIO . async $ connect' (peerAddr peer) port
sock <- MaybeT . fix $ \loop -> do
e <- atomically $ orElse
(fmap Right $ waitSTM connected)
(fmap Left $ readTChan chan)
case e of
Left cmd -> cacheCmd cmd cached >> loop
Right sock -> return sock
retry <- do
(cache,snk) <- liftIO $ do
h <- socketToHandle sock ReadWriteMode
hSetBuffering h NoBuffering
cache <- readIORef $ cached
-- hint garbage collector: we're done with this...
writeIORef cached newCache
return (cache,packetSink h)
MaybeT $ handleOutgoingToPeer (restrictSocket sock) cache chan snk
liftIO $ cacheCmd retry cached
liftIO $ putStrLn $ "retrying " ++ show retry
sendmsgs
greetPeer =
[ EventBeginDocument
, EventBeginElement (streamP "stream")
[ attr "xmlns" "jabber:server"
, attr "version" "1.0"
]
]
goodbyePeer =
[ EventEndElement (streamP "stream")
, EventEndDocument
]
presenceStanza sock fromjid tojid typ = do
addr <- getSocketName sock
let jidstr jid = toStrict . L.decodeUtf8
$ name jid <$++> "@"
++> showPeer (RemotePeer addr)
<++?> "/" <++$> resource jid
from = jidstr fromjid
to = toStrict . L.decodeUtf8
$ name tojid <$++> "@"
++> showPeer (peer tojid)
return
[ EventBeginElement "{jabber:server}presence"
[ attr "from" from
, attr "to" to
, attr "type" typ
]
, EventEndElement "{jabber:server}presence"
]
{-
toPeerChain
:: SocketLike sock =>
sock
-> CachedMessages
-> TChan OutBoundMessage
-> Sink ByteString IO b
-> IO b
toPeerChain sock cache chan snk = toPeer sock cache chan $$ renderChunks =$ snk
-}
toPeer
:: SocketLike sock =>
sock
-> CachedMessages
-> TChan OutBoundMessage
-> (Maybe OutBoundMessage -> IO ())
-> ConduitM i [Event] IO ()
toPeer sock cache chan fail = do
let -- log = liftIO . L.putStrLn . ("(>P) " <++>)
send xs = yield xs >> prettyPrint ">P: " xs -- >> return (3::Int)
checkConnection cmd = do
liftIO $ catch (getPeerName sock >> return ())
(\_ -> fail . Just $ cmd)
sendOrFail getXML cmd = do
checkConnection cmd
r <- liftIO $ getXML
-- handleIO (\e -> putStrLn ("ERROR: "++show e) >> return []) getXML
yieldOr r (fail . Just $ cmd)
prettyPrint ">P: " r
sendPresence presence =
sendOrFail (xmlifyPresenceForPeer sock presence)
(OutBoundPresence presence)
sendProbe from to =
sendOrFail (presenceStanza sock from to "probe")
(PresenceProbe from to)
sendSolicitation from to =
sendOrFail (presenceStanza sock from to "subscribe")
(Solicitation from to)
sendApproval from to =
sendOrFail (presenceStanza sock from to "subscribed")
(Approval from to)
send greetPeer
forM_ (Map.assocs . approvals $ cache) $ \(to,froms) -> do
forM_ (Set.toList froms) $ \from -> do
liftIO $ L.putStrLn "sending cached approval..."
sendApproval from to
forM_ (Map.assocs . presences $ cache) $ \(jid,st) -> do
sendPresence (Presence jid st)
forM_ (Map.assocs . probes $ cache) $ \(to,froms) -> do
forM_ (Set.toList froms) $ \(got,from) -> do
if got
then do
liftIO $ L.putStrLn "sending cached probe..."
sendProbe from to
else do
liftIO $ L.putStrLn "sending cached solicitation..."
sendSolicitation from to
fix $ \loop -> do
event <- lift . atomically $ readTChan chan
case event of
OutBoundPresence p -> sendPresence p
PresenceProbe from to -> do
liftIO $ L.putStrLn "sending live probe..."
sendProbe from to
Solicitation from to -> do
liftIO $ L.putStrLn "sending live solicitation..."
sendSolicitation from to
Approval from to -> do
liftIO . L.putStrLn $ "sending approval "<++>bshow (from,to)
sendApproval from to
loop
send goodbyePeer
handleOutgoingToPeer
:: SocketLike sock =>
sock
-> CachedMessages
-> TChan OutBoundMessage
-> Sink ByteString IO ()
-> IO (Maybe OutBoundMessage)
handleOutgoingToPeer sock cache chan snk = do
p <- getPeerName sock
L.putStrLn $ "(>P) connected " <++> showPeer (RemotePeer p)
failed <- newIORef Nothing
let failure cmd = do
writeIORef failed cmd
putStrLn $ "Failed: " ++ show cmd
finally (
#ifdef RENDERFLUSH
handle (\(IOError _ _ _ _ _ _) -> return ()) $
toPeer sock cache chan failure
$$ flushList
=$= renderBuilderFlush def
=$= builderToByteStringFlush
=$= discardFlush
=$ snk
#else
handle (\(IOError _ _ _ _ _ _) -> return ()) $ toPeer sock cache chan failure $$ renderChunks =$ snk
#endif
) $ L.putStrLn $ "(>P) disconnected " <++> showPeer (RemotePeer p)
readIORef failed
connect' :: SockAddr -> Int -> IO (Maybe Socket)
connect' addr port = do
proto <- getProtocolNumber "tcp"
{-
-- Given (host :: HostName) ...
let hints = defaultHints { addrFlags = [AI_ADDRCONFIG]
, addrProtocol = proto
, addrSocketType = Stream }
addrs <- getAddrInfo (Just hints) (Just host) (Just serv)
firstSuccessful $ map tryToConnect addrs
-}
let getport (SockAddrInet port _) = port
getport (SockAddrInet6 port _ _ _) = port
let doException e@(IOError _ _ _ _ _ _) = do
-- let doException (SomeException e) = do
L.putStrLn $ "\nFailed to reach "<++> showPeer (RemotePeer addr) <++> " on port "<++>bshow port<++>": " <++> bshow e
return Nothing
handle doException
$ tryToConnect proto (addr `withPort` port)
where
tryToConnect proto addr =
bracketOnError
(socket (socketFamily addr) Stream proto)
(sClose ) -- only done if there's an error
(\sock -> do
connect sock addr
return (Just sock) -- socketToHandle sock ReadWriteMode
)
sendMessage :: TVar (Map Peer (TChan OutBoundMessage, ThreadId)) -> OutBoundMessage -> Peer -> IO ()
sendMessage cons msg peer0 = do
let peer = discardPort peer0
found <- atomically $ do
consmap <- readTVar cons
return (Map.lookup peer consmap)
let newEntry = do
chan <- atomically newTChan
t <- forkIO $ connect_to_server chan peer
-- L.putStrLn $ "remote-map new: " <++> showPeer peer
return (True,(chan,t))
(is_new,entry) <- maybe newEntry
( \(chan,t) -> do
st <- threadStatus t
let running = do
-- L.putStrLn $ "remote-map, thread running: " <++> showPeer peer
return (False,(chan,t))
died = do
-- L.putStrLn $ "remote-map, thread died("<++>bshow st<++>"): " <++> showPeer peer
newEntry
case st of
ThreadRunning -> running
ThreadBlocked _ -> running
ThreadDied -> died
ThreadFinished -> died
)
found
-- L.putStrLn $ "sendMessage ->"<++>showPeer peer<++>": "<++>bshow msg
atomically $ writeTChan (fst entry) msg
when is_new . atomically $
readTVar cons >>= writeTVar cons . Map.insert peer entry
seekRemotePeers :: JabberPeerSession config =>
XMPPPeerClass config -> TChan Presence -> TVar (Map Peer (TChan OutBoundMessage, ThreadId)) -> IO b0
seekRemotePeers config chan server_connections = do
fix $ \loop -> do
event <- atomically $ readTChan chan
case event of
p@(Presence jid stat) | not (is_remote (peer jid)) -> do
-- L.putStrLn $ "seekRemotePeers: " <++> L.show jid <++> " " <++> bshow stat
runMaybeT $ do
u <- MaybeT . return $ name jid
subscribers <- liftIO $ do
subs <- getSubscribers config u
mapM parseHostNameJID subs
-- liftIO . L.putStrLn $ "subscribers: " <++> bshow subscribers
let peers = Set.map peer (Set.fromList subscribers)
forM_ (Set.toList peers) $ \peer -> do
when (is_remote peer) $
liftIO $ sendMessage server_connections (OutBoundPresence p) peer
-- TODO: send presence probes for buddies
-- TODO: cache remote presences for clients
_ -> return (Just ())
loop
xmlifyPresenceForPeer sock (Presence jid stat) = do
-- TODO: accept socket argument and determine local ip address
-- connected to this peer.
addr <- getSocketName sock
let n = name jid
rsc = resource jid
jidstr = toStrict . L.decodeUtf8
$ n <$++> "@" ++> showPeer (RemotePeer addr) <++?> "/" <++$> rsc
return
[ EventBeginElement "{jabber:server}presence"
(attr "from" jidstr:typ stat)
, EventBeginElement "{jabber:server}show" []
, EventContent (ContentText . shw $ stat)
, EventEndElement "{jabber:server}show"
, EventEndElement "{jabber:server}presence"
]
where
typ Offline = [attr "type" "unavailable"]
typ _ = []
shw Available = "chat"
shw Away = "away"
shw Offline = "away" -- Is this right?