amqp-worker 0.2.0 → 0.2.1

raw patch · 5 files changed

+37/−22 lines, 5 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• amqp-worker.cabal +3/−3
• src/Network/AMQP/Worker/Connection.hs +29/−16
• src/Network/AMQP/Worker/Message.hs +3/−2
• src/Network/AMQP/Worker/Worker.hs +1/−1
• test/Spec.hs +1/−0

@@ -2,9 +2,9 @@ -- 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+version:             0.2.1+synopsis:            High level functions for working with message queues+description:         High level functions for working with message queues license:             BSD3 license-file:        LICENSE author:              Sean Hess

@@ -6,38 +6,54 @@   , withChannel   ) where +import Control.Concurrent.MVar (readMVar, newEmptyMVar, putMVar) 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)+  Connection (Pool Channel) --- | Connect to the AMQP server. This returns a connection pool. It will automatically re-open the connect if an exception occurs+-- | Connect to the AMQP server. This returns a connection pool which will automatically re-open the connection as needed 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++    chansVar <- newEmptyMVar++    -- keep one connection open+    conns <- Pool.createPool createConn (destroyConn chansVar) 1 connOpenTime 1++    -- the channels use the pool to create themselves+    chans <- Pool.createPool (create conns) destroy numStripes openTime numResources++    putMVar chansVar chans++    pure $ Connection chans   where     numStripes = 1     openTime = 10     numResources = 4+    connOpenTime = 60 -    create = do+    create connPool = do+      Pool.withResource connPool $ AMQP.openChannel++    destroy chan = do+      AMQP.closeChannel chan++    createConn = do       conn <- AMQP.openConnection'' opts-      chan <- AMQP.openChannel conn-      return $ ConnResource conn chan+      pure conn -    destroy (ConnResource conn _) =+    destroyConn chans conn = do+      -- destroy all channels or they will be stale and throw exceptions+      chanPool <- readMVar chans+      Pool.destroyAllResources chanPool       AMQP.closeConnection conn  @@ -49,7 +65,4 @@  withChannel :: MonadBaseControl IO m => Connection -> (Channel -> m b) -> m b withChannel (Connection p) action =-    Pool.withResource p chanAction-  where-    chanAction (ConnResource _ chan) =-      action chan+    Pool.withResource p action

@@ -71,8 +71,9 @@ -- >   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)+  mme <- withChannel conn $ \chan -> do+    m <- liftBase $ AMQP.getMsg chan Ack (queueName options)+    pure m    case mme of     Nothing ->

@@ -3,13 +3,13 @@  import Control.Concurrent (threadDelay) import Control.Exception (SomeException(..))+import Control.Monad.Base (liftBase) 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(..))

@@ -12,5 +12,6 @@ tests = testGroup "Tests"   [ testCase "TODO" testExample ] + testExample :: Assertion testExample = pure ()