git.iko.soy/typeable/wai
on master help sign in
First import
334c9edMathias Biilmann Christensen
eventsource-wai.cabalnew file+34
compare with master
unifiedsplit
@@ -0,0 +1,34 @@
1+Name: eventsource-wai
2+Version: 0.1
3+Synopsis: Project Synopsis Here
4+Description: Project Description Here
5+License: AllRightsReserved
6+Author: Author
7+Maintainer: maintainer@example.com
8+Stability: Experimental
9+Category: Web
10+Build-type: Simple
11+Cabal-version: >=1.2
12+
13+Executable eventsource-wai
14+ hs-source-dirs: src
15+ main-is: Main.hs
16+
17+ Build-depends:
18+ base >= 4.3 && < 5,
19+ bytestring >= 0.9.1 && < 0.10,
20+ blaze-builder >= 0.3,
21+ MonadCatchIO-transformers >= 0.2.1 && < 0.3,
22+ mtl >= 2 && < 3,
23+ http-types >= 0.6,
24+ enumerator >= 0.4,
25+ time >= 1.2,
26+ wai >= 0.4,
27+ warp >= 0.4
28+
29+ if impl(ghc >= 6.12.0)
30+ ghc-options: -threaded -Wall -fwarn-tabs -funbox-strict-fields -O2
31+ -fno-warn-unused-do-bind
32+ else
33+ ghc-options: -threaded -Wall -fwarn-tabs -funbox-strict-fields -O2
34+
src/AMQPListener.hsnew file+80
@@ -0,0 +1,80 @@
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/EventStream.hsnew file+82
@@ -0,0 +1,82 @@
1+{-# LANGUAGE OverloadedStrings #-}
2+
3+{-|
4+ A Snap adapter to the HTML5 Server-Sent Events API. Push-mode and
5+ pull-mode interfaces are both available.
6+-}
7+module EventStream (
8+ ServerEvent(..),
9+ eventToBuilder
10+ ) where
11+
12+import Blaze.ByteString.Builder
13+import Blaze.ByteString.Builder.Char8
14+import Control.Monad.Trans
15+import Control.Concurrent
16+import Data.Monoid
17+import Data.Enumerator hiding (map)
18+import Data.Enumerator.List (generateM)
19+
20+{-|
21+ Type representing a communication over an event stream. This can be an
22+ actual event, a comment, a modification to the retry timer, or a special
23+ "close" event indicating the server should close the connection.
24+-}
25+data ServerEvent
26+ = ServerEvent {
27+ eventName :: Maybe Builder,
28+ eventId :: Maybe Builder,
29+ eventData :: [Builder]
30+ }
31+ | CommentEvent {
32+ eventComment :: Builder
33+ }
34+ | RetryEvent {
35+ eventRetry :: Int
36+ }
37+ | CloseEvent
38+
39+
40+{-|
41+ Newline as a Builder.
42+-}
43+nl = fromChar '\n'
44+
45+
46+{-|
47+ Field names as Builder
48+-}
49+nameField = fromString "event:"
50+idField = fromString "id:"
51+dataField = fromString "data:"
52+retryField = fromString "retry:"
53+commentField = fromChar ':'
54+
55+
56+{-|
57+ Wraps the text as a labeled field of an event stream.
58+-}
59+field l b = l `mappend` b `mappend` nl
60+
61+
62+{-|
63+ Appends a buffer flush to the end of a Builder.
64+-}
65+flushAfter b = b `mappend` flush
66+
67+
68+{-|
69+ Converts a 'ServerEvent' to its wire representation as specified by the
70+ @text/event-stream@ content type.
71+-}
72+eventToBuilder :: ServerEvent -> Maybe Builder
73+eventToBuilder (CommentEvent txt) = Just $ flushAfter $ field commentField txt
74+eventToBuilder (RetryEvent n) = Just $ flushAfter $ field retryField (fromShow n)
75+eventToBuilder (CloseEvent) = Nothing
76+eventToBuilder (ServerEvent n i d)= Just $ flushAfter $
77+ (name n $ evid i $ mconcat (map (field dataField) d)) `mappend` nl
78+ where
79+ name Nothing = id
80+ name (Just n) = mappend (field nameField n)
81+ evid Nothing = id
82+ evid (Just i) = mappend (field idField i)
src/Main.hsnew file+43
@@ -0,0 +1,43 @@
1+{-# LANGUAGE OverloadedStrings #-}
2+
3+module Main where
4+
5+import Control.Monad
6+import Control.Monad.Trans
7+import Control.Concurrent (forkIO, threadDelay)
8+import Control.Concurrent.Chan
9+import Network.Wai.Handler.Warp (run)
10+import Network.Wai
11+import Network.HTTP.Types (statusOK)
12+import Data.Enumerator (Iteratee, Step(Continue), Stream(Chunks,EOF), returnI, run_, (>>==), ($$), generateM)
13+import Data.Time.Clock.POSIX
14+import qualified Data.ByteString as B
15+import Blaze.ByteString.Builder.Char.Utf8 (fromString)
16+
17+
18+import EventStream
19+
20+
21+app :: Chan ServerEvent -> Application
22+app chan = return $ res chan
23+
24+res :: Chan ServerEvent -> Response
25+res chan = ResponseEnumerator (resE chan)
26+
27+source :: Chan ServerEvent -> IO ()
28+source chan = forever $ do
29+ threadDelay 1000000
30+ time <- getPOSIXTime
31+ writeChan chan (ServerEvent Nothing Nothing [fromString . show $ time])
32+
33+resE :: Chan ServerEvent -> ResponseEnumerator a
34+resE chan genIter =
35+ run_ $ generateM (fmap eventToBuilder $ readChan chan) $$ iter
36+ where
37+ iter = genIter statusOK [("Content-Type", "text/event-stream")]
38+
39+main :: IO ()
40+main = do
41+ chan <- newChan
42+ forkIO . source $ chan
43+ run 8000 (app chan)