git.iko.soy/typeable/wai
on master help sign in
Basic EventSource prototype working
b860311Mathias Biilmann Christensen
src/AMQPListener.hsdeleted−80
compare with master
unifiedsplit
@@ -1,80 +0,0 @@
1−{-# LANGUAGE OverloadedStrings #-}
2−module AMQPListener
3− (
4− AMQPEvent(..)
5− , openEventChannel
6− ) where
7−
8−import Control.Applicative((<$>), (<*>))
9−import Control.Monad(mzero)
10−import Control.Monad.Fix(fix)
11−import Control.Concurrent(forkIO)
12−import Control.Concurrent.Chan(Chan, newChan, readChan, writeChan)
13−
14−import Data.Aeson(FromJSON(..), Value(..), Result(..), fromJSON, json, (.:), (.:?))
15−import Data.Attoparsec(parse, maybeResult)
16−
17−import qualified Data.ByteString as B
18−import qualified Data.ByteString.Lazy as LB
19−
20−import Data.Maybe(fromJust, fromMaybe)
21−import Data.String.Utils(split)
22−
23−import Text.URI(URI(..), parseURI)
24−import System.Posix.Env(getEnvDefault)
25−import Network.AMQP
26−
27−-- |The AMQPEvent represents and incomming message that should be
28−-- mapped to an EventSource event.
29−data AMQPEvent = AMQPEvent
30− { amqpChannel :: B.ByteString
31− , amqpData :: B.ByteString
32− , amqpId :: Maybe B.ByteString
33− , amqpName :: Maybe B.ByteString
34− }
35−
36−instance FromJSON AMQPEvent where
37− parseJSON (Object v) = AMQPEvent <$>
38− v .: "channel" <*>
39− v .: "data" <*>
40− v .:? "id" <*>
41− v .:? "name"
42− parseJSON _ = mzero
43−
44−-- |Connects to an AMQP broker.
45−-- Tries to get credentials, host and vhost from the AMQP_URL
46−-- environment variable
47−-- Take an exchange name and a queue name
48−openEventChannel :: String -> String -> IO (Chan AMQPEvent)
49−openEventChannel exchange queue = do
50− amqpURI <- getEnvDefault "AMQP_URL" "amqp://guest:guest@127.0.0.1/"
51−
52− let uri = fromJust $ parseURI amqpURI
53− let auth = fromMaybe "guest:guest" $ uriUserInfo uri
54− let host = fromMaybe "127.0.0.1" $ uriRegName uri
55− let vhost = uriPath uri
56−
57− let [user,password] = split ":" auth
58−
59− conn <- openConnection host vhost user password
60− chan <- openChannel conn
61−
62− declareQueue chan newQueue {queueName = queue, queueAutoDelete = True, queueDurable = False}
63− declareExchange chan newExchange {exchangeName = exchange, exchangeType = "fanout", exchangeDurable = False}
64− bindQueue chan queue exchange queue
65−
66− listener <- newChan
67− forkIO $ fix $ \loop -> readChan listener >> loop
68− consumeMsgs chan queue NoAck (sendTo listener)
69− return listener
70−
71−-- |Write messages from AMQP to a channel
72−sendTo :: Chan AMQPEvent -> (Message, Envelope) -> IO ()
73−sendTo chan (msg, _) =
74− case maybeResult $ parse json (B.concat $ LB.toChunks (msgBody msg)) of
75− Just value -> case fromJSON value of
76− Success event -> do
77− writeChan chan event
78− Error _ -> do
79− return ()
80− Nothing -> return ()
src/Main.hs+12−2
@@ -1,5 +1,4 @@
1{-# LANGUAGE OverloadedStrings #-}
1{-# LANGUAGE OverloadedStrings #-}
2−
3module Main where
2module Main where
4
3
5import Control.Monad
4import Control.Monad
@@ -11,7 +10,9 @@ import Network.Wai
11import Network.HTTP.Types (statusOK)
10import Network.HTTP.Types (statusOK)
12import Data.Enumerator (Iteratee, Step(Continue), Stream(Chunks,EOF), returnI, run_, (>>==), ($$), generateM)
11import Data.Enumerator (Iteratee, Step(Continue), Stream(Chunks,EOF), returnI, run_, (>>==), ($$), generateM)
13import Data.Time.Clock.POSIX
12import Data.Time.Clock.POSIX
13+import Data.ByteString (ByteString)
14import qualified Data.ByteString as B
14import qualified Data.ByteString as B
15+import Blaze.ByteString.Builder (fromByteString)
15import Blaze.ByteString.Builder.Char.Utf8 (fromString)
16import Blaze.ByteString.Builder.Char.Utf8 (fromString)
16
17
17
18
@@ -19,23 +20,32 @@ import EventStream
19
20
20
21
21app :: Chan ServerEvent -> Application
22app :: Chan ServerEvent -> Application
23+app chan req =
24+ case pathInfo req of
25+ [] -> return $ ResponseFile statusOK [("Content-Type", "text/html")] "static/index.html" Nothing
26+ ["es"] -> do
27+ chan' <- liftIO $ dupChan chan
22−app chan = return $ res chan
28+ return $ res chan'
29+
23
30
24res :: Chan ServerEvent -> Response
31res :: Chan ServerEvent -> Response
25res chan = ResponseEnumerator (resE chan)
32res chan = ResponseEnumerator (resE chan)
26
33
34+
27source :: Chan ServerEvent -> IO ()
35source :: Chan ServerEvent -> IO ()
28source chan = forever $ do
36source chan = forever $ do
29 threadDelay 1000000
37 threadDelay 1000000
30 time <- getPOSIXTime
38 time <- getPOSIXTime
31 writeChan chan (ServerEvent Nothing Nothing [fromString . show $ time])
39 writeChan chan (ServerEvent Nothing Nothing [fromString . show $ time])
32
40
41+
33resE :: Chan ServerEvent -> ResponseEnumerator a
42resE :: Chan ServerEvent -> ResponseEnumerator a
34resE chan genIter =
43resE chan genIter =
35 run_ $ generateM (fmap eventToBuilder $ readChan chan) $$ iter
44 run_ $ generateM (fmap eventToBuilder $ readChan chan) $$ iter
36 where
45 where
37 iter = genIter statusOK [("Content-Type", "text/event-stream")]
46 iter = genIter statusOK [("Content-Type", "text/event-stream")]
38
47
48+
39main :: IO ()
49main :: IO ()
40main = do
50main = do
41 chan <- newChan
51 chan <- newChan
static/index.htmlnew file+19
@@ -0,0 +1,19 @@
1+<!doctype html>
2+<head>
3+ <title>Hello</title>
4+</head>
5+<h1>Testing</h1>
6+
7+<div id="output"></div>
8+
9+<script>
10+ var output = document.getElementById("output");
11+
12+ var src = new EventSource("/es");
13+
14+ src.onmessage = function(msg) {
15+ var p = document.createElement("p");
16+ p.appendChild(document.createTextNode(msg.data));
17+ output.appendChild(p);
18+ }
19+</script>