def start
      channel.prefetch(10)
      queue.subscribe(manual_ack: true, exclusive: false) do |delivery_info, metadata, payload|
        begin
          body = JSON.parse(payload).with_indifferent_access
          status = run(body)
        rescue => e
          status = :error
        end

        if status == :ok
          channel.ack(delivery_info.delivery_tag)
        elsif status == :retry
          channel.reject(delivery_info.delivery_tag, true)
        else # :error, nil etc
          channel.reject(delivery_info.delivery_tag, false)
        end
      end

      wait_for_threads
    end
def start
      channel.prefetch(10)
      queue.subscribe(manual_ack: true, exclusive: false) do |delivery_info, metadata, payload|
        begin
          body = JSON.parse(payload).with_indifferent_access
          status = run(body)
        rescue => e
          status = :error
        end

        if status == :ok
          channel.ack(delivery_info.delivery_tag)
        elsif status == :retry
          channel.reject(delivery_info.delivery_tag, true)
        else # :error, nil etc
          channel.reject(delivery_info.delivery_tag, false)
        end
      end

      wait_for_threads
    end
Method which kick start the consumer process thread