freddy (empty) → 0.1.0.0

raw patch · 6 files changed

+493/−0 lines, 6 filesdep +amqpdep +asyncdep +basesetup-changed

Dependencies added: amqp, async, base, broadcast-chan, bytestring, data-default, hspec, random, text, uuid

Files

• LICENSE +22/−0
• README.md +96/−0
• Setup.hs +2/−0
• freddy.cabal +47/−0
• src/Network/Freddy.hs +325/−0
• test/Spec.hs +1/−0

@@ -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.

@@ -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`

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

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

@@ -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+  }

@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}