     _generate_consumer_options: (options) ->
         noAck: not options.ack
 
+    # Channel utilities
+    # -----------------
+
+    _assert_exchange: (options, channel) ->
+        exch_name = options.channel
+        exch_options = @_generate_exchange_options(options)
+        if @_confirmed_exchanges[exch_name]
+            return when_(channel)
+
+        channel.assertExchange(exch_name, 'topic', exch_options).then =>
+            @_confirmed_exchanges[exch_name] = 1
+            channel
+
+    # Message encoding
+    # ----------------
+
+    _encode_message: (payload) ->
+        content = null
+        if typeof payload is 'string'
+            content = payload
+            content_type = 'text/plain'
+        else
+            content = JSON.stringify(payload)
+            content_type = 'application/json'
+        content = new Buffer(content)
+        {content, properties:
+            contentType: content_type
+        }
+
     _decode_message: (msg) =>
         content = msg.content
-        content_type = msg.content_type
+        content_type = msg.properties.contentType
         switch content_type
             when 'application/json'
                 try

            options = {}

        event_options = @_parse_event(event)
        pub_options = _.extend({}, @options, event_options, options)

        _cb = (errors) ->
            return unless cb
            if errors then cb(new Error('message publish fail')) else cb()

        logger.trace("publish message", {@options, pub_options, payload})
        @_when_exchange(pub_options)
            .then (exchange) ->
                exchange.publish(pub_options.topic, payload, {}, _cb)

            .fail cb
        @

    subscribe: (event, options, listener) ->

                throw new Error("Do not know how to hange message type: " + content_type)
        content

    # Message callbacks
    # -----------------

    _wrap_consumer_callback: (fn, channel, options) ->
        """ Wrapper for unpacking message content """
        decode = @_decode_message
