amqp 0.9 → 0.10

raw patch · 7 files changed

+73/−61 lines, 7 filessetup-changedPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• LICENSE +27/−27
• Network/AMQP/ChannelAllocator.hs +18/−9
• Network/AMQP/Internal.hs +9/−19
• Network/AMQP/Types.hs +14/−1
• Setup.lhs +3/−3
• amqp.cabal +1/−1
• test/ChannelSpec.hs +1/−1

@@ -1,27 +1,27 @@-Copyright (c) 2011, Holger Reinhardt
-
-All rights reserved.
-
-Redistribution and use in source and binary forms, with or without
-modification, are permitted provided that the following conditions
-are met:
-1. Redistributions of source code must retain the above copyright
-   notice, this list of conditions and the following disclaimer.
-2. 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.
-3. Neither the name of the author nor the names of his contributors
-   may be used to endorse or promote products derived from this software
-   without specific prior written permission.
-
-THIS SOFTWARE IS PROVIDED BY THE REGENTS 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 AUTHORS 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.
+Copyright (c) 2011, Holger Reinhardt++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions+are met:+1. Redistributions of source code must retain the above copyright+   notice, this list of conditions and the following disclaimer.+2. 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.+3. Neither the name of the author nor the names of his contributors+   may be used to endorse or promote products derived from this software+   without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE REGENTS 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 AUTHORS 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.

@@ -1,28 +1,37 @@ module Network.AMQP.ChannelAllocator where
 
 import qualified Data.Vector.Mutable as V
+import Control.Exception (throwIO)
 import Data.Word
 import Data.Bits
 
-newtype ChannelAllocator = ChannelAllocator (V.IOVector Word64)
+import Network.AMQP.Types
 
-newChannelAllocator :: IO ChannelAllocator
-newChannelAllocator =
-    fmap ChannelAllocator $ V.replicate 1024 0
+data ChannelAllocator = ChannelAllocator Int -- highest permitted channel id
+                                         (V.IOVector Word64)
 
+
+newChannelAllocator :: Int -> IO ChannelAllocator
+newChannelAllocator maxChannel =
+    fmap (ChannelAllocator maxChannel) $ V.replicate 1024 0
+
 allocateChannel :: ChannelAllocator -> IO Int
-allocateChannel (ChannelAllocator c) = do
+allocateChannel (ChannelAllocator maxChannel c) = do
     maybeIx <- findFreeIndex c
     case maybeIx of
         Just chunk -> do
             word <- V.read c chunk
             let offset = findUnsetBit word
-            V.write c chunk (setBit word offset)
-            return (chunk*64 + offset)
-        Nothing -> error "all channels are already allocated"
+            let channelID = chunk*64 + offset
+            if channelID > maxChannel
+                then throwIO $ AllChannelsAllocatedException maxChannel
+                else do
+                    V.write c chunk (setBit word offset)
+                    return channelID
+        Nothing -> throwIO $ AllChannelsAllocatedException maxChannel
 
 freeChannel :: ChannelAllocator -> Int -> IO Bool
-freeChannel (ChannelAllocator c) ix = do
+freeChannel (ChannelAllocator _maxChannel c) ix = do
     let (chunk, offset) = divMod ix 64
     word <- V.read c chunk
     if testBit word offset

