diff options
Diffstat (limited to 'xmppServer.hs')
-rw-r--r-- | xmppServer.hs | 160 |
1 files changed, 140 insertions, 20 deletions
diff --git a/xmppServer.hs b/xmppServer.hs index c720ea5f..459d8f8d 100644 --- a/xmppServer.hs +++ b/xmppServer.hs | |||
@@ -1,6 +1,9 @@ | |||
1 | {-# LANGUAGE OverloadedStrings #-} | ||
1 | import Control.Monad.Trans.Resource (runResourceT) | 2 | import Control.Monad.Trans.Resource (runResourceT) |
2 | import Control.Monad.Trans (lift) | 3 | import Control.Monad.Trans (lift) |
4 | import Control.Monad.IO.Class (liftIO) | ||
3 | import Control.Monad.Fix (fix) | 5 | import Control.Monad.Fix (fix) |
6 | import Control.Monad | ||
4 | import Control.Concurrent (forkIO) | 7 | import Control.Concurrent (forkIO) |
5 | import Control.Concurrent.STM | 8 | import Control.Concurrent.STM |
6 | -- import Control.Concurrent.STM.TChan | 9 | -- import Control.Concurrent.STM.TChan |
@@ -8,18 +11,42 @@ import Network.Socket | |||
8 | import XMPPTypes (withPort) | 11 | import XMPPTypes (withPort) |
9 | import Text.Printf | 12 | import Text.Printf |
10 | import System.Posix.Signals | 13 | import System.Posix.Signals |
14 | import Data.ByteString (ByteString) | ||
15 | import qualified Data.ByteString.Char8 as Strict8 | ||
16 | -- import qualified Data.ByteString.Lazy.Char8 as Lazy8 | ||
11 | 17 | ||
12 | import Data.Conduit | 18 | import Data.Conduit |
19 | import qualified Data.Conduit.List as CL | ||
20 | import qualified Data.Conduit.Binary as CB | ||
13 | 21 | ||
14 | import qualified Text.XML.Stream.Render as XML | 22 | import qualified Text.XML.Stream.Render as XML |
15 | import qualified Text.XML.Stream.Parse as XML | 23 | import qualified Text.XML.Stream.Parse as XML |
24 | import Data.XML.Types as XML | ||
25 | import Data.Maybe (catMaybes) | ||
26 | import Data.Monoid ( (<>) ) | ||
16 | 27 | ||
28 | import qualified Control.Concurrent.STM.UpdateStream as Slotted | ||
29 | import ControlMaybe | ||
30 | import NestingXML | ||
17 | import Server | 31 | import Server |
18 | 32 | ||
33 | |||
19 | wlog s = putStrLn s | 34 | wlog s = putStrLn s |
35 | where _ = s :: String | ||
36 | wlogb s = Strict8.putStrLn s | ||
20 | 37 | ||
21 | control sv = atomically . putTMVar (serverCommand sv) | 38 | control sv = atomically . putTMVar (serverCommand sv) |
22 | 39 | ||
40 | -- Note: This function ignores name space qualification | ||
41 | elementAttrs expected (EventBeginElement name attrs) | ||
42 | | nameLocalName name==expected | ||
43 | = return attrs | ||
44 | elementAttrs _ _ = mzero | ||
45 | |||
46 | getStreamName (EventBeginElement name _) = name | ||
47 | |||
48 | xmlStream :: ReadCommand -> WriteCommand -> ( Source IO XML.Event | ||
49 | , Sink XML.Event IO () ) | ||
23 | xmlStream conread conwrite = (xsrc,xsnk) | 50 | xmlStream conread conwrite = (xsrc,xsnk) |
24 | where | 51 | where |
25 | xsrc = src $= XML.parseBytes XML.def | 52 | xsrc = src $= XML.parseBytes XML.def |
@@ -30,34 +57,125 @@ xmlStream conread conwrite = (xsrc,xsnk) | |||
30 | maybe (return ()) -- lift . wlog $ "conread: Nothing") | 57 | maybe (return ()) -- lift . wlog $ "conread: Nothing") |
31 | (\v -> yield v >> src) | 58 | (\v -> yield v >> src) |
32 | v | 59 | v |
33 | snk = awaitForever $ lift . conwrite | 60 | snk = awaitForever $ liftIO . conwrite |
34 | 61 | ||
35 | 62 | ||
36 | forkConnection k pingflag src snk = do | 63 | type FlagCommand = IO Bool |
64 | type ReadCommand = IO (Maybe ByteString) | ||
65 | type WriteCommand = ByteString -> IO Bool | ||
66 | |||
67 | data Stanza | ||
68 | = UnrecognizedStanza { stanzaChan :: TChan (Maybe XML.Event) } | ||
69 | |||
70 | prettyPrint prefix xs = | ||
71 | liftIO $ | ||
72 | CL.sourceList xs | ||
73 | $= XML.renderBytes (XML.def { XML.rsPretty=True }) | ||
74 | =$= CB.lines | ||
75 | $$ CL.mapM_ (wlogb . (prefix <>)) | ||
76 | |||
77 | xmppInbound :: ConnectionKey -> FlagCommand | ||
78 | -> Source IO XML.Event | ||
79 | -> TChan Stanza | ||
80 | -> Sink XML.Event IO () | ||
81 | xmppInbound k pingflag src stanzas = doNestingXML $ do | ||
82 | withXML $ \begindoc -> do | ||
83 | when (begindoc==EventBeginDocument) $ do | ||
84 | -- liftIO . wlog $ "begin-doc" | ||
85 | withXML $ \xml -> do | ||
86 | withJust (elementAttrs "stream" xml) $ \stream_attrs -> do | ||
87 | -- liftIO . wlog $ "stream: " ++ show (getStreamName xml) | ||
88 | -- liftIO . wlog $ "stream atributes: " ++ show stream_attrs | ||
89 | fix $ \loop -> do | ||
90 | -- liftIO . wlog $ "waiting for stanza." | ||
91 | chan <- liftIO $ atomically newTChan | ||
92 | whenJust nextElement $ \stanza -> do | ||
93 | stanza_lvl <- nesting | ||
94 | liftIO . atomically $ writeTChan chan (Just stanza) | ||
95 | -- liftIO . wlog $ "stanza: "++show stanza | ||
96 | |||
97 | liftIO . atomically $ writeTChan stanzas $ | ||
98 | UnrecognizedStanza chan | ||
99 | doUntilCloser stanza_lvl $ \xml -> do | ||
100 | liftIO . atomically $ writeTChan chan (Just xml) | ||
101 | -- liftIO . wlog $ "-stanza: " ++ show xml | ||
102 | liftIO . atomically $ writeTChan chan Nothing | ||
103 | loop | ||
104 | |||
105 | |||
106 | chanContents :: TChan x -> IO [x] | ||
107 | chanContents ch = do | ||
108 | x <- atomically $ do | ||
109 | bempty <- isEmptyTChan ch | ||
110 | if bempty | ||
111 | then return Nothing | ||
112 | else fmap Just $ readTChan ch | ||
113 | maybe (return []) | ||
114 | (\x -> do | ||
115 | xs <- chanContents ch | ||
116 | return (x:xs)) | ||
117 | x | ||
118 | |||
119 | |||
120 | isEventBeginElement (EventBeginElement {}) = True | ||
121 | isEventBeginElement _ = False | ||
122 | |||
123 | isEventEndElement (EventEndElement {}) = True | ||
124 | isEventEndElement _ = False | ||
125 | |||
126 | forkConnection :: ConnectionKey | ||
127 | -> FlagCommand | ||
128 | -> Source IO XML.Event | ||
129 | -> Sink XML.Event IO () | ||
130 | -> TChan Stanza | ||
131 | -> IO (Slotted.UpdateStream () XML.Event) | ||
132 | forkConnection k pingflag src snk stanzas = do | ||
133 | rdone <- atomically newEmptyTMVar | ||
37 | forkIO $ do | 134 | forkIO $ do |
38 | src $$ awaitForever (lift . putStrLn . takeWhile (/=' ') . show) | 135 | -- src $$ awaitForever (lift . putStrLn . takeWhile (/=' ') . show) |
39 | wlog $ "end fork: " ++ show k | 136 | src $$ xmppInbound k pingflag src stanzas |
40 | return () | 137 | atomically $ putTMVar rdone () |
138 | wlog $ "end reader fork: " ++ show k | ||
139 | output <- atomically $ Slotted.new isEventBeginElement isEventEndElement | ||
140 | let slot_src = do | ||
141 | what <- lift . atomically $ orElse | ||
142 | (Slotted.pull output >>= \x -> return $ do | ||
143 | yield x | ||
144 | slot_src) | ||
145 | (takeTMVar rdone >> return (return ())) | ||
146 | what | ||
147 | forkIO $ do slot_src $$ snk | ||
148 | wlog $ "end writer fork: " ++ show k | ||
149 | return output | ||
41 | 150 | ||
42 | monitor sv params = do | 151 | monitor sv params = do |
43 | chan <- return $ serverEvent sv | 152 | chan <- return $ serverEvent sv |
153 | stanzas <- atomically newTChan | ||
44 | fix $ \loop -> do | 154 | fix $ \loop -> do |
45 | (k,e) <- atomically $ readTChan chan | 155 | action <- atomically $ foldr1 orElse |
46 | case e of | 156 | [ readTChan chan >>= \(k,e) -> return $ do |
47 | Connection pingflag conread conwrite -> do | 157 | case e of |
48 | let (xsrc,xsnk) = xmlStream conread conwrite | 158 | Connection pingflag conread conwrite -> do |
49 | forkConnection k pingflag xsrc xsnk | 159 | let (xsrc,xsnk) = xmlStream conread conwrite |
50 | wlog $ tomsg k "Connection" | 160 | forkConnection k pingflag xsrc xsnk stanzas |
51 | EOF -> wlog $ tomsg k "EOF" | 161 | wlog $ tomsg k "Connection" |
52 | HalfConnection In -> do | 162 | EOF -> wlog $ tomsg k "EOF" |
53 | wlog $ tomsg k "ReadOnly" | 163 | HalfConnection In -> do |
54 | control sv (Connect (callBackAddress k) params) | 164 | wlog $ tomsg k "ReadOnly" |
55 | HalfConnection Out -> wlog $ tomsg k "WriteOnly" | 165 | control sv (Connect (callBackAddress k) params) |
56 | RequiresPing -> wlog $ tomsg k "RequiresPing" | 166 | HalfConnection Out -> wlog $ tomsg k "WriteOnly" |
57 | _ -> return () | 167 | RequiresPing -> wlog $ tomsg k "RequiresPing" |
168 | _ -> return () | ||
169 | , readTChan stanzas >>= \stanza -> return $ do | ||
170 | xs <- chanContents (stanzaChan stanza) | ||
171 | prettyPrint "STANZA: " (catMaybes xs) ---- wlog $ "STANZA: "++ show (catMaybes xs) | ||
172 | ] | ||
173 | action | ||
58 | loop | 174 | loop |
59 | where | 175 | where |
60 | tomsg k str = printf "%12s %s" str (show k) | 176 | tomsg k str = printf "%12s %s" str (show k) |
177 | where | ||
178 | _ = str :: String | ||
61 | 179 | ||
62 | data ConnectionKey | 180 | data ConnectionKey |
63 | = PeerKey { callBackAddress :: SockAddr } | 181 | = PeerKey { callBackAddress :: SockAddr } |
@@ -77,11 +195,13 @@ main = runResourceT $ do | |||
77 | sv <- server | 195 | sv <- server |
78 | lift $ do | 196 | lift $ do |
79 | peer_params <- return (connectionDefaults peerKey) | 197 | peer_params <- return (connectionDefaults peerKey) |
80 | { pingInterval = 2000, duplex = False } | 198 | { pingInterval = 0 |
199 | , timeout = 0 | ||
200 | , duplex = False } | ||
81 | client_params <- return $ connectionDefaults clientKey | 201 | client_params <- return $ connectionDefaults clientKey |
82 | forkIO $ monitor sv peer_params | 202 | forkIO $ monitor sv peer_params |
83 | control sv (Listen peerport peer_params) | 203 | control sv (Listen peerport peer_params) |
84 | control sv (Listen clientport client_params) | 204 | -- control sv (Listen clientport client_params) |
85 | 205 | ||
86 | -- atomically $ newEmptyTMVar >>= readTMVar -- Wait for control-c | 206 | -- atomically $ newEmptyTMVar >>= readTMVar -- Wait for control-c |
87 | quitVar <- newEmptyTMVarIO | 207 | quitVar <- newEmptyTMVarIO |