Parse/Lex fixes. Presence now received by remote peer.

1 files changed, 149 insertions, 28 deletions

diff --git a/Presence/XMPPServer.hs b/Presence/XMPPServer.hs
index 062fcacb..76c14f52 100644
--- a/Presence/XMPPServer.hs
+++ b/Presence/XMPPServer.hs
@@ -10,12 +10,19 @@ import Todo
 import Data.HList.TypeEqGeneric1()
 import Data.HList.TypeCastGeneric1()
 import ByteStringOperators
+import System.IO 
+    ( IOMode(..)
+    , BufferMode(..)
+    , hSetBuffering
+    )
 
 import Server
 import Data.ByteString.Lazy.Char8 as L 
     ( hPutStrLn
     , unlines
+    , lines
     , splitWith
+    , drop
     , ByteString
     , pack
     , unpack )
@@ -26,11 +33,11 @@ import System.IO
     )
 import Data.HList
 import AdaptServer
-import Text.XML.HaXml.Lex (xmlLex)
+import Text.XML.HaXml.Lex (xmlLex,TokenT(..))
 import Text.XML.HaXml.Parse  (XParser,xmlParseWith,element,{-doctypedecl,-}processinginstruction,elemOpenTag,elemCloseTag)
 import Text.XML.HaXml.Types as Hax hiding (Element) -- (ElemTag,QName(..),Namespace(..),Element(..),Content(..),AttValue(..))
 import qualified Text.XML.HaXml.Types as Hax (Element(..))
-import Text.XML.HaXml.Posn (Posn)
+import Text.XML.HaXml.Posn (Posn, posnLine, posnColumn)
 import Text.XML.HaXml.Lex  (TokenT)
 import qualified Text.XML.HaXml.Pretty as PP
 import Text.PrettyPrint
@@ -45,12 +52,14 @@ import Control.Monad.IO.Class
 import Control.DeepSeq
 import Control.Concurrent.STM
 import Control.Concurrent
-import Control.Exception
+import Control.Exception as Exception
 import Text.Show.ByteString as L
 import Data.Binary.Builder as B
 import Data.Binary.Put
 import qualified Data.Map as Map
 import GHC.Conc
