                 ])
                 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)

                    # where errors coudl be forwarded for operator intervention.
                    # We defenitelly don't want to re-put into queue, because if it is
                    # problem with message itself, we can end up in indefenite loop
                    logger.error('Error in message consumer', {err})
                .finally ->
                    channel.ack(msg)


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

    _wrap_ack_callback: (fn, channel) ->
