Contents

# frozen_string_literal: true

# :reek:InstanceVariableAssumption

module RubyRabbitmqJanus
  module Rabbit
    module Listener
      # Listener to admin queue
      class FromAdmin < From
        private

        def subscribe_queue
          reply = rabbit.queue(Tools::Config.instance.queue_admin_from)
          rabbit.prefetch(1)
          reply.bind(binding).subscribe(opts_subs) do |info, prop, payload|
            info_subscribe(info, prop, payload)
            synchronize_response(info, payload)
          end
        rescue => exception
          raise RubyRabbitmqJanus::Errors::Rabbit::Listener::From::ListenEvents,
                exception
        end

        def synchronize_response(info, payload)
          lock.synchronize do
            response = Janus::Responses::Admin.new(JSON.parse(payload))
            responses.push(response)
          end
          rabbit.acknowledge(info.delivery_tag, false)
          semaphore.signal
        end
      end
    end
  end
end

Version data entries

3 entries across 3 versions & 1 rubygems

Version	Path
ruby_rabbitmq_janus-2.7.2.pre.319	lib/rrj/rabbit/listener/from_admin.rb
ruby_rabbitmq_janus-2.7.2.pre.318	lib/rrj/rabbit/listener/from_admin.rb
ruby_rabbitmq_janus-2.7.2.pre.317	lib/rrj/rabbit/listener/from_admin.rb