amqp-worker 0.2.2 → 0.2.3

raw patch · 2 files changed

+37/−12 lines, 2 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• amqp-worker.cabal +1/−1
• src/Network/AMQP/Worker/Connection.hs +36/−11

@@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/  name:                amqp-worker-version:             0.2.2+version:             0.2.3 synopsis:            High level functions for working with message queues description:         High level functions for working with message queues license:             BSD3

@@ -6,48 +6,73 @@   , withChannel   ) where +import Control.Concurrent.MVar (MVar, putMVar, newEmptyMVar, readMVar, takeMVar)+import Control.Monad.Catch (throwM, catch) import Data.Pool (Pool) import qualified Data.Pool as Pool import qualified Network.AMQP as AMQP-import Network.AMQP (Channel)+import Network.AMQP (Channel, AMQPException(..)) import Control.Monad.Trans.Control (MonadBaseControl)  data Connection =-  Connection AMQP.Connection (Pool Channel)+  Connection (MVar AMQP.Connection) (Pool Channel) --- | Connect to the AMQP server. This returns a connection pool which will automatically re-open the connection as needed if an exception occurs.+-- | Connect to the AMQP server. -- -- > conn <- connect (fromURI "amqp://guest:guest@localhost:5672") -- connect :: AMQP.ConnectionOpts -> IO Connection connect opts = do -    -- open one connection-    conn <- AMQP.openConnection'' opts+    -- create a single connection in an mvar+    cvar <- newEmptyMVar+    openConnection cvar      -- open a shared pool for channels-    chans <- Pool.createPool (create conn) destroy numStripes openTime numChans+    chans <- Pool.createPool (create cvar) destroy numStripes openTime numChans -    pure $ Connection conn chans++    pure $ Connection cvar chans   where     numStripes = 1     openTime = 10     numChans = 4 -    create conn =-      AMQP.openChannel conn+    openConnection cvar = do+      -- open a connection and store in the mvar+      conn <- AMQP.openConnection'' opts+      putMVar cvar conn -    destroy chan =+    reopenConnection cvar = do+      -- clear the mvar and reopen+      _ <- takeMVar cvar+      openConnection cvar+++    create cvar = do+      conn <- readMVar cvar+      chan <- catch (AMQP.openChannel conn) (createEx cvar)+      return chan++    createEx cvar (ConnectionClosedException _) = do+      reopenConnection cvar+      create cvar++    createEx _ ex = throwM ex++    destroy chan = do       AMQP.closeChannel chan   disconnect :: Connection -> IO () disconnect (Connection c p) = do+    conn <- readMVar c     Pool.destroyAllResources p-    AMQP.closeConnection c+    AMQP.closeConnection conn    withChannel :: MonadBaseControl IO m => Connection -> (Channel -> m b) -> m b withChannel (Connection _ p) action = do     Pool.withResource p action+