                 ])
                 topic = '10' # for x-consistent hash echange topic is a weight
 
+            if options.prefetch?
+                steps.push(ch.prefetch(options.prefetch))
+
             steps.concat([
                 ch.bindQueue(queue_name, queue_binding_source, topic)
-                ch.consume(queue_name, consumer)
+                ch.consume(queue_name, consumer, consumer_options)
             ])
             # TODO: add channel bindings to restart whole subscription if channel is closed
             p = when_.all(steps)

        wrapped = (msg) ->
            when_.try(-> decode(msg)).then(fn)
        if options.ack
            wrapped = @_wrap_ack_promise(wrapped, channel)
        wrapped

    _wrap_ack_callback: (fn, channel) ->

        or
        "<group_name>" = options

        you can subscripe with 'ack' option. In that case listener can return promise.
        Acknowledgemnt will be sent only one promise is resolved. It will return value
        ack will be sent imediately. Still usefull to limit rate
        ###
        unless listener
            listener = options