@@ -1,4 +1,4 @@-{-# LANGUAGE BangPatterns, CPP, DeriveDataTypeable, OverloadedStrings, ScopedTypeVariables #-}
+{-# LANGUAGE BangPatterns, CPP, OverloadedStrings, ScopedTypeVariables #-}
 module Network.AMQP.Internal where
 
 import Control.Concurrent
@@ -9,7 +9,6 @@ import Data.Int (Int64)
 import Data.Maybe
 import Data.Text (Text)
-import Data.Typeable
 import Network
 
 import qualified Control.Exception as CE
@@ -141,8 +140,6 @@ 
 -- | Represents the parameters to connect to a broker or a cluster of brokers.
 -- See 'defaultConnectionOpts'.
---
--- /NOTICE/: The field 'coMaxChannel' was only added for future use, as the respective functionality is not yet implemented.
 data ConnectionOpts = ConnectionOpts {
                             coServers :: ![(String, PortNumber)], -- ^ A list of host-port pairs. Useful in a clustered setup to connect to the first available host.
                             coVHost :: !Text, -- ^ The VHost to connect to.
@@ -199,7 +196,7 @@ openConnection'' :: ConnectionOpts -> IO Connection
 openConnection'' connOpts = withSocketsDo $ do
     handle <- connect $ coServers connOpts
-    (maxFrameSize, heartbeatTimeout) <- CE.handle (\(_ :: CE.IOException) -> CE.throwIO $ ConnectionClosedException "Handshake failed. Please check the RabbitMQ logs for more information") $ do
+    (maxFrameSize, maxChannel, heartbeatTimeout) <- CE.handle (\(_ :: CE.IOException) -> CE.throwIO $ ConnectionClosedException "Handshake failed. Please check the RabbitMQ logs for more information") $ do
         Conn.connectionPut handle $ BS.append (BC.pack "AMQP")
                 (BS.pack [
                           1
@@ -215,15 +212,17 @@         -- C: start_ok
         writeFrame handle $ start_ok selectedSASL
         -- S: secure or tune
-        Frame 0 (MethodPayload (Connection_tune _ frame_max sendHeartbeat)) <- handleSecureUntilTune handle selectedSASL
+        Frame 0 (MethodPayload (Connection_tune channel_max frame_max sendHeartbeat)) <- handleSecureUntilTune handle selectedSASL
         -- C: tune_ok
         let maxFrameSize = chooseMin frame_max $ coMaxFrameSize connOpts
             finalHeartbeatSec = fromMaybe sendHeartbeat (coHeartbeatDelay connOpts)
             heartbeatTimeout = mfilter (/=0) (Just finalHeartbeatSec)
 
+            fixChanNum x = if x == 0 then 65535 else x
+            maxChannel = chooseMin (fixChanNum channel_max) $ fmap fixChanNum $ coMaxChannel connOpts
+
         writeFrame handle (Frame 0 (MethodPayload
-            --TODO: handle channel_max
-            (Connection_tune_ok 0 maxFrameSize finalHeartbeatSec)
+            (Connection_tune_ok maxChannel maxFrameSize finalHeartbeatSec)
             ))
         -- C: open
         writeFrame handle open
@@ -232,12 +231,12 @@         Frame 0 (MethodPayload (Connection_open_ok _)) <- readFrame handle
 
         -- Connection established!
-        return (maxFrameSize, heartbeatTimeout)
+        return (maxFrameSize, maxChannel, heartbeatTimeout)
 
     --build Connection object
     cChannels <- newMVar IM.empty
     cClosed <- newMVar Nothing
-    cChanAllocator <- newChannelAllocator
+    cChanAllocator <- newChannelAllocator $ fromIntegral maxChannel
     _ <- allocateChannel cChanAllocator -- allocate channel 0
     writeLock <- newMVar ()
     ccl <- newEmptyMVar
@@ -677,12 +676,3 @@             case chc of
                 Just r -> CE.throwIO $ ChannelClosedException r
                 Nothing -> CE.throwIO $ ConnectionClosedException "unknown reason"
-
------------------------------ EXCEPTIONS ---------------------------
-
-data AMQPException =
-    -- | the 'String' contains the reason why the channel was closed
-    ChannelClosedException String
-    | ConnectionClosedException String -- ^ String may contain a reason
-  deriving (Typeable, Show, Ord, Eq)
-instance CE.Exception AMQPException

@@ -1,8 +1,9 @@-{-# LANGUAGE BangPatterns #-}
+{-# LANGUAGE BangPatterns, DeriveDataTypeable #-}
 -- |
 --
 -- This module contains data-types specified in the AMQP spec
 module Network.AMQP.Types (
+    AMQPException(..),
     Octet,
     Bit,
     ChannelID,
@@ -27,11 +28,23 @@ import Data.Binary.Put
 import Data.Char
 import Data.Text (Text)
+import Data.Typeable
 
+import qualified Control.Exception as CE
 import qualified Data.ByteString as BS
 import qualified Data.ByteString.Lazy as BL
 import qualified Data.Map as M
 import qualified Data.Text.Encoding as T
+
+
+data AMQPException =
+    -- | the 'String' contains the reason why the channel was closed
+    ChannelClosedException String
+    | ConnectionClosedException String -- ^ String may contain a reason
+    | AllChannelsAllocatedException Int -- ^ the 'Int' contains the channel-max property of the connection (i.e. the highest permitted channel id)
+  deriving (Typeable, Show, Ord, Eq)
+instance CE.Exception AMQPException
+
 
 -- performs runGet on a bytestring until the string is empty
 readMany :: (Show a, Binary a) => BL.ByteString -> [a]

@@ -1,3 +1,3 @@-#!/usr/bin/env runhaskell
-> import Distribution.Simple
-> main = defaultMain
+#!/usr/bin/env runhaskell+> import Distribution.Simple+> main = defaultMain

@@ -1,5 +1,5 @@ Name:                amqp
-Version:             0.9
+Version:             0.10
 Synopsis:            Client library for AMQP servers (currently only RabbitMQ)
 Description:         Client library for AMQP servers (currently only RabbitMQ)
                      .

@@ -39,6 +39,6 @@                 conn <- openConnection "127.0.0.1" "/" "guest" "guest"
                 ch   <- openChannel conn
 
-                qos ch 0 5
+                qos ch 0 5 False
 
                 closeConnection conn