+import Network.BSD
+import Control.Concurrent.Async
 
 -- | Jabber ID (JID) datatype
 data JID = JID { name :: Maybe ByteString
@@ -332,7 +341,8 @@ xmppParse ls = runTryParse $ do
 listenForXmppClients session_factory port st = do
     -- standard port: 5222
     let (start,dopkt) = 
-            adaptServer ( xmlLex "local-client" . unpack
+            adaptServer ( dropTill
+                        , xmlLexPartial "local-client" . unpack
                         , xmppParse)
                         (startCon session_factory,doCon)
     doServer (port .*. st)
@@ -343,39 +353,49 @@ listenForXmppClients session_factory port st = do
 startPeer session_factory sock st = do
     let h = hOccursFst st :: Handle 
     name <- fmap bshow $ getPeerName sock
-    L.putStrLn $ "REMOTE: connected " <++> name
-    let quit = L.putStrLn $ "REMOTE: disconnected " <++> name
+    L.putStrLn $ "REMOTE-IN: connected " <++> name
+    let quit = L.putStrLn $ "REMOTE-IN: disconnected " <++> name
     return ( ConnectionFinalizer quit .*. st )
 
 doPeer st elem cont = do
+    L.putStrLn $ "REMOTE-IN: received " <++> bshow elem
     cont ()
 
+xmlLexPartial name cs =
+    let ls = xmlLex name cs
+        isTokError (_,TokError _) = True
+        isTokError _              = False
+        (gs,bs) = break isTokError ls
+    in if any (not . isTokError) bs
+        then ls
+        else gs
+        
+
 listenForRemotePeers session_factory port st = do
     -- standard port: 5269
     let (start,dopkt) = 
-            adaptServer ( xmlLex "remote-peer" . unpack
+            adaptServer ( dropTill
+                        , xmlLexPartial "remote-inbound" . unpack
                         , xmppParse)
                         (startPeer session_factory,doPeer)
     doServer (port .*. st)
              dopkt
              start
 
-newServerConnections = atomically $ newTVar Map.empty
-{-
-sendMessage cons msg peer = do
-    (is_new,entry) <- atomically $ do
-        consmap <- readTVar cons
-        let found = Map.lookup peer consmap
-            newEntry = ()
-            entry = maybe newEntry id found
-            is_new = isNothing found
-        when is_new
-          $ writeTVar cons (Map.insert peer entry consmap)
-        return (is_new,entry)
-    L.putStrLn $ "sendMessage ->"<++>peer<++>": "<++>bshow msg
-    when is_new $ connect_to_server entry peer
+dropTill bs ((fst->posn):_) =
+  let ls = zip [1..] (L.lines bs)
+      ln = posnLine posn
+      col = posnColumn posn
+      ls' = map snd $ dropWhile ((<ln).fst) ls
+  in case ls' of
+        []           -> ""
+        fstLine:ls'' -> foldr1 (<++>) (L.drop (fromIntegral (col-1)) fstLine : ls'')
 
--}
+
+data OutBoundMessage = OutBoundPresence Presence
+ deriving Prelude.Show
+
+newServerConnections = atomically $ newTVar Map.empty
 
 sendMessage cons msg peer = do
     found <- atomically $ do
@@ -384,18 +404,87 @@ sendMessage cons msg peer = do
     let newEntry = do
           chan <- atomically newTChan
           t <- forkIO $ connect_to_server chan peer
-          return (chan,t) 
-    entry <- maybe newEntry
+          L.putStrLn $ "remote-map new: " <++> 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: " <++> peer
+                                return (False,(chan,t))
+                            died = do
+                                L.putStrLn $ "remote-map, thread died("<++>bshow st<++>"): " <++> peer
+                                newEntry 
                         case st of 
-                          ThreadRunning -> return (chan,t)
-                          _             -> newEntry 
+                          ThreadRunning   -> running
+                          ThreadBlocked _ -> running
+                          ThreadDied      -> died
+                          ThreadFinished  -> died
                         )
           found
     L.putStrLn $ "sendMessage ->"<++>peer<++>": "<++>bshow msg
+    atomically $ writeTChan (fst entry) msg
+    when is_new . atomically $
+        readTVar cons >>= writeTVar cons . Map.insert peer entry
+
+connect_to_server chan peer = (>> return ()) . runMaybeT $ do
+    let port = "5269"
+
+    connected <- liftIO . async $ connect' (unpack peer) port
+
+    sock <- MaybeT . fix $ \loop -> do
+        e <- atomically $ orElse
+                (fmap Right $ waitSTM connected)
+                (fmap Left  $ readTChan chan)
+        case e of
+            Left event -> do
+                L.putStrLn $ "REMOTE-OUT NOT READY: " <++> bshow event
+                loop
+            Right sock -> return sock
+
+    liftIO $ do
+        h <- socketToHandle sock ReadWriteMode
+        hSetBuffering h NoBuffering
+        hPutStrLn h "<stream>"
+        L.putStrLn $ "REMOTE-OUT: <stream>"
+        fix $ \loop -> do
+            event <- atomically $ readTChan chan
+            case event of
+              OutBoundPresence p -> do
+                let r = xmlifyPresence p
+                hPutStrLn h r
+                L.putStrLn $ "REMOTE-OUT:\n" <++> r <++> "\n"
+            loop
+        hPutStrLn h "</stream>"
+        L.putStrLn $ "REMOTE-OUT: </stream>"
 
-connect_to_server chan peer = return ()
+
+    {-
+    pending <- newTVarIO True
+    thread <- forkIO $
+        runMaybeT $ do
+            let port = "5269"
+            sock <- MaybeT $ connect' (unpack peer) port
+            liftIO $ do
+                h <- socketToHandle sock ReadWriteMode
+                hSetBuffering h NoBuffering
+                hPutStrLn h "<stream>"
+                atomically $ writeTVar pending False
+                fix $ \loop -> do
+                    event <- atomically $ readTChan chan
+                    case event of
+                      OutBoundPresence p -> do
+                        let r = xmlifyPresence p
+                        hPutStrLn h r
+                        L.putStrLn $ "REMOTE:\n" <++> r <++> "\n"
+                    loop
+                hPutStrLn h "</stream>"
+    fix $ \loop -> do
+        event <- atomically $ readTChan chan
+        when (readTVarIO pending) loop
+    joinThread thread
+    return ()
+    -}
 
 parseJID :: ByteString -> JID
 parseJID bjid = 
@@ -410,6 +499,38 @@ parseJID bjid =
                 _       -> Nothing
     in JID name server rsrc
 
+connect' :: HostName -> ServiceName -> IO (Maybe Socket)
+connect' host serv = do
+    proto <- getProtocolNumber "tcp"
+    let hints = defaultHints { addrFlags = [AI_ADDRCONFIG]
+                             , addrProtocol = proto
+                             , addrSocketType = Stream }
+    addrs <- getAddrInfo (Just hints) (Just host) (Just serv)
+    firstSuccessful $ map tryToConnect addrs
+  where
+  tryToConnect addr =
+    bracketOnError
+        (socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr))
+        (sClose )  -- only done if there's an error
+        (\sock -> do
+          connect sock (addrAddress addr)
+          return (Just sock) -- socketToHandle sock ReadWriteMode
+        )
+
+catchIO :: IO a -> (IOException -> IO a) -> IO a
+catchIO a h = Exception.catch a h
+
+-- Returns the first action from a list which does not throw an exception.
+-- If all the actions throw exceptions (and the list of actions is not empty),
+-- the last exception is thrown.
+firstSuccessful :: [IO a] -> IO a
+firstSuccessful [] = error "firstSuccessful: empty list"
+firstSuccessful (p:ps) = catchIO p $ \e ->
+    case ps of
+        [] -> Exception.throwIO e
+        _  -> firstSuccessful ps
+
+
 seekRemotePeers :: XMPPConfig config =>
      (ByteString -> Bool) -> config -> TChan Presence -> IO b0
 seekRemotePeers is_peer config chan = do
@@ -427,5 +548,5 @@ seekRemotePeers is_peer config chan = do
                     let jid = parseJID bjid
                         peer = server jid
                     when (is_peer peer) $
-                        liftIO $ sendMessage server_connections p peer
+                        liftIO $ sendMessage server_connections (OutBoundPresence p) peer
         loop