@@ -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 ()