diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,22 @@ +Copyright (c) 2013-2016 SaleMove Inc. + +MIT License + +Permission is hereby granted, free of charge, to any person obtaining +a copy of this software and associated documentation files (the +"Software"), to deal in the Software without restriction, including +without limitation the rights to use, copy, modify, merge, publish, +distribute, sublicense, and/or sell copies of the Software, and to +permit persons to whom the Software is furnished to do so, subject to +the following conditions: + +The above copyright notice and this permission notice shall be +included in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE +LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION +WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,96 @@ +# RabbitMQ Messaging API supporting request-response + +[![Build Status](https://travis-ci.org/salemove/freddy-hs.svg?branch=master)](https://travis-ci.org/salemove/freddy-hs) +[![Code Climate](https://codeclimate.com/github/salemove/freddy-hs/badges/gpa.svg)](https://codeclimate.com/github/salemove/freddy-hs) + +## Setup + +* Create a connection: + +```haskell +import qualified Network.Freddy as Freddy +import qualified Network.Freddy.Request as R + +connection <- Freddy.connect "127.0.0.1" "/" "guest" "guest" +``` + +## Delivering messages + +### Simple delivery + +#### Send and forget +Sends a `message` to the given `destination`. If there is no consumer then the +message stays in the queue until somebody consumes it. +```haskell + Freddy.deliver connection R.newReq { + R.queueName = "notifications.user_signed_in", + R.body = "{\"user_id\": 1}" + } +``` + +#### Expiring messages +Sends a `message` to the given `destination`. If nobody consumes the message in +`timeoutInMs` milliseconds then the message is discarded. This is useful for +showing notifications that must happen in a certain timeframe but guaranteed +delivery is not a strict requirement. +```haskell + Freddy.deliver connection R.newReq { + R.queueName = "notifications.user_signed_in", + R.body = "{\"user_id\": 1}", + R.timeoutIsMs = 5000 + } +``` + +### Request delivery +Sends a `message` to the given `destination`. Has a default timeout of 3 +seconds and discards the message from the queue if a response hasn't been +returned in that time. +```haskell + response <- Freddy.deliverWithResponse connection R.newReq { + R.queueName = "echo", + R.body = "{\"msg\": \"what did you say?\"}" + } + + case response of + Right payload -> putStrLn "Received positive result" + Left (Freddy.InvalidRequest payload) -> putStrLn "Received error" + Left Freddy.TimeoutError -> putStrLn "Request timed out" +``` + +## Responding to messages +```haskell + processMessage (Freddy.Delivery body replyWith failWith) = replyWith body + + connection <- Freddy.connect "127.0.0.1" "/" "guest" "guest" + Freddy.respondTo connection "echo" processMessage +``` + +## Tapping into messages +Listens for messages on a given destination or destinations without +consuming them. + +```haskell + processMessage body = putStrLn body + + connection <- Freddy.connect "127.0.0.1" "/" "guest" "guest" + Freddy.tapInto connection "notifications.*" processMessage +``` + +* Note that it is not possible to respond to the message while tapping. +* When tapping the following wildcards are supported in the `queueName`: + * `#` matching 0 or more words + * `*` matching exactly one word + +Examples: + +```haskell +Freddy.tapInto connection "i.#.free" processMessage +``` + +receives messages that are delivered to `"i.want.to.break.free"` + +```haskell +Freddy.tapInto connection "somebody.*.love" processMessage +``` + +receives messages that are delivered to `somebody.to.love` but doesn't receive messages delivered to `someboy.not.to.love` diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/freddy.cabal b/freddy.cabal new file mode 100644 --- /dev/null +++ b/freddy.cabal @@ -0,0 +1,47 @@ +name: freddy +version: 0.1.0.0 +homepage: https://github.com/salemove/freddy-hs +license: MIT +license-file: LICENSE +author: Indrek Juhkam +maintainer: indrek@urgas.eu +category: Network +build-type: Simple +cabal-version: >=1.10 +extra-source-files: README.md +synopsis: RabbitMQ Messaging API supporting request-response + +library + exposed-modules: Network.Freddy + build-depends: + base >= 4.8 && < 5, + bytestring >= 0.9, + text >= 0.11.2, + amqp >= 0.13.1, + broadcast-chan >= 0.1.0, + uuid >= 1.3.11, + random >= 1.1, + data-default >= 0.6.0 + hs-source-dirs: src + default-language: Haskell2010 + +source-repository head + type: git + location: https://github.com/salemove/freddy-hs + +test-suite spec + type: exitcode-stdio-1.0 + hs-source-dirs: src, test + main-is: Spec.hs + build-depends: + hspec >= 1.3, + base >= 4.8 && < 5, + bytestring >= 0.9, + text >= 0.11.2, + amqp >= 0.13.1, + broadcast-chan >= 0.1.0, + uuid >= 1.3.11, + random >= 1.1, + data-default >= 0.6.0, + async >= 2.1.0 + default-language: Haskell2010 diff --git a/src/Network/Freddy.hs b/src/Network/Freddy.hs new file mode 100644 --- /dev/null +++ b/src/Network/Freddy.hs @@ -0,0 +1,325 @@ +{-# LANGUAGE OverloadedStrings #-} +module Network.Freddy ( + connect, + disconnect, + Connection, + respondTo, + tapInto, + deliverWithResponse, + deliver, + cancelConsumer, + Consumer, + Delivery (..), + Error (..) +) where + +import Control.Concurrent (forkIO) +import qualified Network.AMQP as AMQP +import Data.Text (Text, pack) +import Data.ByteString.Lazy.Char8 (ByteString) +import qualified Control.Concurrent.BroadcastChan as BC +import System.Timeout (timeout) +import Network.Freddy.ResultType (ResultType) +import qualified Network.Freddy.ResultType as ResultType +import qualified Network.Freddy.Request as Request +import Network.Freddy.CorrelationIdGenerator (CorrelationId, generateCorrelationId) + +type Payload = ByteString +type QueueName = Text + +type ReplyWith = Payload -> IO () +type FailWith = Payload -> IO () + +type ResponseChannelEmitter = BC.BroadcastChan BC.In (Maybe AMQP.PublishError, AMQP.Message) +type ResponseChannelListener = BC.BroadcastChan BC.Out (Maybe AMQP.PublishError, AMQP.Message) + +data Error = InvalidRequest Payload | TimeoutError deriving (Show, Eq) +type Response = Either Error Payload + +data Delivery = Delivery Payload ReplyWith FailWith +data Reply = Reply QueueName AMQP.Message + +data Connection = Connection { + amqpConnection :: AMQP.Connection, + amqpProduceChannel :: AMQP.Channel, + amqpResponseChannel :: AMQP.Channel, + responseQueueName :: Text, + eventChannel :: ResponseChannelEmitter +} + +data Consumer = Consumer { + consumerTag :: AMQP.ConsumerTag, + consumerChannel :: AMQP.Channel +} + +{-| + Creates a connection with the message queue. + + __Example__: + + @ + connection <- Freddy.connect "127.0.0.1" "/" "guest" "guest" + @ +-} +connect :: String -- ^ server host + -> Text -- ^ virtual host + -> Text -- ^ user name + -> Text -- ^ password + -> IO Connection +connect host vhost user pass = do + connection <- AMQP.openConnection host vhost user pass + produceChannel <- AMQP.openChannel connection + responseChannel <- AMQP.openChannel connection + + AMQP.declareExchange produceChannel AMQP.newExchange { + AMQP.exchangeName = topicExchange, + AMQP.exchangeType = "topic", + AMQP.exchangeDurable = False + } + + eventChannel <- BC.newBroadcastChan + + (responseQueueName, _, _) <- declareQueue responseChannel "" + AMQP.consumeMsgs responseChannel responseQueueName AMQP.NoAck $ responseCallback eventChannel + + AMQP.addReturnListener produceChannel (returnCallback eventChannel) + + return $ Connection { + amqpConnection = connection, + amqpResponseChannel = responseChannel, + amqpProduceChannel = produceChannel, + responseQueueName = responseQueueName, + eventChannel = eventChannel + } + +{-| + Closes the connection with the message queue. + + __Example__: + + @ + connection <- Freddy.connect "127.0.0.1" "/" "guest" "guest" + Freddy.disconnect connection + @ +-} +disconnect :: Connection -> IO () +disconnect = AMQP.closeConnection . amqpConnection + +{-| + Sends a message and waits for the response. + + __Example__: + + @ + import qualified Network.Freddy as Freddy + import qualified Network.Freddy.Request as R + + connection <- Freddy.connect "127.0.0.1" "/" "guest" "guest" + + response <- Freddy.deliverWithResponse connection R.newReq { + R.queueName = "echo", + R.body = "{\\"msg\\": \\"what did you say?\\"}" + } + + case response of + Right payload -> putStrLn "Received positive result" + Left (Freddy.InvalidRequest payload) -> putStrLn "Received error" + Left Freddy.TimeoutError -> putStrLn "Request timed out" + @ +-} +deliverWithResponse :: Connection -> Request.Request -> IO Response +deliverWithResponse connection request = do + correlationId <- generateCorrelationId + + let msg = AMQP.newMsg { + AMQP.msgBody = Request.body request, + AMQP.msgCorrelationID = Just correlationId, + AMQP.msgDeliveryMode = Just AMQP.NonPersistent, + AMQP.msgType = Just "request", + AMQP.msgReplyTo = Just $ responseQueueName connection, + AMQP.msgExpiration = Request.expirationInMs request + } + + responseChannelListener <- BC.newBChanListener $ eventChannel connection + + AMQP.publishMsg' (amqpProduceChannel connection) "" (Request.queueName request) True msg + AMQP.publishMsg (amqpProduceChannel connection) topicExchange (Request.queueName request) msg + + responseBody <- timeout (Request.timeoutInMicroseconds request) $ do + let messageMatcher = matchingCorrelationId correlationId + waitForResponse responseChannelListener messageMatcher + + case responseBody of + Just (Nothing, msg) -> return . createResponse $ msg + Just (Just error, _) -> return . Left . InvalidRequest $ "Publish Error" + Nothing -> return $ Left TimeoutError + +createResponse :: AMQP.Message -> Either Error Payload +createResponse msg = do + let msgBody = AMQP.msgBody msg + + case AMQP.msgType msg of + Just msgType -> + if (ResultType.fromText msgType) == ResultType.Success then + Right msgBody + else + Left . InvalidRequest $ msgBody + _ -> Left . InvalidRequest $ "No message type" + +{-| + Send and forget type of delivery. It sends a message to given destination + without waiting for a response. This is useful when there are multiple + consumers that are using 'tapInto' or you just do not care about the + response. + + __Example__: + + @ + import qualified Network.Freddy as Freddy + import qualified Network.Freddy.Request as R + + connection <- Freddy.connect "127.0.0.1" "/" "guest" "guest" + Freddy.deliver connection R.newReq { + R.queueName = "notifications.user_signed_in", + R.body = "{\\"user_id\\": 1}" + } + @ +-} +deliver :: Connection -> Request.Request -> IO () +deliver connection request = do + let msg = AMQP.newMsg { + AMQP.msgBody = Request.body request, + AMQP.msgDeliveryMode = Just AMQP.NonPersistent, + AMQP.msgExpiration = Request.expirationInMs request + } + + AMQP.publishMsg (amqpProduceChannel connection) "" (Request.queueName request) msg + AMQP.publishMsg (amqpProduceChannel connection) topicExchange (Request.queueName request) msg + +{-| + Responds to messages on a given destination. It is useful for messages that + have to be processed once and then a result must be sent. + + __Example__: + + @ + processMessage (Freddy.Delivery body replyWith failWith) = replyWith body + + connection <- Freddy.connect "127.0.0.1" "/" "guest" "guest" + Freddy.respondTo connection "echo" processMessage + @ +-} +respondTo :: Connection -> QueueName -> (Delivery -> IO ()) -> IO Consumer +respondTo connection queueName callback = do + let produceChannel = amqpProduceChannel connection + consumeChannel <- AMQP.openChannel . amqpConnection $ connection + declareQueue consumeChannel queueName + tag <- AMQP.consumeMsgs consumeChannel queueName AMQP.NoAck $ + replyCallback callback produceChannel + return Consumer { consumerChannel = consumeChannel, consumerTag = tag } + +{-| + Listens for messages on a given destination or destinations without + consuming them. + + __Example__: + + @ + processMessage body = putStrLn body + + connection <- Freddy.connect "127.0.0.1" "/" "guest" "guest" + Freddy.tapInto connection "notifications.*" processMessage + @ +-} +tapInto :: Connection -> QueueName -> (Payload -> IO ()) -> IO Consumer +tapInto connection queueName callback = do + consumeChannel <- AMQP.openChannel . amqpConnection $ connection + + declareExlusiveQueue consumeChannel queueName + AMQP.bindQueue consumeChannel "" topicExchange queueName + + let consumer = callback . AMQP.msgBody .fst + tag <- AMQP.consumeMsgs consumeChannel queueName AMQP.NoAck consumer + + return Consumer { consumerChannel = consumeChannel, consumerTag = tag } + +{-| + Stops the consumer from listening new messages. + + __Example__: + + @ + connection <- Freddy.connect "127.0.0.1" "/" "guest" "guest" + consumer <- Freddy.tapInto connection "notifications.*" processMessage + threadDelay tenMinutes $ Freddy.cancelConsumer consumer + @ +-} +cancelConsumer :: Consumer -> IO () +cancelConsumer consumer = do + AMQP.cancelConsumer (consumerChannel consumer) $ consumerTag consumer + AMQP.closeChannel (consumerChannel consumer) + +returnCallback :: ResponseChannelEmitter -> (AMQP.Message, AMQP.PublishError) -> IO () +returnCallback eventChannel (msg, error) = + BC.writeBChan eventChannel (Just error, msg) + +responseCallback :: ResponseChannelEmitter -> (AMQP.Message, AMQP.Envelope) -> IO () +responseCallback eventChannel (msg, _) = + BC.writeBChan eventChannel (Nothing, msg) + +replyCallback :: (Delivery -> IO ()) -> AMQP.Channel -> (AMQP.Message, AMQP.Envelope) -> IO () +replyCallback userCallback channel (msg, env) = do + let requestBody = AMQP.msgBody msg + let replyWith = sendReply msg channel ResultType.Success + let failWith = sendReply msg channel ResultType.Error + let delivery = Delivery requestBody replyWith failWith + forkIO . userCallback $ delivery + return () + +sendReply :: AMQP.Message -> AMQP.Channel -> ResultType -> Payload -> IO () +sendReply originalMsg channel resType body = + case buildReply originalMsg resType body of + Just (Reply queueName message) -> AMQP.publishMsg channel "" queueName message + Nothing -> putStrLn "Could not reply" + +buildReply :: AMQP.Message -> ResultType -> Payload -> Maybe Reply +buildReply originalMsg resType body = do + queueName <- AMQP.msgReplyTo originalMsg + + let msg = AMQP.newMsg { + AMQP.msgBody = body, + AMQP.msgCorrelationID = AMQP.msgCorrelationID originalMsg, + AMQP.msgDeliveryMode = Just AMQP.NonPersistent, + AMQP.msgType = Just . ResultType.serializeResultType $ resType + } + + Just $ Reply queueName msg + +matchingCorrelationId :: CorrelationId -> AMQP.Message -> Bool +matchingCorrelationId correlationId msg = + case AMQP.msgCorrelationID msg of + Just msgCorrelationId -> msgCorrelationId == correlationId + Nothing -> False + +topicExchange :: Text +topicExchange = "freddy-topic" + +waitForResponse :: ResponseChannelListener -> (AMQP.Message -> Bool) -> IO (Maybe AMQP.PublishError, AMQP.Message) +waitForResponse eventChannelListener predicate = do + (error, msg) <- BC.readBChan eventChannelListener + + if predicate msg then + return (error, msg) + else + waitForResponse eventChannelListener predicate + +declareQueue :: AMQP.Channel -> QueueName -> IO (Text, Int, Int) +declareQueue channel queueName = + AMQP.declareQueue channel AMQP.newQueue {AMQP.queueName = queueName} + +declareExlusiveQueue :: AMQP.Channel -> QueueName -> IO (Text, Int, Int) +declareExlusiveQueue channel queueName = + AMQP.declareQueue channel AMQP.newQueue { + AMQP.queueName = queueName, + AMQP.queueExclusive = True + } diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -0,0 +1,1 @@ +{-# OPTIONS_GHC -F -pgmF hspec-discover #-}