Subscriber stops receiving messages.

[gaurav-gupta123]

I've integrated Bunny Gem for publishing and subscribing the active record events like create update, destroy the object among distributed systems.

As of now, it is working fine, but whenever there is no activity for a few minutes like 5 minutes after that subscriber stops receiving further messages. All the messages get piled up in the queue.

Below is the sample code and configuration for subscribing to the queue:

    @connection = Bunny.new(
      host: Settings.rabbit_mq.host,
      port: Settings.rabbit_mq.port,
      vhost: Settings.rabbit_mq.vhost,
      user: Settings.rabbit_mq.user,
      password: Settings.rabbit_mq.password,
      automatically_recover: true,
      heartbeat: 30
    )

    Thread.new do
      @connection.start
      channel = @connection.create_channel
      queue = channel.queue('test_queue', durable: true)
      begin
        queue.subscribe(manual_ack: true, block: true) do |delivery_info, _properties, payload|
          puts "=====Received Payload #{payload}====="
          channel.ack(delivery_info.delivery_tag)
        end
      rescue Interrupt => _e
        channel.reject(delivery_info.delivery_tag, true)
      end
    end

Kindly help me out If I missed something.

[michaelklishin]

Start with inspecting server logs. You can make Bunny log at debug level, too. We do not guess in the RabbitMQ community.

Two issues stand out in your code:

• Bunny::subscribe(block: true) was introduced for very basic scenarios like tests and tutorials. Avoid it in all other cases
• Should it return for any reason, your thread will finish executing and the connection will be GC'ed, which means it will be closed, and no new deliveries can possibly happen on it. RabbitMQ will log a message about unexpected TCP connection closure by the client

If you need to process deliveries concurrently and are sure you understand the risks in doing so (e.g. your deliveries can be processed in parallel because they do not depend on one another in any way), then dispatching deliveries to a thread pool any way you like should be enough:

@connection.start
ch = @connection.create_channel
q = ch.queue('test_queue', durable: true)

queue.subscribe(manual_ack: true) do |delivery_info, _properties, payload|
  puts "=====Received Payload #{payload}====="
  # This is just an example. I cannot know what thread pool implementation you will use
  # and what API it will have. But dispatching a delivery processing to it
  # will allow Bunny delivery dispatcher to move on to the next delivery
  pool.submit do
    # do something with the delivery
  end
  # it's a better idea to ack here to avoid double acknowledgements
  # https://rabbitmq.com/confirms.html#consumer-acks-double-acking
  #
  # If you are sure that double acknowledging is not a risk, you can choose
  # to acknowledge in the submit block above, right after doing the actual work
  # on/with the delivery
  channel.ack(delivery_info.delivery_tag)
end

# keep the thread alive to let the connection-channel and consumer run
loop do
  sleep 1
end

[gaurav-gupta123]

Thanks for the reply.

• I'll remove the block: true argument.
• I can go with concurrent processing of the deliveries as there is no impact on the others.
• As I've written my subscriber inside the thread it will keep running asynchronously and yes it works for some time but after being idle for some time it stops receiving further messages.

Below is the approach I followed to make it available asynchronously.

1. Wrote a service inside the services:

app/services/rabbitmq_service.rb

# frozen_string_literal: true

require 'bunny'

class RabbitmqService
  def initialize
    @connection = Bunny.new(
      host: Rails.application.credentials.rabbitmq.host,
      port: Rails.application.credentials.rabbitmq.port,
      username: Rails.application.credentials.rabbitmq.username,
      vhost: Rails.application.credentials.rabbitmq.vhost,
      password: Rails.application.credentials.rabbitmq.password,
      heartbeat: 30
    )
  end

  def subscribe(queue_name)
    Thread.new do
      channel = create_channel
      queue = channel.queue(queue_name, durable: true)
      Rails.logger.info("========Subscribed to queue=====#{queue_name}=======")
      begin
        queue.subscribe(manual_ack: true, block: true) do |delivery_info, _properties, payload|
          Rails.logger.info "---payload--#{payload}---"
          # processs the payload
          channel.ack(delivery_info.delivery_tag)
        end
      rescue Interrupt => _e
        channel.reject(delivery_info.delivery_tag, true)
      end
    end
  end

  def create_channel
    @connection.start
    @connection.create_channel
  end
end

2. Whenever we run the rails server we invoke this service after initializing the application:

config/initializers/rabbitmq_subscriber.rb

# frozen_string_literal: true

Rails.application.config.after_initialize do

  if defined?(Rails::Server) && Rails.env != 'test'
    rabbitmq_setting = Rails.application.credentials.rabbitmq
    return if rabbitmq_setting.blank?
    RabbitmqService.new.subscribe(rabbitmq_setting.queue_name)
  end
end

I am not sure if we need to wrap the thread block inside the loop block.