amqp 0.5.0 → 0.6.0

raw patch · 5 files changed

+688/−503 lines, 5 filesdep ~textPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: text

API changes (from Hackage documentation)

Files

• Network/AMQP.hs +68/−490
• Network/AMQP/Helpers.hs +5/−1
• Network/AMQP/Internal.hs +601/−0
• Network/AMQP/Types.hs +11/−9
• amqp.cabal +3/−3

@@ -1,9 +1,9 @@ {-# LANGUAGE BangPatterns, DeriveDataTypeable, OverloadedStrings, ScopedTypeVariables #-}
 -- |
 --
--- A client library for AMQP servers implementing the 0-8 spec; currently only supports RabbitMQ (see <http://www.rabbitmq.com>)
+-- A client library for AMQP servers implementing the 0-9-1 spec; currently only supports RabbitMQ (see <http://www.rabbitmq.com>)
 --
--- A good introduction to AMQP can be found here (though it uses Python): <http://blogs.digitar.com/jjww/2009/01/rabbits-and-warrens/>
+-- A good introduction to RabbitMQ and AMQP 0-9-1 (in various languages): <http://www.rabbitmq.com/getstarted.html>, <http://www.rabbitmq.com/tutorials/amqp-concepts.html>
 --
 -- /Example/:
 --
@@ -48,14 +48,18 @@ module Network.AMQP (
     -- * Connection
     Connection,
+    ConnectionOpts(..),
+    defaultConnectionOpts,
     openConnection,
     openConnection',
+    openConnection'',
     closeConnection,
     addConnectionClosedHandler,
 
     -- * Channel
     Channel,
     openChannel,
+    addReturnListener,
     qos,
 
     -- * Exchanges
@@ -82,6 +86,8 @@     -- * Messaging
     Message(..),
     DeliveryMode(..),
+    PublishError(..),
+    ReturnReplyCode(..),
     newMsg,
     Envelope(..),
     ConsumerTag,
@@ -90,6 +96,7 @@     consumeMsgs',
     cancelConsumer,
     publishMsg,
+    publishMsg',
     getMsg,
     rejectMsg,
     recoverMsgs,
@@ -105,40 +112,31 @@     -- * Flow Control
     flow,
 
+    -- * SASL
+    SASLMechanism(..),
+    plain,
+    amqplain,
+    rabbitCRdemo,
+
     -- * Exceptions
     AMQPException(..)
 ) where
 
 import Control.Concurrent
-import Control.Monad
 import Data.Binary
-import Data.Binary.Get
-import Data.Binary.Put as BPut
-import Data.Maybe
-import Data.Text (Text)
-import Data.Typeable
+import Data.Binary.Put
 import Network
-import System.IO
+import Data.Text (Text)
 
-import qualified Control.Exception as CE
-import qualified Data.ByteString.Char8 as BS
-import qualified Data.ByteString.Lazy.Char8 as BL
+import qualified Data.ByteString.Lazy as BL
+import qualified Data.ByteString as BS
+import qualified Data.Text.Encoding as E
 import qualified Data.Map as M
-import qualified Data.Foldable as F
-import qualified Data.IntMap as IM
-import qualified Data.Sequence as Seq
 import qualified Data.Text as T
 
-import Network.AMQP.Protocol
 import Network.AMQP.Types
-import Network.AMQP.Helpers
 import Network.AMQP.Generated
-
-{-
-TODO:
-- handle basic.return
-- connection.secure
--}
+import Network.AMQP.Internal
 
 ----- EXCHANGE -----
 
@@ -317,6 +315,10 @@ 
 ----- MSG (the BASIC class in AMQP) -----
 
+-- | a 'Msg' with defaults set; you should override at least 'msgBody'
+newMsg :: Message
+newMsg = Message (BL.empty) Nothing Nothing Nothing Nothing Nothing Nothing Nothing
+
 type ConsumerTag = Text
 
 -- | specifies whether you have to acknowledge messages that you receive from 'consumeMsgs' or 'getMsg'. If you use 'Ack', you have to call 'ackMsg' or 'ackEnv' after you have processed a message, otherwise it might be delivered again in the future
@@ -367,17 +369,21 @@     --unregister the consumer
     modifyMVar_ (consumers chan) $ return . M.delete consumerTag
 
--- | @publishMsg chan exchange routingKey msg@ publishes @msg@ to the exchange with the provided @exchange@. The effect of @routingKey@ depends on the type of the exchange
+-- | @publishMsg chan exchange routingKey msg@ publishes @msg@ to the exchange with the provided @exchange@. The effect of @routingKey@ depends on the type of the exchange.
 --
--- NOTE: This method may temporarily block if the AMQP server requested us to stop sending content data (using the flow control mechanism). So don't rely on this method returning immediately
+-- NOTE: This method may temporarily block if the AMQP server requested us to stop sending content data (using the flow control mechanism). So don't rely on this method returning immediately.
 publishMsg :: Channel -> Text -> Text -> Message -> IO ()
-publishMsg chan exchange routingKey msg = do
+publishMsg chan exchange routingKey msg = publishMsg' chan exchange routingKey False msg
+
+-- | Like 'publishMsg', but additionally allows you to specify whether the 'mandatory' flag should be set.
+publishMsg' :: Channel -> Text -> Text -> Bool -> Message -> IO ()
+publishMsg' chan exchange routingKey mandatory msg = do
     writeAssembly chan (ContentMethod (Basic_publish
             1 -- ticket; ignored by rabbitMQ
             (ShortString exchange)
             (ShortString routingKey)
-            False -- mandatory; if true, the server might return the msg, which is currently not handled
-            False) --immediate; if true, the server might return the msg, which is currently not handled
+            mandatory -- mandatory; if true, the server might return the msg, which is currently not handled
+            False) --immediate; not customizable, as it is currently not supported anymore by RabbitMQ
 
             --TODO: add more of these to 'Message'
             (CHBasic
@@ -488,477 +494,58 @@     (SimpleMethod (Channel_flow_ok _)) <- request chan $ SimpleMethod (Channel_flow active)
     return ()
 
--------------------------- MESSAGE / ENVELOPE ------------------
-
--- | contains meta-information of a delivered message (through 'getMsg' or 'consumeMsgs')
-data Envelope = Envelope
-              {
-                envDeliveryTag :: LongLongInt,
-                envRedelivered :: Bool,
-                envExchangeName :: Text,
-                envRoutingKey :: Text,
-                envChannel :: Channel
-              }
-
-data DeliveryMode = Persistent -- ^ the message will survive server restarts (if the queue is durable)
-                  | NonPersistent -- ^ the message may be lost after server restarts
-    deriving (Eq, Ord, Read, Show)
-
-deliveryModeToInt :: DeliveryMode -> Octet
-deliveryModeToInt NonPersistent = 1
-deliveryModeToInt Persistent = 2
-
-intToDeliveryMode :: Octet -> DeliveryMode
-intToDeliveryMode 1 = NonPersistent
-intToDeliveryMode 2 = Persistent
-intToDeliveryMode n = error ("Unknown delivery mode int: " ++ show n)
-
--- | An AMQP message
-data Message = Message {
-                msgBody :: BL.ByteString, -- ^ the content of your message
-                msgDeliveryMode :: Maybe DeliveryMode, -- ^ see 'DeliveryMode'
-                msgTimestamp :: Maybe Timestamp, -- ^ use in any way you like; this doesn't affect the way the message is handled
-                msgID :: Maybe Text, -- ^ use in any way you like; this doesn't affect the way the message is handled
-                msgContentType :: Maybe Text,
-                msgReplyTo :: Maybe Text,
-                msgCorrelationID :: Maybe Text,
-                msgHeaders :: Maybe FieldTable
-                }
-    deriving (Eq, Ord, Read, Show)
-
--- | a 'Msg' with defaults set; you should override at least 'msgBody'
-newMsg :: Message
-newMsg = Message (BL.empty) Nothing Nothing Nothing Nothing Nothing Nothing Nothing
-
-------------- ASSEMBLY -------------------------
--- an assembly is a higher-level object consisting of several frames (like in amqp 0-10)
-data Assembly = SimpleMethod MethodPayload
-              | ContentMethod MethodPayload ContentHeaderProperties BL.ByteString --method, properties, content-data
-    deriving Show
-
--- | reads all frames necessary to build an assembly
-readAssembly :: Chan FramePayload -> IO Assembly
-readAssembly chan = do
-    m <- readChan chan
-    case m of
-        MethodPayload p -> --got a method frame
-            if hasContent m
-                then do
-                    --several frames containing the content will follow, so read them
-                    (props, msg) <- collectContent chan
-                    return $ ContentMethod p props msg
-                else do
-                    return $ SimpleMethod p
-        x -> error $ "didn't expect frame: " ++ show x
-
--- | reads a contentheader and contentbodies and assembles them
-collectContent :: Chan FramePayload -> IO (ContentHeaderProperties, BL.ByteString)
-collectContent chan = do
-    (ContentHeaderPayload _ _ bodySize props) <- readChan chan
-
-    content <- collect $ fromIntegral bodySize
-    return (props, BL.concat content)
-  where
-    collect x | x <= 0 = return []
-    collect x = do
-        (ContentBodyPayload payload) <- readChan chan
-        r <- collect (x - (BL.length payload))
-        return $ payload : r
-
------------- CONNECTION -------------------
-
-{- general concept:
-Each connection has its own thread. Each channel has its own thread.
-Connection reads data from socket and forwards it to channel. Channel processes data and forwards it to application.
-Outgoing data is written directly onto the socket.
-
-Incoming Data: Socket -> Connection-Thread -> Channel-Thread -> Application
-Outgoing Data: Application -> Socket
--}
-
-data Connection = Connection {
-                    connHandle :: Handle,
-                    connChannels :: (MVar (IM.IntMap (Channel, ThreadId))), --open channels (channelID => (Channel, ChannelThread))
-                    connMaxFrameSize :: Int, --negotiated maximum frame size
-                    connClosed :: MVar (Maybe String),
-                    connClosedLock :: MVar (), -- used by closeConnection to block until connection-close handshake is complete
-                    connWriteLock :: MVar (), -- to ensure atomic writes to the socket
-                    connClosedHandlers :: MVar [IO ()],
-                    lastChannelID :: MVar Int --for auto-incrementing the channelIDs
-                }
-
--- | reads incoming frames from socket and forwards them to the opened channels
-connectionReceiver :: Connection -> IO ()
-connectionReceiver conn = do
-    Frame chanID payload <- readFrame (connHandle conn)
-    forwardToChannel chanID payload
-    connectionReceiver conn
-  where
-    forwardToChannel 0 (MethodPayload Connection_close_ok) = do
-        modifyMVar_ (connClosed conn) $ const $ return $ Just "closed by user"
-        killThread =<< myThreadId
-    forwardToChannel 0 (MethodPayload (Connection_close _ (ShortString errorMsg) _ _)) = do
-        writeFrame (connHandle conn) $ Frame 0 $ MethodPayload Connection_close_ok
-        modifyMVar_ (connClosed conn) $ const $ return $ Just $ T.unpack errorMsg
-        killThread =<< myThreadId
-    forwardToChannel 0 payload = putStrLn $ "Got unexpected msg on channel zero: " ++ show payload
-    forwardToChannel chanID payload = do
-        --got asynchronous msg => forward to registered channel
-        withMVar (connChannels conn) $ \cs -> do
-            case IM.lookup (fromIntegral chanID) cs of
-                Just c -> writeChan (inQueue $ fst c) payload
-                Nothing -> putStrLn $ "ERROR: channel not open " ++ show chanID
+-- | Constructs default connection options with the following settings :
+--
+--     * Broker: @amqp:\/\/guest:guest\@localhost:5672\/%2F@ using the @PLAIN@ SASL mechanism
+--
+--     * max frame size: @131072@
+--
+--     * no heartbeat expected from the server
+--
+--     * no limit on the number of used channels
+--
+defaultConnectionOpts :: ConnectionOpts
+defaultConnectionOpts = ConnectionOpts [("localhost", 5672)] "/" [plain "guest" "guest"] (Just 131072) Nothing Nothing
 
 -- | @openConnection hostname virtualHost loginName loginPassword@ opens a connection to an AMQP server running on @hostname@.
 -- @virtualHost@ is used as a namespace for AMQP resources (default is \"/\"), so different applications could use multiple virtual hosts on the same AMQP server.
 --
 -- You must call 'closeConnection' before your program exits to ensure that all published messages are received by the server.
 --
+-- The @loginName@ and @loginPassword@ will be used to authenticate via the 'PLAIN' SASL mechanism.
+--
 -- NOTE: If the login name, password or virtual host are invalid, this method will throw a 'ConnectionClosedException'. The exception will not contain a reason why the connection was closed, so you'll have to find out yourself.
 openConnection :: String -> Text -> Text -> Text -> IO Connection
 openConnection host = openConnection' host 5672
 
 -- | same as 'openConnection' but allows you to specify a non-default port-number as the 2nd parameter
 openConnection' :: String -> PortNumber -> Text -> Text -> Text -> IO Connection
-openConnection' host port vhost loginName loginPassword = withSocketsDo $ do
-    handle <- connectTo host $ PortNumber port
-    BL.hPut handle $ BPut.runPut $ do
-        BPut.putByteString $ BS.pack "AMQP"
-        BPut.putWord8 1
-        BPut.putWord8 1 --TCP/IP
-        BPut.putWord8 0 --Major Version
-        BPut.putWord8 9 --Minor Version
-
-    -- S: connection.start
-    Frame 0 (MethodPayload (Connection_start _ _ _ _ _)) <- readFrame handle
-
-    -- C: start_ok
-    writeFrame handle start_ok
-    -- S: tune
-    Frame 0 (MethodPayload (Connection_tune _ frame_max _)) <- readFrame handle
-    -- C: tune_ok
-    let maxFrameSize = (min 131072 frame_max)
-
-    writeFrame handle (Frame 0 (MethodPayload
-        --TODO: handle channel_max
-        (Connection_tune_ok 0 maxFrameSize 0)
-        ))
-    -- C: open
-    writeFrame handle open
-
-    -- S: open_ok
-    Frame 0 (MethodPayload (Connection_open_ok _)) <- readFrame handle
-
-    -- Connection established!
-
-    --build Connection object
-    cChannels <- newMVar IM.empty
-    lastChanID <- newMVar 0
-    cClosed <- newMVar Nothing
-    writeLock <- newMVar ()
-    ccl <- newEmptyMVar
-    cClosedHandlers <- newMVar []
-    let conn = Connection handle cChannels (fromIntegral maxFrameSize) cClosed ccl writeLock cClosedHandlers lastChanID
-
-    --spawn the connectionReceiver
-    void $ forkIO $ CE.finally (connectionReceiver conn)
-            (do
-                -- try closing socket
-                CE.catch (hClose handle) (\(_ :: CE.SomeException) -> return ())
-
-                -- mark as closed
-                modifyMVar_ cClosed $ return . Just . maybe "closed" id
-
-                --kill all channel-threads
-                void $ withMVar cChannels $ mapM_ (\c -> killThread $ snd c) . IM.elems
-                void $ withMVar cChannels $ const $ return $ IM.empty
-
-                -- mark connection as closed, so all pending calls to 'closeConnection' can now return
-                void $ tryPutMVar ccl ()
-
-                -- notify connection-close-handlers
-                withMVar cClosedHandlers sequence
-                )
-    return conn
-  where
-    start_ok = (Frame 0 (MethodPayload (Connection_start_ok (FieldTable M.empty)
-        (ShortString "AMQPLAIN")
-        --login has to be a table without first 4 bytes
-        (LongString $ T.pack $ drop 4 $ BL.unpack $ runPut $ put $ FieldTable $ M.fromList [("LOGIN",FVString loginName), ("PASSWORD", FVString loginPassword)])
-        (ShortString "en_US")) ))
-    open = (Frame 0 (MethodPayload (Connection_open
-        (ShortString vhost)  --virtual host
-        (ShortString $ T.pack "") -- capabilities; deprecated in 0-9-1
-        True))) -- insist; deprecated in 0-9-1
-
--- | closes a connection
---
--- Make sure to call this function before your program exits to ensure that all published messages are received by the server.
-closeConnection :: Connection -> IO ()
-closeConnection c = do
-    CE.catch (
-        withMVar (connWriteLock c) $ \_ -> writeFrame (connHandle c) $ (Frame 0 (MethodPayload (Connection_close
-            --TODO: set these values
-            0 -- reply_code
-            (ShortString "") -- reply_text
-            0 -- class_id
-            0 -- method_id
-            )))
-            )
-        (\ (_ :: CE.IOException) ->
-            --do nothing if connection is already closed
-            return ()
-        )
-
-    -- wait for connection_close_ok by the server; this MVar gets filled in the CE.finally handler in openConnection'
-    readMVar $ connClosedLock c
-    return ()
-
--- | @addConnectionClosedHandler conn ifClosed handler@ adds a @handler@ that will be called after the connection is closed (either by calling @closeConnection@ or by an exception). If the @ifClosed@ parameter is True and the connection is already closed, the handler will be called immediately. If @ifClosed == False@ and the connection is already closed, the handler will never be called
-addConnectionClosedHandler :: Connection -> Bool -> IO () -> IO ()
-addConnectionClosedHandler conn ifClosed handler = do
-    withMVar (connClosed conn) $ \cc ->
-        case cc of
-            -- connection is already closed, so call the handler directly
-            Just _ | ifClosed == True -> handler
-
-            -- otherwise add it to the list
-            _ -> modifyMVar_ (connClosedHandlers conn) $ \old -> return $ handler:old
-
-readFrame :: Handle -> IO Frame
-readFrame handle = do
-    dat <- BL.hGet handle 7
-    let len = fromIntegral $ peekFrameSize dat
-    dat' <- BL.hGet handle (len+1) -- +1 for the terminating 0xCE
-    let ret = runGetOrFail get (BL.append dat dat')
-    case ret of
-        Left (_, _, errMsg) -> error $ "readFrame fail: " ++ errMsg
-        Right (_, consumedBytes, _) | consumedBytes /= fromIntegral (len+8) ->
-            error $ "readFrame: parser should read " ++ show (len+8) ++ " bytes; but read " ++ show consumedBytes
-        Right (_, _, frame) -> return frame
-
-writeFrame :: Handle -> Frame -> IO ()
-writeFrame handle = BL.hPut handle . runPut . put
-
------------------------- CHANNEL -----------------------------
-
-{- | A connection to an AMQP server is made up of separate channels. It is recommended to use a separate channel for each thread in your application that talks to the AMQP server (but you don't have to as channels are thread-safe)
--}
-data Channel = Channel {
-                    connection :: Connection,
-                    inQueue :: Chan FramePayload, --incoming frames (from Connection)
-                    outstandingResponses :: MVar (Seq.Seq (MVar Assembly)), -- for every request an MVar is stored here waiting for the response
-                    channelID :: Word16,
-                    lastConsumerTag :: MVar Int,
-
-                    chanActive :: Lock, -- used for flow-control. if lock is closed, no content methods will be sent
-                    chanClosed :: MVar (Maybe String),
-                    consumers :: MVar (M.Map Text ((Message, Envelope) -> IO ())) -- who is consumer of a queue? (consumerTag => callback)
-                }
-
-msgFromContentHeaderProperties :: ContentHeaderProperties -> BL.ByteString -> Message
-msgFromContentHeaderProperties (CHBasic content_type _ headers delivery_mode _ correlation_id reply_to _ message_id timestamp _ _ _ _) body =
-    let msgId = fromShortString message_id
-        contentType = fromShortString content_type
-        replyTo = fromShortString reply_to
-        correlationID = fromShortString correlation_id
-    in Message body (fmap intToDeliveryMode delivery_mode) timestamp msgId contentType replyTo correlationID headers
-  where
-    fromShortString (Just (ShortString s)) = Just s
-    fromShortString _ = Nothing
-msgFromContentHeaderProperties c _ = error ("Unknown content header properties: " ++ show c)
-
--- | The thread that is run for every channel
-channelReceiver :: Channel -> IO ()
-channelReceiver chan = do
-    --read incoming frames; they are put there by a Connection thread
-    p <- readAssembly $ inQueue chan
-    if isResponse p
-        then do
-            action <- modifyMVar (outstandingResponses chan) $ \val -> do
-                        case Seq.viewl val of
-                            x Seq.:< rest -> do
-                                return (rest, putMVar x p)
-                            Seq.EmptyL -> do
-                                return (val, CE.throwIO $ userError "got response, but have no corresponding request")
-            action
+openConnection' host port vhost loginName loginPassword = openConnection'' $ defaultConnectionOpts {
+    coServers = [(host, port)],
+    coVHost   = vhost,
+    coAuth    = [plain loginName loginPassword]
+}
 
-        --handle asynchronous assemblies
-        else handleAsync p
-    channelReceiver chan
+-- | The @PLAIN@ SASL mechanism. See <http://tools.ietf.org/html/rfc4616 RFC4616>
+plain :: Text -> Text -> SASLMechanism
+plain loginName loginPassword = SASLMechanism "PLAIN" initialResponse Nothing
   where
-    isResponse :: Assembly -> Bool
-    isResponse (ContentMethod (Basic_deliver _ _ _ _ _) _ _) = False
-    isResponse (ContentMethod (Basic_return _ _ _ _) _ _) = False
-    isResponse (SimpleMethod (Channel_flow _)) = False
-    isResponse (SimpleMethod (Channel_close _ _ _ _)) = False
-    isResponse _ = True
-
-    --Basic.Deliver: forward msg to registered consumer
-    handleAsync (ContentMethod (Basic_deliver (ShortString consumerTag) deliveryTag redelivered (ShortString exchange)
-                                                (ShortString routingKey))
-                                properties body) =
-        withMVar (consumers chan) (\s -> do
-            case M.lookup consumerTag s of
-                Just subscriber -> do
-                    let msg = msgFromContentHeaderProperties properties body
-                    let env = Envelope {envDeliveryTag = deliveryTag, envRedelivered = redelivered,
-                                    envExchangeName = exchange, envRoutingKey = routingKey, envChannel = chan}
-
-                    CE.catch (subscriber (msg, env))
-                        (\(e::CE.SomeException) -> putStrLn $ "AMQP callback threw exception: " ++ show e)
-                Nothing ->
-                    -- got a message, but have no registered subscriber; so drop it
-                    return ()
-            )
-    handleAsync (SimpleMethod (Channel_close _ (ShortString errorMsg) _ _)) = do
-        closeChannel' chan errorMsg
-        killThread =<< myThreadId
-    handleAsync (SimpleMethod (Channel_flow active)) = do
-        if active
-            then openLock $ chanActive chan
-            else closeLock $ chanActive chan
-        -- in theory we should respond with flow_ok but rabbitMQ 1.7 ignores that, so it doesn't matter
-        return ()
-    --Basic.return
-    handleAsync (ContentMethod (Basic_return _ _ _ _) _ _) =
-        --TODO: implement handling
-        -- this won't be called currently, because publishMsg sets "mandatory" and "immediate" to false
-        putStrLn ("BASIC.RETURN not implemented" :: String)
-    handleAsync m = error ("Unknown method: " ++ show m)
+    nul = '\0'
+    initialResponse = E.encodeUtf8 $ (T.cons nul loginName) `T.append` (T.cons nul loginPassword)
 
--- closes the channel internally; but doesn't tell the server
-closeChannel' :: Channel -> Text -> IO ()
-closeChannel' c reason = do
-    modifyMVar_ (connChannels $ connection c) $ \old -> return $ IM.delete (fromIntegral $ channelID c) old
-    -- mark channel as closed
-    modifyMVar_ (chanClosed c) $ \x -> do
-        if isNothing x
-            then do
-                void $ killLock $ chanActive c
-                killOutstandingResponses $ outstandingResponses c
-                return $ Just $ maybe (T.unpack reason) id x
-            else return x
+-- | The @AMQPLAIN@ SASL mechanism. See <http://www.rabbitmq.com/authentication.html>.
+amqplain :: Text -> Text -> SASLMechanism
+amqplain loginName loginPassword = SASLMechanism "AMQPLAIN" initialResponse Nothing
   where
-    killOutstandingResponses :: MVar (Seq.Seq (MVar a)) -> IO ()
-    killOutstandingResponses outResps = do
-        modifyMVar_ outResps $ \val -> do
-            F.mapM_ (\x -> tryPutMVar x $ error "channel closed") val
-            return undefined
-
--- | opens a new channel on the connection
---
--- There's currently no closeChannel method, but you can always just close the connection (the maximum number of channels is 65535).
-openChannel :: Connection -> IO Channel
-openChannel c = do
-    newInQueue <- newChan
-    outRes <- newMVar Seq.empty
-    lastConsTag <- newMVar 0
-    ca <- newLock
-
-    closed <- newMVar Nothing
-    conss <- newMVar M.empty
-
-    --get a new unused channelID
-    newChannelID <- modifyMVar (lastChannelID c) $ \x -> return (x+1, x+1)
-
-    let newChannel = Channel c newInQueue outRes (fromIntegral newChannelID) lastConsTag ca closed conss
-
-    thrID <- forkIO $ CE.finally (channelReceiver newChannel)
-        (closeChannel' newChannel "closed")
-
-    --add new channel to connection's channel map
-    modifyMVar_ (connChannels c) (return . IM.insert newChannelID (newChannel, thrID))
-
-    (SimpleMethod (Channel_open_ok _)) <- request newChannel (SimpleMethod (Channel_open (ShortString "")))
-    return newChannel
-
--- | writes multiple frames to the channel atomically
-writeFrames :: Channel -> [FramePayload] -> IO ()
-writeFrames chan payloads =
-    let conn = connection chan in
-        withMVar (connChannels conn) $ \chans ->
-            if IM.member (fromIntegral $ channelID chan) chans
-                then
-                    CE.catch
-                        -- ensure at most one thread is writing to the socket at any time
-                        (withMVar (connWriteLock conn) $ \_ ->
-                            mapM_ (\payload -> writeFrame (connHandle conn) (Frame (channelID chan) payload)) payloads)
-                        ( \(_ :: CE.IOException) -> do
-                            CE.throwIO $ userError "connection not open"
-                        )
-                else do
-                    CE.throwIO $ userError "channel not open"
+    initialResponse = BL.toStrict $ BL.drop 4 $ runPut $ put $ FieldTable $ M.fromList [("LOGIN",FVString loginName), ("PASSWORD", FVString loginPassword)]
 
-writeAssembly' :: Channel -> Assembly -> IO ()
-writeAssembly' chan (ContentMethod m properties msg) = do
-    -- wait iff the AMQP server instructed us to withhold sending content data (flow control)
-    waitLock $ chanActive chan
-    let !toWrite =
-           [(MethodPayload m),
-            (ContentHeaderPayload
-                (getClassIDOf properties) --classID
-                0 --weight is deprecated in AMQP 0-9
-                (fromIntegral $ BL.length msg) --bodySize
-                properties)] ++
-            (if BL.length msg > 0
-             then do
-                -- split into frames of maxFrameSize
-                -- (need to substract 8 bytes to account for frame header and end-marker)
-                map ContentBodyPayload
-                    (splitLen msg $ (fromIntegral $ connMaxFrameSize $ connection chan) - 8)
-             else []
-            )
-    writeFrames chan toWrite
+-- | The @RABBIT-CR-DEMO@ SASL mechanism needs to be explicitly enabled on the RabbitMQ server and should only be used for demonstration purposes of the challenge-response cycle.
+-- See <http://www.rabbitmq.com/authentication.html>.
+rabbitCRdemo :: Text -> Text -> SASLMechanism
+rabbitCRdemo loginName loginPassword = SASLMechanism "RABBIT-CR-DEMO" initialResponse (Just $ const challengeResponse)
   where
-    splitLen str len | BL.length str > len = (BL.take len str):(splitLen (BL.drop len str) len)
-    splitLen str _ = [str]
-writeAssembly' chan (SimpleMethod m) = writeFrames chan [MethodPayload m]
-
--- most exported functions in this module will use either 'writeAssembly' or 'request' to talk to the server
--- so we perform the exception handling here
-
--- | writes an assembly to the channel
-writeAssembly :: Channel -> Assembly -> IO ()
-writeAssembly chan m =
-    CE.catches
-        (writeAssembly' chan m)
-        [CE.Handler (\ (_ :: AMQPException) -> throwMostRelevantAMQPException chan),
-         CE.Handler (\ (_ :: CE.ErrorCall) -> throwMostRelevantAMQPException chan),
-         CE.Handler (\ (_ :: CE.IOException) -> throwMostRelevantAMQPException chan)]
-
--- | sends an assembly and receives the response
-request :: Channel -> Assembly -> IO Assembly
-request chan m = do
-    res <- newEmptyMVar
-    CE.catches (do
-            withMVar (chanClosed chan) $ \cc -> do
-                if isNothing cc
-                    then do
-                        modifyMVar_ (outstandingResponses chan) $ \val -> return $! val Seq.|> res
-                        writeAssembly' chan m
-                    else CE.throwIO $ userError "closed"
-
-            -- res might contain an exception, so evaluate it here
-            !r <- takeMVar res
-            return r
-            )
-        [CE.Handler (\ (_ :: AMQPException) -> throwMostRelevantAMQPException chan),
-         CE.Handler (\ (_ :: CE.ErrorCall) -> throwMostRelevantAMQPException chan),
-         CE.Handler (\ (_ :: CE.IOException) -> throwMostRelevantAMQPException chan)]
-
--- this throws an AMQPException based on the status of the connection and the channel
--- if both connection and channel are closed, it will throw a ConnectionClosedException
-throwMostRelevantAMQPException :: Channel -> IO a
-throwMostRelevantAMQPException chan = do
-    cc <- readMVar $ connClosed $ connection chan
-    case cc of
-        Just r -> CE.throwIO $ ConnectionClosedException r
-        Nothing -> do
-            chc <- readMVar $ chanClosed chan
-            case chc of
-                Just r -> CE.throwIO $ ChannelClosedException r
-                Nothing -> CE.throwIO $ ConnectionClosedException "unknown reason"
+    initialResponse = E.encodeUtf8 loginName
+    challengeResponse = return $ (E.encodeUtf8 "My password is ") `BS.append` (E.encodeUtf8 loginPassword)
 
 -- | @qos chan prefetchSize prefetchCount@ limits the amount of data the server
 -- delivers before requiring acknowledgements. @prefetchSize@ specifies the
@@ -974,12 +561,3 @@         False
         ))
     return ()
-
------------------------------ 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

@@ -32,4 +32,8 @@ killLock :: Lock -> IO Bool
 killLock (Lock a b) = do
     modifyMVar_ a $ const (return True)
-    tryPutMVar b ()+    tryPutMVar b ()
+
+chooseMin :: Ord a => a -> Maybe a -> a
+chooseMin a (Just b) = min a b
+chooseMin a Nothing  = a

@@ -0,0 +1,601 @@+{-# LANGUAGE BangPatterns, DeriveDataTypeable, OverloadedStrings, ScopedTypeVariables #-}
+module Network.AMQP.Internal where
+
+import Control.Concurrent
+import Control.Monad
+import Data.Binary
+import Data.Binary.Get
+import Data.Binary.Put as BPut
+import Data.Maybe
+import Data.Text (Text)
+import Data.Typeable
+import Network
+import System.IO
+
+import qualified Control.Exception as CE
+import qualified Data.ByteString as BS
+import qualified Data.ByteString.Char8 as BC
+import qualified Data.ByteString.Lazy as BL
+import qualified Data.Map as M
+import qualified Data.Foldable as F
+import qualified Data.IntMap as IM
+import qualified Data.Sequence as Seq
+import qualified Data.Text as T
+import qualified Data.Text.Encoding as E
+
+import Network.AMQP.Protocol
+import Network.AMQP.Types
+import Network.AMQP.Helpers
+import Network.AMQP.Generated
+
+
+data DeliveryMode = Persistent -- ^ the message will survive server restarts (if the queue is durable)
+                  | NonPersistent -- ^ the message may be lost after server restarts
+    deriving (Eq, Ord, Read, Show)
+
+deliveryModeToInt :: DeliveryMode -> Octet
+deliveryModeToInt NonPersistent = 1
+deliveryModeToInt Persistent = 2
+
+intToDeliveryMode :: Octet -> DeliveryMode
+intToDeliveryMode 1 = NonPersistent
+intToDeliveryMode 2 = Persistent
+intToDeliveryMode n = error ("Unknown delivery mode int: " ++ show n)
+
+-- | An AMQP message
+data Message = Message {
+                msgBody :: BL.ByteString, -- ^ the content of your message
+                msgDeliveryMode :: Maybe DeliveryMode, -- ^ see 'DeliveryMode'
+                msgTimestamp :: Maybe Timestamp, -- ^ use in any way you like; this doesn't affect the way the message is handled
+                msgID :: Maybe Text, -- ^ use in any way you like; this doesn't affect the way the message is handled
+                msgContentType :: Maybe Text,
+                msgReplyTo :: Maybe Text,
+                msgCorrelationID :: Maybe Text,
+                msgHeaders :: Maybe FieldTable
+                }
+    deriving (Eq, Ord, Read, Show)
+
+-- | contains meta-information of a delivered message (through 'getMsg' or 'consumeMsgs')
+data Envelope = Envelope
+              {
+                envDeliveryTag :: LongLongInt,
+                envRedelivered :: Bool,
+                envExchangeName :: Text,
+                envRoutingKey :: Text,
+                envChannel :: Channel
+              }
+
+data PublishError = PublishError
+                  {
+                    errReplyCode :: ReturnReplyCode,
+                    errExchange :: Maybe Text,
+                    errRoutingKey :: Text
+                  }
+    deriving (Eq, Read, Show)
+
+data ReturnReplyCode = Unroutable Text
+                     | NoConsumers Text
+                     | NotFound Text
+    deriving (Eq, Read, Show)
+
+------------- ASSEMBLY -------------------------
+-- an assembly is a higher-level object consisting of several frames (like in amqp 0-10)
+data Assembly = SimpleMethod MethodPayload
+              | ContentMethod MethodPayload ContentHeaderProperties BL.ByteString --method, properties, content-data
+    deriving Show
+
+-- | reads all frames necessary to build an assembly
+readAssembly :: Chan FramePayload -> IO Assembly
+readAssembly chan = do
+    m <- readChan chan
+    case m of
+        MethodPayload p -> --got a method frame
+            if hasContent m
+                then do
+                    --several frames containing the content will follow, so read them
+                    (props, msg) <- collectContent chan
+                    return $ ContentMethod p props msg
+                else do
+                    return $ SimpleMethod p
+        x -> error $ "didn't expect frame: " ++ show x
+
+-- | reads a contentheader and contentbodies and assembles them
+collectContent :: Chan FramePayload -> IO (ContentHeaderProperties, BL.ByteString)
+collectContent chan = do
+    (ContentHeaderPayload _ _ bodySize props) <- readChan chan
+
+    content <- collect $ fromIntegral bodySize
+    return (props, BL.concat content)
+  where
+    collect x | x <= 0 = return []
+    collect x = do
+        (ContentBodyPayload payload) <- readChan chan
+        r <- collect (x - (BL.length payload))
+        return $ payload : r
+
+------------ CONNECTION -------------------
+
+{- general concept:
+Each connection has its own thread. Each channel has its own thread.
+Connection reads data from socket and forwards it to channel. Channel processes data and forwards it to application.
+Outgoing data is written directly onto the socket.
+
+Incoming Data: Socket -> Connection-Thread -> Channel-Thread -> Application
+Outgoing Data: Application -> Socket
+-}
+
+data Connection = Connection {
+                    connHandle :: Handle,
+                    connChannels :: (MVar (IM.IntMap (Channel, ThreadId))), --open channels (channelID => (Channel, ChannelThread))
+                    connMaxFrameSize :: Int, --negotiated maximum frame size
+                    connClosed :: MVar (Maybe String),
+                    connClosedLock :: MVar (), -- used by closeConnection to block until connection-close handshake is complete
+                    connWriteLock :: MVar (), -- to ensure atomic writes to the socket
+                    connClosedHandlers :: MVar [IO ()],
+                    lastChannelID :: MVar Int --for auto-incrementing the channelIDs
+                }
+
+-- | Represents the parameters to connect to a broker or a cluster of brokers.
+-- See 'defaultConnectionOpts'.
+--
+-- /NOTICE/: The fields 'coHeartbeatDelay' and 'coMaxChannel' were 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.
+                            coAuth :: ![SASLMechanism], -- ^ The 'SASLMechanism's to use for authenticating with the broker.
+                            coMaxFrameSize :: !(Maybe Word32), -- ^ The maximum frame size to be used. If not specified, no limit is assumed.
+                            coHeartbeatDelay :: !(Maybe Word16), -- ^ The delay in seconds, after which the client expects a heartbeat frame from the broker. If not specified, no heartbeat is expected.
+                            coMaxChannel :: !(Maybe Word16) -- ^ The maximum number of channels the client will use.
+                        }
+
+-- | A 'SASLMechanism' is described by its name ('saslName'), its initial response ('saslInitialResponse'), and an optional function ('saslChallengeFunc') that
+-- transforms a security challenge provided by the server into response, which is then sent back to the server for verification.
+data SASLMechanism = SASLMechanism {
+                        saslName :: !Text, -- ^ mechanism name
+                        saslInitialResponse :: !BS.ByteString, -- ^ initial response
+                        saslChallengeFunc :: !(Maybe (BS.ByteString -> IO BS.ByteString)) -- ^ challenge processing function
+                    }
+
+-- | reads incoming frames from socket and forwards them to the opened channels
+connectionReceiver :: Connection -> IO ()
+connectionReceiver conn = do
+    CE.catch (do
+        Frame chanID payload <- readFrame (connHandle conn)
+        forwardToChannel chanID payload
+        )
+        (\(e :: CE.IOException) -> do
+            modifyMVar_ (connClosed conn) $ const $ return $ Just $ show e
+            killThread =<< myThreadId
+            )
+    connectionReceiver conn
+  where
+    forwardToChannel 0 (MethodPayload Connection_close_ok) = do
+        modifyMVar_ (connClosed conn) $ const $ return $ Just "closed by user"
+        killThread =<< myThreadId
+    forwardToChannel 0 (MethodPayload (Connection_close _ (ShortString errorMsg) _ _)) = do
+        writeFrame (connHandle conn) $ Frame 0 $ MethodPayload Connection_close_ok
+        modifyMVar_ (connClosed conn) $ const $ return $ Just $ T.unpack errorMsg
+        killThread =<< myThreadId
+    forwardToChannel 0 payload = putStrLn $ "Got unexpected msg on channel zero: " ++ show payload
+    forwardToChannel chanID payload = do
+        --got asynchronous msg => forward to registered channel
+        withMVar (connChannels conn) $ \cs -> do
+            case IM.lookup (fromIntegral chanID) cs of
+                Just c -> writeChan (inQueue $ fst c) payload
+                Nothing -> putStrLn $ "ERROR: channel not open " ++ show chanID
+
+-- | Opens a connection to a broker specified by the given 'ConnectionOpts' parameter.
+openConnection'' :: ConnectionOpts -> IO Connection
+openConnection'' connOpts = withSocketsDo $ do
+    handle <- connect $ coServers connOpts
+    maxFrameSize <- CE.handle (\(_ :: CE.IOException) -> CE.throwIO $ ConnectionClosedException "Handshake failed. Please check the RabbitMQ logs for more information") $ do
+        BL.hPut handle $ BPut.runPut $ do
+            BPut.putByteString $ BC.pack "AMQP"
+            BPut.putWord8 1
+            BPut.putWord8 1 --TCP/IP
+            BPut.putWord8 0 --Major Version
+            BPut.putWord8 9 --Minor Version
+        hFlush handle
+
+        -- S: connection.start
+        Frame 0 (MethodPayload (Connection_start _ _ _ (LongString serverMechanisms) _)) <- readFrame handle
+        selectedSASL <- selectSASLMechanism handle serverMechanisms
+
+        -- C: start_ok
+        writeFrame handle $ start_ok selectedSASL
+        -- S: secure or tune
+        Frame 0 (MethodPayload (Connection_tune _ frame_max _)) <- handleSecureUntilTune handle selectedSASL
+        -- C: tune_ok
+        let maxFrameSize = chooseMin frame_max $ coMaxFrameSize connOpts
+
+        writeFrame handle (Frame 0 (MethodPayload
+            --TODO: handle channel_max
+            (Connection_tune_ok 0 maxFrameSize 0)
+            ))
+        -- C: open
+        writeFrame handle open
+
+        -- S: open_ok
+        Frame 0 (MethodPayload (Connection_open_ok _)) <- readFrame handle
+
+        -- Connection established!
+        return maxFrameSize
+
+    --build Connection object
+    cChannels <- newMVar IM.empty
+    lastChanID <- newMVar 0
+    cClosed <- newMVar Nothing
+    writeLock <- newMVar ()
+    ccl <- newEmptyMVar
+    cClosedHandlers <- newMVar []
+    let conn = Connection handle cChannels (fromIntegral maxFrameSize) cClosed ccl writeLock cClosedHandlers lastChanID
+
+    --spawn the connectionReceiver
+    void $ forkIO $ CE.finally (connectionReceiver conn) $ do
+                -- try closing socket
+                CE.catch (hClose handle) (\(_ :: CE.SomeException) -> return ())
+
+                -- mark as closed
+                modifyMVar_ cClosed $ return . Just . maybe "unknown reason" id
+
+                --kill all channel-threads
+                modifyMVar_ cChannels $ \x -> do
+                    mapM_ (killThread . snd) $ IM.elems x
+                    return IM.empty
+
+                -- mark connection as closed, so all pending calls to 'closeConnection' can now return
+                void $ tryPutMVar ccl ()
+
+                -- notify connection-close-handlers
+                withMVar cClosedHandlers sequence
+    return conn
+  where
+    connect ((host, port) : rest) = do
+        result <- CE.try (connectTo host $ PortNumber port)
+        either
+            (\(ex :: CE.SomeException) -> do
+                putStrLn $ "Error connecting to "++show (host, port)++": "++show ex
+                connect rest)
+            (return)
+            result
+    connect [] = CE.throwIO $ ConnectionClosedException $ "Could not connect to any of the provided brokers: " ++ show (coServers connOpts)
+    selectSASLMechanism handle serverMechanisms =
+        let serverSaslList = T.split (== ' ') $ E.decodeUtf8 serverMechanisms
+            clientMechanisms = coAuth connOpts
+            clientSaslList = map saslName clientMechanisms
+            maybeSasl = F.find (\(SASLMechanism name _ _) -> elem name serverSaslList) clientMechanisms
+        in abortIfNothing maybeSasl handle
+            ("None of the provided SASL mechanisms "++show clientSaslList++" is supported by the server "++show serverSaslList++".")
+
+    start_ok sasl = (Frame 0 (MethodPayload (Connection_start_ok (FieldTable M.empty)
+        (ShortString $ saslName sasl)
+        (LongString $ saslInitialResponse sasl)
+        (ShortString "en_US")) ))
+
+    handleSecureUntilTune handle sasl = do
+        tuneOrSecure <- readFrame handle
+        case tuneOrSecure of
+            Frame 0 (MethodPayload (Connection_secure (LongString challenge))) -> do
+                processChallenge <- abortIfNothing (saslChallengeFunc sasl)
+                    handle $ "The server provided a challenge, but the selected SASL mechanism "++show (saslName sasl)++" is not equipped with a challenge processing function."
+                challengeResponse <- processChallenge challenge
+                writeFrame handle (Frame 0 (MethodPayload (Connection_secure_ok (LongString challengeResponse))))
+                handleSecureUntilTune handle sasl
+
+            tune@(Frame 0 (MethodPayload (Connection_tune _ _ _))) -> return tune
+            x -> error $ "handleSecureUntilTune fail. received message: "++show x
+
+    open = (Frame 0 (MethodPayload (Connection_open
+        (ShortString $ coVHost connOpts)
+        (ShortString $ T.pack "") -- capabilities; deprecated in 0-9-1
+        True))) -- insist; deprecated in 0-9-1
+
+    abortHandshake handle msg = do
+        hClose handle
+        CE.throwIO $ ConnectionClosedException msg
+
+    abortIfNothing m handle msg = case m of
+        Nothing -> abortHandshake handle msg
+        Just a  -> return a
+
+-- | closes a connection
+--
+-- Make sure to call this function before your program exits to ensure that all published messages are received by the server.
+closeConnection :: Connection -> IO ()
+closeConnection c = do
+    CE.catch (
+        withMVar (connWriteLock c) $ \_ -> writeFrame (connHandle c) $ (Frame 0 (MethodPayload (Connection_close
+            --TODO: set these values
+            0 -- reply_code
+            (ShortString "") -- reply_text
+            0 -- class_id
+            0 -- method_id
+            )))
+            )
+        (\ (_ :: CE.IOException) ->
+            --do nothing if connection is already closed
+            return ()
+        )
+
+    -- wait for connection_close_ok by the server; this MVar gets filled in the CE.finally handler in openConnection'
+    readMVar $ connClosedLock c
+    return ()
+
+-- | @addConnectionClosedHandler conn ifClosed handler@ adds a @handler@ that will be called after the connection is closed (either by calling @closeConnection@ or by an exception). If the @ifClosed@ parameter is True and the connection is already closed, the handler will be called immediately. If @ifClosed == False@ and the connection is already closed, the handler will never be called
+addConnectionClosedHandler :: Connection -> Bool -> IO () -> IO ()
+addConnectionClosedHandler conn ifClosed handler = do
+    withMVar (connClosed conn) $ \cc ->
+        case cc of
+            -- connection is already closed, so call the handler directly
+            Just _ | ifClosed == True -> handler
+
+            -- otherwise add it to the list
+            _ -> modifyMVar_ (connClosedHandlers conn) $ \old -> return $ handler:old
+
+readFrame :: Handle -> IO Frame
+readFrame handle = do
+    dat <- BL.hGet handle 7
+    -- NB: userError returns an IOException so it will be catched in 'connectionReceiver'
+    when (BL.null dat) $ CE.throwIO $ userError "connection not open"
+    let len = fromIntegral $ peekFrameSize dat
+    dat' <- BL.hGet handle (len+1) -- +1 for the terminating 0xCE
+    when (BL.null dat') $ CE.throwIO $ userError "connection not open"
+    let ret = runGetOrFail get (BL.append dat dat')
+    case ret of
+        Left (_, _, errMsg) -> error $ "readFrame fail: " ++ errMsg
+        Right (_, consumedBytes, _) | consumedBytes /= fromIntegral (len+8) ->
+            error $ "readFrame: parser should read " ++ show (len+8) ++ " bytes; but read " ++ show consumedBytes
+        Right (_, _, frame) -> return frame
+
+writeFrame :: Handle -> Frame -> IO ()
+writeFrame handle f = do
+    BL.hPut handle . runPut . put $ f
+    hFlush handle
+
+------------------------ CHANNEL -----------------------------
+
+{- | A connection to an AMQP server is made up of separate channels. It is recommended to use a separate channel for each thread in your application that talks to the AMQP server (but you don't have to as channels are thread-safe)
+-}
+data Channel = Channel {
+                    connection :: Connection,
+                    inQueue :: Chan FramePayload, --incoming frames (from Connection)
+                    outstandingResponses :: MVar (Seq.Seq (MVar Assembly)), -- for every request an MVar is stored here waiting for the response
+                    channelID :: Word16,
+                    lastConsumerTag :: MVar Int,
+
+                    chanActive :: Lock, -- used for flow-control. if lock is closed, no content methods will be sent
+                    chanClosed :: MVar (Maybe String),
+                    consumers :: MVar (M.Map Text ((Message, Envelope) -> IO ())), -- who is consumer of a queue? (consumerTag => callback)
+                    returnListeners :: MVar ([(Message, PublishError) -> IO ()])
+                }
+
+msgFromContentHeaderProperties :: ContentHeaderProperties -> BL.ByteString -> Message
+msgFromContentHeaderProperties (CHBasic content_type _ headers delivery_mode _ correlation_id reply_to _ message_id timestamp _ _ _ _) body =
+    let msgId = fromShortString message_id
+        contentType = fromShortString content_type
+        replyTo = fromShortString reply_to
+        correlationID = fromShortString correlation_id
+    in Message body (fmap intToDeliveryMode delivery_mode) timestamp msgId contentType replyTo correlationID headers
+  where
+    fromShortString (Just (ShortString s)) = Just s
+    fromShortString _ = Nothing
+msgFromContentHeaderProperties c _ = error ("Unknown content header properties: " ++ show c)
+
+-- | The thread that is run for every channel
+channelReceiver :: Channel -> IO ()
+channelReceiver chan = do
+    --read incoming frames; they are put there by a Connection thread
+    p <- readAssembly $ inQueue chan
+    if isResponse p
+        then do
+            action <- modifyMVar (outstandingResponses chan) $ \val -> do
+                        case Seq.viewl val of
+                            x Seq.:< rest -> do
+                                return (rest, putMVar x p)
+                            Seq.EmptyL -> do
+                                return (val, CE.throwIO $ userError "got response, but have no corresponding request")
+            action
+
+        --handle asynchronous assemblies
+        else handleAsync p
+    channelReceiver chan
+  where
+    isResponse :: Assembly -> Bool
+    isResponse (ContentMethod (Basic_deliver _ _ _ _ _) _ _) = False
+    isResponse (ContentMethod (Basic_return _ _ _ _) _ _) = False
+    isResponse (SimpleMethod (Channel_flow _)) = False
+    isResponse (SimpleMethod (Channel_close _ _ _ _)) = False
+    isResponse _ = True
+
+    --Basic.Deliver: forward msg to registered consumer
+    handleAsync (ContentMethod (Basic_deliver (ShortString consumerTag) deliveryTag redelivered (ShortString exchange)
+                                                (ShortString routingKey))
+                                properties body) =
+        withMVar (consumers chan) (\s -> do
+            case M.lookup consumerTag s of
+                Just subscriber -> do
+                    let msg = msgFromContentHeaderProperties properties body
+                    let env = Envelope {envDeliveryTag = deliveryTag, envRedelivered = redelivered,
+                                    envExchangeName = exchange, envRoutingKey = routingKey, envChannel = chan}
+
+                    CE.catch (subscriber (msg, env))
+                        (\(e::CE.SomeException) -> putStrLn $ "AMQP callback threw exception: " ++ show e)
+                Nothing ->
+                    -- got a message, but have no registered subscriber; so drop it
+                    return ()
+            )
+    handleAsync (SimpleMethod (Channel_close _ (ShortString errorMsg) _ _)) = do
+        closeChannel' chan errorMsg
+        killThread =<< myThreadId
+    handleAsync (SimpleMethod (Channel_flow active)) = do
+        if active
+            then openLock $ chanActive chan
+            else closeLock $ chanActive chan
+        -- in theory we should respond with flow_ok but rabbitMQ 1.7 ignores that, so it doesn't matter
+        return ()
+    --Basic.return
+    handleAsync (ContentMethod basicReturn@(Basic_return _ _ _ _) props body) = do
+        let msg      = msgFromContentHeaderProperties props body
+            pubError = basicReturnToPublishError basicReturn
+        withMVar (returnListeners chan) $ \listeners ->
+            forM_ listeners $ \l -> CE.catch (l (msg, pubError)) $ \(ex :: CE.SomeException) ->
+                putStrLn $ "return listener on channel ["++(show $ channelID chan)++"] handling error ["++show pubError++"] threw exception: "++show ex
+    handleAsync m = error ("Unknown method: " ++ show m)
+
+    basicReturnToPublishError (Basic_return code (ShortString errText) (ShortString exchange) (ShortString routingKey)) =
+        let replyError = case code of
+                312 -> Unroutable errText
+                313 -> NoConsumers errText
+                404 -> NotFound errText
+                num -> error $ "unexpected return error code: " ++ (show num)
+            pubError = PublishError replyError (Just exchange) routingKey
+        in pubError
+    basicReturnToPublishError x = error $ "basicReturnToPublishError fail: "++show x
+
+-- | registers a callback function that is called whenever a message is returned from the broker ('basic.return').
+addReturnListener :: Channel -> ((Message, PublishError) -> IO ()) -> IO ()
+addReturnListener chan listener = do
+    modifyMVar_ (returnListeners chan) $ \listeners -> return $ listener:listeners
+
+-- closes the channel internally; but doesn't tell the server
+closeChannel' :: Channel -> Text -> IO ()
+closeChannel' c reason = do
+    modifyMVar_ (connChannels $ connection c) $ \old -> return $ IM.delete (fromIntegral $ channelID c) old
+    -- mark channel as closed
+    modifyMVar_ (chanClosed c) $ \x -> do
+        if isNothing x
+            then do
+                void $ killLock $ chanActive c
+                killOutstandingResponses $ outstandingResponses c
+                return $ Just $ maybe (T.unpack reason) id x
+            else return x
+  where
+    killOutstandingResponses :: MVar (Seq.Seq (MVar a)) -> IO ()
+    killOutstandingResponses outResps = do
+        modifyMVar_ outResps $ \val -> do
+            F.mapM_ (\x -> tryPutMVar x $ error "channel closed") val
+            return undefined
+
+-- | opens a new channel on the connection
+--
+-- There's currently no closeChannel method, but you can always just close the connection (the maximum number of channels is 65535).
+openChannel :: Connection -> IO Channel
+openChannel c = do
+    newInQueue <- newChan
+    outRes <- newMVar Seq.empty
+    lastConsTag <- newMVar 0
+    ca <- newLock
+
+    closed <- newMVar Nothing
+    conss <- newMVar M.empty
+    listeners <- newMVar []
+
+    --get a new unused channelID
+    newChannelID <- modifyMVar (lastChannelID c) $ \x -> return (x+1, x+1)
+
+    let newChannel = Channel c newInQueue outRes (fromIntegral newChannelID) lastConsTag ca closed conss listeners
+
+    thrID <- forkIO $ CE.finally (channelReceiver newChannel)
+        (closeChannel' newChannel "closed")
+
+    --add new channel to connection's channel map
+    modifyMVar_ (connChannels c) (return . IM.insert newChannelID (newChannel, thrID))
+
+    (SimpleMethod (Channel_open_ok _)) <- request newChannel (SimpleMethod (Channel_open (ShortString "")))
+    return newChannel
+
+-- | writes multiple frames to the channel atomically
+writeFrames :: Channel -> [FramePayload] -> IO ()
+writeFrames chan payloads =
+    let conn = connection chan in
+        withMVar (connChannels conn) $ \chans ->
+            if IM.member (fromIntegral $ channelID chan) chans
+                then
+                    CE.catch
+                        -- ensure at most one thread is writing to the socket at any time
+                        (withMVar (connWriteLock conn) $ \_ ->
+                            mapM_ (\payload -> writeFrame (connHandle conn) (Frame (channelID chan) payload)) payloads)
+                        ( \(_ :: CE.IOException) -> do
+                            CE.throwIO $ userError "connection not open"
+                        )
+                else do
+                    CE.throwIO $ userError "channel not open"
+
+writeAssembly' :: Channel -> Assembly -> IO ()
+writeAssembly' chan (ContentMethod m properties msg) = do
+    -- wait iff the AMQP server instructed us to withhold sending content data (flow control)
+    waitLock $ chanActive chan
+    let !toWrite =
+           [(MethodPayload m),
+            (ContentHeaderPayload
+                (getClassIDOf properties) --classID
+                0 --weight is deprecated in AMQP 0-9
+                (fromIntegral $ BL.length msg) --bodySize
+                properties)] ++
+            (if BL.length msg > 0
+             then do
+                -- split into frames of maxFrameSize
+                -- (need to substract 8 bytes to account for frame header and end-marker)
+                map ContentBodyPayload
+                    (splitLen msg $ (fromIntegral $ connMaxFrameSize $ connection chan) - 8)
+             else []
+            )
+    writeFrames chan toWrite
+  where
+    splitLen str len | BL.length str > len = (BL.take len str):(splitLen (BL.drop len str) len)
+    splitLen str _ = [str]
+writeAssembly' chan (SimpleMethod m) = writeFrames chan [MethodPayload m]
+
+-- most exported functions in this module will use either 'writeAssembly' or 'request' to talk to the server
+-- so we perform the exception handling here
+
+-- | writes an assembly to the channel
+writeAssembly :: Channel -> Assembly -> IO ()
+writeAssembly chan m =
+    CE.catches
+        (writeAssembly' chan m)
+        [CE.Handler (\ (_ :: AMQPException) -> throwMostRelevantAMQPException chan),
+         CE.Handler (\ (_ :: CE.ErrorCall) -> throwMostRelevantAMQPException chan),
+         CE.Handler (\ (_ :: CE.IOException) -> throwMostRelevantAMQPException chan)]
+
+-- | sends an assembly and receives the response
+request :: Channel -> Assembly -> IO Assembly
+request chan m = do
+    res <- newEmptyMVar
+    CE.catches (do
+            withMVar (chanClosed chan) $ \cc -> do
+                if isNothing cc
+                    then do
+                        modifyMVar_ (outstandingResponses chan) $ \val -> return $! val Seq.|> res
+                        writeAssembly' chan m
+                    else CE.throwIO $ userError "closed"
+
+            -- res might contain an exception, so evaluate it here
+            !r <- takeMVar res
+            return r
+            )
+        [CE.Handler (\ (_ :: AMQPException) -> throwMostRelevantAMQPException chan),
+         CE.Handler (\ (_ :: CE.ErrorCall) -> throwMostRelevantAMQPException chan),
+         CE.Handler (\ (_ :: CE.IOException) -> throwMostRelevantAMQPException chan)]
+
+-- this throws an AMQPException based on the status of the connection and the channel
+-- if both connection and channel are closed, it will throw a ConnectionClosedException
+throwMostRelevantAMQPException :: Channel -> IO a
+throwMostRelevantAMQPException chan = do
+    cc <- readMVar $ connClosed $ connection chan
+    case cc of
+        Just r -> CE.throwIO $ ConnectionClosedException r
+        Nothing -> do
+            chc <- readMVar $ chanClosed chan
+            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,4 +1,7 @@ {-# LANGUAGE BangPatterns #-}
+-- |
+--
+-- This module contains data-types specified in the AMQP spec
 module Network.AMQP.Types (
     Octet,
     Bit,
@@ -25,8 +28,8 @@ import Data.Char
 import Data.Text (Text)
 
-import qualified Data.ByteString.Char8 as BS
-import qualified Data.ByteString.Lazy.Char8 as BL
+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
 
@@ -73,18 +76,17 @@                 putWord8 $ fromIntegral (BS.length s)
                 putByteString s
 
-newtype LongString = LongString Text
+newtype LongString = LongString BS.ByteString
     deriving (Eq, Ord, Read, Show)
 instance Binary LongString where
     get = do
       len <- getWord32be
       dat <- getByteString (fromIntegral len)
-      return $ LongString $ T.decodeUtf8 dat
+      return $ LongString dat
 
     put (LongString x) = do
-        let s = T.encodeUtf8 x
-        putWord32be $ fromIntegral (BS.length s)
-        putByteString s
+        putWord32be $ fromIntegral (BS.length x)
+        putByteString x
 
 type Timestamp = Word64
 
@@ -140,7 +142,7 @@             'D' -> FVDecimal <$> get
             'S' -> do
                 LongString x <- get :: Get LongString
-                return $ FVString x
+                return $ FVString $ T.decodeUtf8 x
             'A' -> do
                 len <- get :: Get Int32
                 if len > 0
@@ -165,7 +167,7 @@     put (FVFloat x) = put 'f' >> putFloat32be x
     put (FVDouble x) = put 'd' >> putFloat64be x
     put (FVDecimal x) = put 'D' >> put x
-    put (FVString x) = put 'S' >> put (LongString x)
+    put (FVString x) = put 'S' >> put (LongString $ T.encodeUtf8 x)
     put (FVFieldArray x) = do
         put 'A'
         if length x == 0

@@ -1,5 +1,5 @@ Name:                amqp
-Version:             0.5.0
+Version:             0.6.0
 Synopsis:            Client library for AMQP servers (currently only RabbitMQ)
 Description:         Client library for AMQP servers (currently only RabbitMQ)
                      .
@@ -18,9 +18,9 @@                      examples/ExampleProducer.hs
 
 Library
-  Build-Depends:      base >= 4 && < 5, binary >= 0.7, containers>=0.2, bytestring>=0.9, network>=2.2.3.1, data-binary-ieee754>=0.4.2.1, text>=0.11.2.3
+  Build-Depends:      base >= 4 && < 5, binary >= 0.7, containers>=0.2, bytestring>=0.9, network>=2.2.3.1, data-binary-ieee754>=0.4.2.1, text>=0.11.2
   Exposed-modules:    Network.AMQP, Network.AMQP.Types
-  Other-modules:      Network.AMQP.Generated, Network.AMQP.Helpers, Network.AMQP.Protocol
+  Other-modules:      Network.AMQP.Generated, Network.AMQP.Helpers, Network.AMQP.Protocol, Network.AMQP.Internal
   GHC-Options:        -Wall
 
 Executable amqp-builder