outgoing connections to peers added to XMPP.hs.

It still uses Handle for now, TODO: change to a ByteString sink.

3 files changed, 279 insertions, 68 deletions

diff --git a/Presence/XMPP.hs b/Presence/XMPP.hs
index 1a4b0e7b..417b3ce7 100644
--- a/Presence/XMPP.hs
+++ b/Presence/XMPP.hs
@@ -16,8 +16,29 @@ import SocketLike
 import ByteStringOperators
 
 import Data.HList
-import Network.Socket (Family)
-import Network.BSD (PortNumber)
+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.Exception
+    ( bracketOnError )
 import Control.Concurrent.STM
 import Data.Conduit
 import qualified Data.Conduit.List as CL
@@ -27,6 +48,8 @@ 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
@@ -39,7 +62,6 @@ import Control.Monad as Monad
 import Text.XML.Stream.Parse (parseBytes,content)
 import Text.XML.Stream.Render
 import Data.XML.Types as XML
-import Network.BSD (getHostName,hostName,hostAliases)
 import Data.Text.Encoding      as S (decodeUtf8,encodeUtf8)
 import Data.Text.Lazy.Encoding as L (decodeUtf8,encodeUtf8)
 import Data.Text.Lazy (toStrict)
@@ -54,6 +76,11 @@ import Data.List (find)
 import qualified Text.Show.ByteString as L
 import NestingXML
 import qualified Data.Set as Set
+import qualified Data.Map as Map
+import GHC.Conc 
+    ( threadStatus
+    , ThreadStatus(..)
+    )
 
 data Commands = Send [XML.Event] | QuitThread
  deriving Prelude.Show
@@ -534,9 +561,174 @@ fromPeer session = doNestingXML $ do
 
 
 
+{-
 seekRemotePeers :: XMPPConfig config =>
      config -> TChan Presence -> IO ()
 seekRemotePeers config chan = do
     putStrLn "unimplemented: seekRemotePeers"
     -- TODO
     return ()
+-}
+
+data OutBoundMessage = OutBoundPresence Presence
+ deriving Prelude.Show
+
+newServerConnections = atomically $ newTVar Map.empty
+
+connect_to_server chan peer = (>> return ()) . runMaybeT $ do
+    let port = 5269 :: Int
+
+    connected <- liftIO . async $ connect' (peerAddr peer) port
+
+    -- We'll cache Presence notifications until the socket
+    -- is ready.
+    cached <- liftIO $ newIORef Map.empty
+
+    sock <- MaybeT . fix $ \loop -> do
+        e <- atomically $ orElse
+                (fmap Right $ waitSTM connected)
+                (fmap Left  $ readTChan chan)
+        case e of
+            Left (OutBoundPresence (Presence jid Offline)) -> do
+                cached_map <- readIORef cached
+                writeIORef cached (Map.delete jid cached_map)
+                loop
+            Left (OutBoundPresence p@(Presence jid st)) -> do
+                cached_map <- readIORef cached
+                writeIORef cached (Map.insert jid st cached_map)
+                loop
+            {-
+            Left event -> do
+                L.putStrLn $ "REMOTE-OUT DISCARDED: " <++> bshow event
+                loop
+            -}
+            Right sock -> return sock
+
+    liftIO $ do
+        h <- socketToHandle sock ReadWriteMode
+        hSetBuffering h NoBuffering
+        L.hPutStrLn h "<stream>"
+        L.putStrLn $ "OUT peer: <stream>"
+        cache <- fmap Map.assocs . readIORef $ cached
+        writeIORef cached Map.empty -- hint garbage collector: we're done with this
+        forM_ cache $ \(jid,st) -> do
+            r <- xmlifyPresenceForPeer sock (Presence jid st)
+            L.hPutStrLn h r
+            L.putStrLn $ "OUT peer: (cache)\n" <++> r <++> "\n"
+        fix $ \loop -> do
+            event <- atomically $ readTChan chan
+            case event of
+              OutBoundPresence p -> do
+                r <- xmlifyPresenceForPeer sock p
+                L.hPutStrLn h r
+                L.putStrLn $ "OUT peer:\n" <++> r <++> "\n"
+            loop
+        L.hPutStrLn h "</stream>"
+        L.putStrLn $ "OUT peer: </stream>"
+
+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 withPort (SockAddrInet _ a)      port = SockAddrInet (toEnum port) a
+        withPort (SockAddrInet6 _ a b c) port = SockAddrInet6 (toEnum port) a b c
+    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 cons msg peer = do
+    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 :: XMPPConfig config =>
+     config -> TChan Presence -> IO b0
+seekRemotePeers config chan = do
+    server_connections <- newServerConnections
+    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
+        jid_str = n <$++> "@" <?++> showPeer (RemotePeer addr) <++?>  "/" <++$> rsc 
+    return . L.unlines $
+      [ "<presence from='" <++> jid_str <++> "' " <++> typ stat <++> ">"
+      , "<show>" <++> shw stat <++> "</show>"
+      , "</presence>"
+      ]
+ where
+    typ Offline = " type='unavailable'"
+    typ _       = ""
+    shw Available = "chat"
+    shw Away      = "away"
+    shw Offline   = "away" -- Is this right?
diff --git a/Presence/XMPPServer.hs b/Presence/XMPPServer.hs
index f607989d..ff50ab1c 100644
--- a/Presence/XMPPServer.hs
+++ b/Presence/XMPPServer.hs
@@ -67,9 +67,6 @@ import qualified Data.Set as Set
 import GetHostByAddr
 import XMPPTypes
 
