packages feed

amqp-worker (empty) → 0.2.0

raw patch · 11 files changed

+617/−0 lines, 11 filesdep +aesondep +amqpdep +amqp-workersetup-changed

Dependencies added: aeson, amqp, amqp-worker, base, bytestring, data-default, exceptions, monad-control, mtl, resource-pool, split, tasty, tasty-hunit, text, transformers-base

Files

+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2016, Sean Hess++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of Sean Hess nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ amqp-worker.cabal view
@@ -0,0 +1,65 @@+-- Initial amqp-worker.cabal generated by cabal init.  For further+-- documentation, see http://haskell.org/cabal/users-guide/++name:                amqp-worker+version:             0.2.0+synopsis:            High level functions for working with message queue+description:         High level functions for working with message queue+license:             BSD3+license-file:        LICENSE+author:              Sean Hess+maintainer:          seanhess@gmail.com+-- copyright:+category:            Web+build-type:          Simple+-- extra-source-files:+cabal-version:       >=1.10++library+  -- other-extensions:+  exposed-modules:+    Network.AMQP.Worker+  other-modules:+      Network.AMQP.Worker.Connection+    , Network.AMQP.Worker.Key+    , Network.AMQP.Worker.Message+    , Network.AMQP.Worker.Poll+    , Network.AMQP.Worker.Queue+    , Network.AMQP.Worker.Worker+  hs-source-dirs:      src+  default-language:    Haskell2010+  build-depends:+    base >=4.9 && <5+    , aeson+    , amqp+    , bytestring+    , exceptions+    , data-default+    , monad-control+    , mtl+    , resource-pool+    , split+    , text+    , transformers-base+++test-suite test+  type:                exitcode-stdio-1.0+  hs-source-dirs:      test+  main-is:             Spec.hs+  build-depends:+      base+    , bytestring+    , aeson+    , amqp+    , amqp-worker+    , tasty+    , tasty-hunit+    , text+  ghc-options:         -threaded -rtsopts -with-rtsopts=-N+  default-language:    Haskell2010+++source-repository head+  type:     git+  location: https://github.com/seanhess/amqp-worker
+ src/Network/AMQP/Worker.hs view
@@ -0,0 +1,126 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleContexts #-}++-- |+--+-- High level functions for working with message queues. Built on top of Network.AMQP. See https://hackage.haskell.org/package/amqp, which only works with RabbitMQ: https://www.rabbitmq.com/+--+-- /Example/:+--+-- Connect to a server, initialize a queue, publish a message, and create a worker to process them.+--+-- >+-- > {-# LANGUAGE OverloadedStrings #-}+-- > {-# LANGUAGE DeriveGeneric #-}+-- > module Main where+-- >+-- > import Control.Monad.Catch (SomeException)+-- > import Data.Aeson (FromJSON, ToJSON)+-- > import Data.Text (Text)+-- > import GHC.Generics (Generic)+-- > import qualified Network.AMQP.Worker as Worker+-- > import Network.AMQP.Worker (fromURI, Exchange, Queue, Direct, def, WorkerException, Message(..), Connection)+-- >+-- > data TestMessage = TestMessage+-- >  { greeting :: Text }+-- >  deriving (Generic, Show, Eq)+-- >+-- > instance FromJSON TestMessage+-- > instance ToJSON TestMessage+-- >+-- >+-- > exchange :: Exchange+-- > exchange = Worker.exchange "testExchange"+-- >+-- >+-- > queue :: Queue Direct TestMessage+-- > queue = Worker.queue exchange "testQueue"+-- >+-- >+-- > results :: Queue Direct Text+-- > results = Worker.queue exchange "resultQueue"+-- >+-- >+-- > example :: IO ()+-- > example = do+-- >   -- connect+-- >   conn <- Worker.connect (fromURI "amqp://guest:guest@localhost:5672")+-- >+-- >   -- initialize the queues+-- >   Worker.initQueue conn queue+-- >   Worker.initQueue conn results+-- >+-- >   -- publish a message+-- >   Worker.publish conn queue (TestMessage "hello world")+-- >+-- >   -- create a worker, the program loops here+-- >   Worker.worker def conn queue onError (onMessage conn)+-- >+-- >+-- > onMessage :: Connection -> Message TestMessage -> IO ()+-- > onMessage conn m = do+-- >   let testMessage = value m+-- >   putStrLn "Got Message"+-- >   print testMessage+-- >   Worker.publish conn results (greeting testMessage)+-- >+-- >+-- > onError :: WorkerException SomeException -> IO ()+-- > onError e = do+-- >   putStrLn "Do something with errors"+-- >   print e++module Network.AMQP.Worker+  (+  -- * Declaring Queues and Exchanges+    exchange+  , ExchangeName+  , queue+  , topicQueue+  , Exchange(..)+  , Queue(..)+  , Direct, Topic++  -- * Connecting+  , Connection+  , connect+  , disconnect+  , AMQP.fromURI++  -- * Initializing exchanges and queues+  , initQueue++  -- * Publishing Messages+  , publish+  , publishToExchange++  -- * Reading Messages+  , consume+  , consumeNext+  , ConsumeResult(..)+  , ParseError(..)+  , Message(..)++  -- * Worker+  , worker+  , WorkerException(..)+  , WorkerOptions(..)+  , Microseconds+  , Default.def++  -- * Routing Keys+  , RoutingKey(..)+  , BindingKey(..)+  , BindingName(..)+  , QueueKey(..)++  ) where++import qualified Data.Default as Default+import qualified Network.AMQP as AMQP++import Network.AMQP.Worker.Key+import Network.AMQP.Worker.Connection+import Network.AMQP.Worker.Queue+import Network.AMQP.Worker.Message+import Network.AMQP.Worker.Worker
+ src/Network/AMQP/Worker/Connection.hs view
@@ -0,0 +1,55 @@+{-# LANGUAGE FlexibleContexts #-}+module Network.AMQP.Worker.Connection+  ( Connection+  , connect+  , disconnect+  , withChannel+  ) where++import Data.Pool (Pool)+import qualified Data.Pool as Pool+import qualified Network.AMQP as AMQP+import Network.AMQP (Channel)+import Control.Monad.Trans.Control (MonadBaseControl)++data ConnResource =+  ConnResource AMQP.Connection Channel+++newtype Connection =+  Connection (Pool ConnResource)++-- | Connect to the AMQP server. This returns a connection pool. It will automatically re-open the connect if an exception occurs+--+-- > conn <- connect (fromURI "amqp://guest:guest@localhost:5672")+--+connect :: AMQP.ConnectionOpts -> IO Connection+connect opts = do+    p <- Pool.createPool create destroy numStripes openTime numResources+    pure $ Connection p+  where+    numStripes = 1+    openTime = 10+    numResources = 4++    create = do+      conn <- AMQP.openConnection'' opts+      chan <- AMQP.openChannel conn+      return $ ConnResource conn chan++    destroy (ConnResource conn _) =+      AMQP.closeConnection conn+++disconnect :: Connection -> IO ()+disconnect (Connection p) =+    Pool.destroyAllResources p++++withChannel :: MonadBaseControl IO m => Connection -> (Channel -> m b) -> m b+withChannel (Connection p) action =+    Pool.withResource p chanAction+  where+    chanAction (ConnResource _ chan) =+      action chan
+ src/Network/AMQP/Worker/Key.hs view
@@ -0,0 +1,62 @@+{-# LANGUAGE OverloadedStrings #-}+module Network.AMQP.Worker.Key+  ( QueueKey(..)+  , RoutingKey(..)+  , BindingKey(..)+  , BindingName(..)+  ) where++import qualified Data.List as List+import qualified Data.List.Split as List+import Data.String (IsString(..))+import Data.Text (Text)+import qualified Data.Text as Text++-- | A name used to address queues+newtype RoutingKey = RoutingKey Text+  deriving (Show, Eq)++instance IsString RoutingKey where+  fromString = RoutingKey . Text.pack++instance QueueKey RoutingKey where+  showKey (RoutingKey t) = t+++-- | A dynamic binding address for topic queues+--+-- > commentsKey :: BindingKey+-- > commentsKey = "posts.*.comments"+newtype BindingKey = BindingKey [BindingName]+  deriving (Eq)++instance QueueKey BindingKey where+  showKey (BindingKey ns) =+    Text.intercalate "." . List.map bindingNameText $ ns++instance IsString BindingKey where+  fromString s =+    let segments = List.splitOn "." s+        names = List.map fromString segments+    in BindingKey names+++data BindingName+  = Name Text+  | Star+  | Hash+  deriving (Eq)++instance IsString BindingName where+  fromString "*" = Star+  fromString "#" = Hash+  fromString n = Name (Text.pack n)++bindingNameText :: BindingName -> Text+bindingNameText (Name t) = t+bindingNameText Star = "*"+bindingNameText Hash = "#"+++class QueueKey key where+  showKey :: key -> Text
+ src/Network/AMQP/Worker/Message.hs view
@@ -0,0 +1,101 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleContexts #-}+module Network.AMQP.Worker.Message where++import Control.Monad.Trans.Control (MonadBaseControl)+import Control.Monad.Base (liftBase)+import Data.Aeson (ToJSON, FromJSON)+import qualified Data.Aeson as Aeson+import Data.ByteString.Lazy (ByteString)+import Network.AMQP (newMsg, DeliveryMode(..), Ack(..), QueueOpts(..))+import qualified Network.AMQP as AMQP++import Network.AMQP.Worker.Key (RoutingKey(..))+import Network.AMQP.Worker.Poll (poll)+import Network.AMQP.Worker.Connection (Connection, withChannel)+import Network.AMQP.Worker.Queue (Queue(..), Exchange(..), ExchangeName, Direct)++-- types --------------------------++-- | a parsed message from the queue+data Message a = Message+  { body :: ByteString+  , value :: a+  } deriving (Show, Eq)+++data ConsumeResult a+  = Parsed (Message a)+  | Error ParseError++data ParseError = ParseError String ByteString++type Microseconds = Int+++jsonMessage :: ToJSON a => a -> AMQP.Message+jsonMessage a = newMsg+  { AMQP.msgBody = Aeson.encode a+  , AMQP.msgContentType = Just "application/json"+  , AMQP.msgContentEncoding = Just "UTF-8"+  , AMQP.msgDeliveryMode = Just Persistent+  }++++-- | publish a message to a routing key, without making sure a queue exists to handle it or if it is the right type of message body+--+-- > publishToExchange conn "users.admin.created" (User "username")+publishToExchange :: (ToJSON a, MonadBaseControl IO m) => Connection -> ExchangeName -> RoutingKey -> a -> m ()+publishToExchange conn exg (RoutingKey rk) msg =+  withChannel conn $ \chan -> do+    _ <- liftBase $ AMQP.publishMsg chan exg rk (jsonMessage msg)+    return ()+++-- | publish a message to a queue. Enforces that the message type and queue name are correct at the type level+--+-- > let queue = Worker.queue exchange "users" :: Queue Direct User+-- > publish conn queue (User "username")+publish :: (ToJSON msg, MonadBaseControl IO m) => Connection -> Queue Direct msg -> msg -> m ()+publish conn (Queue (Exchange exg) key _) =+  publishToExchange conn (AMQP.exchangeName exg) key+++-- | Check for a message once and attempt to parse it+--+-- > res <- consume conn queue+-- > case res of+-- >   Just (Parsed m) -> print m+-- >   Just (Error e) -> putStrLn "could not parse message"+-- >   Notihng -> putStrLn "No messages on the queue"+consume :: (FromJSON msg, MonadBaseControl IO m) => Connection -> Queue key msg -> m (Maybe (ConsumeResult msg))+consume conn (Queue _ _ options) = do+  mme <- withChannel conn $ \chan ->+    liftBase $ AMQP.getMsg chan Ack (queueName options)++  case mme of+    Nothing ->+      return Nothing++    Just (msg, env) -> do+      liftBase $ AMQP.ackEnv env+      let bd = AMQP.msgBody msg+      case Aeson.eitherDecode bd of+        Left err ->+          return $ Just $ Error (ParseError err bd)++        Right v ->+          return $ Just $ Parsed (Message bd v)++++-- | Block while checking for messages every N microseconds. Return once you find one.+--+-- > res <- consumeNext conn queue+-- > case res of+-- >   (Parsed m) -> print m+-- >   (Error e) -> putStrLn "could not parse message"+consumeNext :: (FromJSON msg, MonadBaseControl IO m) => Microseconds -> Connection -> Queue key msg -> m (ConsumeResult msg)+consumeNext pd conn queue =+    poll pd $ consume conn queue
+ src/Network/AMQP/Worker/Poll.hs view
@@ -0,0 +1,16 @@+{-# LANGUAGE FlexibleContexts #-}++module Network.AMQP.Worker.Poll where++import Control.Concurrent (threadDelay)+import Control.Monad.Trans.Control (MonadBaseControl)+import Control.Monad.Base (liftBase)++poll :: (MonadBaseControl IO m) => Int -> m (Maybe a) -> m a+poll us action = do+    ma <- action+    case ma of+      Just a -> return a+      Nothing -> do+        liftBase $ threadDelay us+        poll us action
+ src/Network/AMQP/Worker/Queue.hs view
@@ -0,0 +1,74 @@+{-# LANGUAGE OverloadedStrings #-}+module Network.AMQP.Worker.Queue where++import Data.Text (Text)+import qualified Network.AMQP as AMQP+import Network.AMQP (ExchangeOpts(..), QueueOpts(..))++import Network.AMQP.Worker.Key (RoutingKey(..), BindingKey(..), QueueKey(..))+import Network.AMQP.Worker.Connection (Connection, withChannel)+++type ExchangeName = Text+++-- | Declare an exchange+--+-- In AMQP, exchanges can be fanout, direct, or topic. This library attempts to simplify this choice by making all exchanges be topic exchanges, and allowing the user to specify topic or direct behavior on the queue itself. See @queue@+--+-- > exchange :: Exchange+-- > exchange = Worker.exchange "testExchange"++exchange :: ExchangeName -> Exchange+exchange nm =+  Exchange $ AMQP.newExchange { exchangeName = nm, exchangeType = "topic" }++-- | Declare a direct queue with the name @RoutingKey@. Direct queues work as you expect: you can publish messages to them and read from them.+--+-- > queue :: Queue Direct MyMessageType+-- > queue = Worker.queue exchange "testQueue"++queue :: Exchange -> RoutingKey -> Queue Direct msg+queue exg (RoutingKey key) =+  Queue exg (RoutingKey key) $ AMQP.newQueue { queueName = key }++-- | Declare a topic queue. Topic queues allow you to bind a queue to a dynamic address with wildcards+--+-- > queue :: Queue Topic MyMessageType+-- > queue = Worker.topicQueue exchange "new-users.*"++topicQueue :: Exchange -> BindingKey -> Queue Topic msg+topicQueue exg key =+  Queue exg key $ AMQP.newQueue { queueName = showKey key }+++data Exchange =+  Exchange AMQP.ExchangeOpts+  deriving (Show, Eq)++-- | Queues consist of an exchange, a type (Direct or Topic), and a message type+--+-- > queue :: Queue Direct MyMessageType+data Queue queueType msg =+  Queue Exchange queueType AMQP.QueueOpts+  deriving (Show, Eq)++type Direct = RoutingKey++type Topic = BindingKey++++-- | Register a queue and its exchange with the AMQP server. Call this before publishing or reading+--+-- > let queue = Worker.queue exchange "my-queue" :: Queue Direct Text+-- > conn <- Worker.connect (fromURI "amqp://guest:guest@localhost:5672")+-- > Worker.initQueue conn queue++initQueue :: (QueueKey key) => Connection -> Queue key msg -> IO ()+initQueue conn (Queue (Exchange exg) key options) =+  withChannel conn $ \chan -> do+    _ <- AMQP.declareExchange chan exg+    _ <- AMQP.declareQueue chan options+    _ <- AMQP.bindQueue chan (AMQP.queueName options) (AMQP.exchangeName exg) (showKey key)+    return ()
+ src/Network/AMQP/Worker/Worker.hs view
@@ -0,0 +1,70 @@+{-# LANGUAGE FlexibleContexts #-}+module Network.AMQP.Worker.Worker where++import Control.Concurrent (threadDelay)+import Control.Exception (SomeException(..))+import Control.Monad.Catch (Exception(..), catch, MonadCatch)+import Control.Monad.Trans.Control (MonadBaseControl)+import Control.Monad (forever)+import Data.Aeson (FromJSON)+import Data.ByteString.Lazy (ByteString)+import Data.Default (Default(..))+import Control.Monad.Base (liftBase)++import Network.AMQP.Worker.Connection (Connection)+import Network.AMQP.Worker.Queue (Queue(..))+import Network.AMQP.Worker.Message (Message(..), ConsumeResult(..), ParseError(..), Microseconds, consumeNext)++++-- | Create a worker which loops, checks for messages, and handles errors+--+-- > startWorker conn queue = do+-- >   Worker.worker def conn queue onError onMessage+-- >+-- >   where+-- >     onMessage :: Message User+-- >     onMessage m = do+-- >       putStrLn "handle user message"+-- >       print (value m)+-- >+-- >     onError :: WorkerException SomeException -> IO ()+-- >     onError e = do+-- >       putStrLn "Do something with errors"++worker :: (FromJSON a, MonadBaseControl IO m, MonadCatch m) => WorkerOptions -> Connection -> Queue key a -> (WorkerException SomeException -> m ()) -> (Message a -> m ()) -> m ()+worker opts conn queue onError action =+  forever $ do+    eres <- consumeNext (pollDelay opts) conn queue+    case eres of+      Error (ParseError reason bd) ->+        onError (MessageParseError bd reason)++      Parsed msg ->+        catch+          (action msg)+          (onError . OtherException (body msg))+    liftBase $ threadDelay (loopDelay opts)+++++-- | Options for worker+data WorkerOptions = WorkerOptions+  { pollDelay :: Microseconds -- ^ Delay between checks to consume. Defaults to 10ms+  , loopDelay :: Microseconds -- ^ Delay between calls to job. Defaults to 0+  } deriving (Show, Eq)++instance Default WorkerOptions where+  def = WorkerOptions+    { pollDelay = 10 * 1000+    , loopDelay = 0+    }++-- | Exceptions created while processing+data WorkerException e+  = MessageParseError ByteString String+  | OtherException ByteString e+  deriving (Show, Eq)++instance (Exception e) => Exception (WorkerException e)
+ test/Spec.hs view
@@ -0,0 +1,16 @@+module Main where++import Test.Tasty+import Test.Tasty.HUnit++main :: IO ()+main = do+  defaultMain tests+++tests :: TestTree+tests = testGroup "Tests"+  [ testCase "TODO" testExample ]++testExample :: Assertion+testExample = pure ()