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