-is_remote (RemotePeer _) = True
-is_remote _              = False
-
 getNamesForPeer :: Peer -> IO [ByteString]
 getNamesForPeer LocalHost = fmap ((:[]) . pack) getHostName
 getNamesForPeer peer@(RemotePeer addr) = do
@@ -85,10 +82,6 @@ getNamesForPeer peer@(RemotePeer addr) = do
     return . map pack $ names
 
 
-peerAddr :: Peer -> SockAddr
-peerAddr (RemotePeer addr) = addr
--- peerAddr LocalHost = throw exception
-
 
 xmlifyPresenceForPeer sock (Presence jid stat) = do
     -- TODO: accept socket argument and determine local ip address
@@ -587,62 +580,6 @@ connect_to_server chan peer = (>> return ()) . runMaybeT $ do
         L.putStrLn $ "OUT peer: </stream>"
 
 
-splitJID :: ByteString -> (Maybe ByteString,ByteString,Maybe ByteString)
-splitJID bjid = 
-    let xs = L.splitWith (=='@') bjid
-        ys = L.splitWith (=='/') (last xs)
-        server = head ys
-        name
-            = case xs of
-                (n:s:_) -> Just n
-                (s:_)   -> Nothing
-        rsrc = case ys of
-                (s:_:_) -> Just $ last ys
-                _       -> Nothing
-    in (name,server,rsrc)
-
-strip_brackets s = 
-    case L.uncons s of
-      Just ('[',t) -> L.takeWhile (/=']') t
-      _            -> s
-
-parseAddressJID :: ByteString -> IO JID
-parseAddressJID jid = do
-    let (name,peer_string,rsc) = splitJID jid
-        hints = Just $ defaultHints { addrFlags = [ {- AI_NUMERICHOST, -} AI_CANONNAME ] }
-        peer_string' = unpack . strip_brackets $ peer_string
-    peer <- do
-        -- putStrLn $ "getAddrInfo 2 " ++ Prelude.show ( Just (unpack peer_string))
-        info <- getAddrInfo hints (Just peer_string') Nothing -- (Just "xmpp-server")
-        let info0 = head info
-        return . RemotePeer . addrAddress $ info0
-    return $ JID name peer rsc
-
-parseHostNameJID :: ByteString -> IO JID
-parseHostNameJID jid = do
-    let (name,peer_string,rsc) = splitJID jid
-        hints = Just $ defaultHints { addrFlags = [ AI_CANONNAME ] }
-    peer <- do
-        if peer_string=="localhost"
-          then return LocalHost
-          else do
-            -- putStrLn $ "getAddrInfo 3 " ++ Prelude.show ( Just (unpack peer_string))
-            info <- getAddrInfo hints (Just (unpack peer_string)) Nothing -- (Just "xmpp-server")
-            let info0 = head info
-                cname = addrCanonName info0
-            if cname==Just "localhost"
-              then return LocalHost
-              else do
-                self <- getHostName
-                return $ if Just self==cname
-                          then LocalHost 
-                          else RemotePeer (addrAddress info0)
-    return $ JID name peer rsc
-
-socketFamily (SockAddrInet _ _)      = AF_INET
-socketFamily (SockAddrInet6 _ _ _ _) = AF_INET6
-socketFamily (SockAddrUnix _)        = AF_UNIX
-
 connect' :: SockAddr -> Int -> IO (Maybe Socket)
 connect' addr port = do
     proto <- getProtocolNumber "tcp"
diff --git a/Presence/XMPPTypes.hs b/Presence/XMPPTypes.hs
index e3bbfd16..8af1018c 100644
--- a/Presence/XMPPTypes.hs
+++ b/Presence/XMPPTypes.hs
@@ -2,10 +2,28 @@
 {-# LANGUAGE TypeFamilies        #-}
 module XMPPTypes where
 
-import Network.Socket (Socket,SockAddr(..))
+import Network.Socket 
+    ( Socket
+    , Family(..)
+    , SockAddr(..)
+    , getAddrInfo
+    , addrCanonName
+    , addrAddress
+    , defaultHints
+    , AddrInfo(..)
+    , AddrInfoFlag(..)
+    )
+import Network.BSD (getHostName)
 import System.IO (Handle)
 import Control.Concurrent.STM (TChan)
-import Data.ByteString.Lazy.Char8 as L  (ByteString,unpack,pack)
+import Data.ByteString.Lazy.Char8 as L  
+    ( ByteString
+    , unpack
+    , pack 
+    , splitWith
+    , uncons
+    , takeWhile
+    )
 import Text.Show.ByteString as L
 import Data.Binary.Builder as B
 import Data.Binary.Put
@@ -82,4 +100,68 @@ showPeer (RemotePeer addr@(SockAddrInet6 _ _ _ _)) = pack $ stripColon (Prelude.
           where
             (pre,bracket) = break (==']') s
 
+is_remote (RemotePeer _) = True
+is_remote _              = False
+
+parseHostNameJID :: ByteString -> IO JID
+parseHostNameJID jid = do
+    let (name,peer_string,rsc) = splitJID jid
+        hints = Just $ defaultHints { addrFlags = [ AI_CANONNAME ] }
+    peer <- do
+        if peer_string=="localhost"
+          then return LocalHost
+          else do
+            -- putStrLn $ "getAddrInfo 3 " ++ Prelude.show ( Just (unpack peer_string))
+            info <- getAddrInfo hints (Just (unpack peer_string)) Nothing -- (Just "xmpp-server")
+            let info0 = head info
+                cname = addrCanonName info0
+            if cname==Just "localhost"
+              then return LocalHost
+              else do
+                self <- getHostName
+                return $ if Just self==cname
+                          then LocalHost 
+                          else RemotePeer (addrAddress info0)
+    return $ JID name peer rsc
+
+splitJID :: ByteString -> (Maybe ByteString,ByteString,Maybe ByteString)
+splitJID bjid = 
+    let xs = L.splitWith (=='@') bjid
+        ys = L.splitWith (=='/') (last xs)
+        server = head ys
+        name
+            = case xs of
+                (n:s:_) -> Just n
+                (s:_)   -> Nothing
+        rsrc = case ys of
+                (s:_:_) -> Just $ last ys
+                _       -> Nothing
+    in (name,server,rsrc)
+
+strip_brackets s = 
+    case L.uncons s of
+      Just ('[',t) -> L.takeWhile (/=']') t
+      _            -> s
+
+
+parseAddressJID :: ByteString -> IO JID
+parseAddressJID jid = do
+    let (name,peer_string,rsc) = splitJID jid
+        hints = Just $ defaultHints { addrFlags = [ {- AI_NUMERICHOST, -} AI_CANONNAME ] }
+        peer_string' = unpack . strip_brackets $ peer_string
+    peer <- do
+        -- putStrLn $ "getAddrInfo 2 " ++ Prelude.show ( Just (unpack peer_string))
+        info <- getAddrInfo hints (Just peer_string') Nothing -- (Just "xmpp-server")
+        let info0 = head info
+        return . RemotePeer . addrAddress $ info0
+    return $ JID name peer rsc
+
+peerAddr :: Peer -> SockAddr
+peerAddr (RemotePeer addr) = addr
+-- peerAddr LocalHost = throw exception
+
+socketFamily (SockAddrInet _ _)      = AF_INET
+socketFamily (SockAddrInet6 _ _ _ _) = AF_INET6
+socketFamily (SockAddrUnix _)        = AF_UNIX
+