# frozen_string_literal: true
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

            require 'json'
          
        
      
        

          

            1
            

            

            require 'active_support'
          
        
      
        

          

            1
            

            

            require 'active_support/inflector'
          
        
      
        

          

            1
            

            

            require 'context_request_subscriber/handler'
          
        
      
        

          

            1
            

            

            require 'context_request_subscriber/processor'
          
        
      
        

          

            1
            

            

            require 'context_request_subscriber/error_handler'
          
        
      
        

          

            1
            

            

            require 'context_request_subscriber/rabbitmq_subscriber'
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

            # Base module for the context request subscriber.
          
        
      
        

          

            
            

            

            # This class provides all the configurables to the logic for the
          
        
      
        

          

            
            

            

            # subscriber and handler logic of the request context logic.
          
        
      
        

          

            1
            

            

            module ContextRequestSubscriber
          
        
      
        

          

            1
            

            

              include ActiveSupport::Configurable
          
        
      
        

          

            1
            

            

              config_accessor(:logger, instance_accessor: false) do
          
        
      
        

          

            1
            

            

                ActiveSupport::TaggedLogging.new(Logger.new(STDOUT))
          
        
      
        

          

            
            

            

              end
          
        
      
        

          

            1
            

            

              config_accessor(:queue_name, instance_accessor: false) do
          
        
      
        

          

            1
            

            

                'fos.context_request'
          
        
      
        

          

            
            

            

              end
          
        
      
        

          

            1
            

            

              config_accessor(:exchange_name, instance_accessor: false) do
          
        
      
        

          

            1
            

            

                'fos.context_request'
          
        
      
        

          

            
            

            

              end
          
        
      
        

          

            
            

            

              # queue_durable: if the queue is durable. Default: true
          
        
      
        

          

            1
            

            

              config_accessor(:queue_durable, instance_accessor: false)
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

              # queue_auto_delete: if the queue gets autodeleted.
          
        
      
        

          

            1
            

            

              config_accessor(:queue_auto_delete, instance_accessor: false)
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

              # queue_exclusive: default false.
          
        
      
        

          

            1
            

            

              config_accessor(:queue_exclusive, instance_accessor: false)
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

              # Callable that is called instead of the default logic to return
          
        
      
        

          

            
            

            

              # a Bunny object for both the channel, the connection and the queue.
          
        
      
        

          

            1
            

            

              config_accessor(:fetch_queue_callback, instance_accessor: false)
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

              # routing_key: the routing_key used. Default #.
          
        
      
        

          

            2
            

            

              config_accessor(:routing_key, instance_accessor: false) { '#' }
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

              # Hash of small cased classname of callable object to be called in order
          
        
      
        

          

            
            

            

              # to handle the message. The type of the message indicates the handler class.
          
        
      
        

          

            
            

            

              # Defaults to JsonApiHandler for request and context types.
          
        
      
        

          

            1
            

            

              config_accessor(:handlers, instance_accessor: false) do
          
        
      
        

          

            1
            

            

                {
          
        
      
        

          

            
            

            

                  context: ContextRequestSubscriber::Handler::JsonApiHandler::Context,
          
        
      
        

          

            
            

            

                  request: ContextRequestSubscriber::Handler::JsonApiHandler::Request
          
        
      
        

          

            
            

            

                }
          
        
      
        

          

            
            

            

              end
          
        
      
        

          

            
            

            

              # Handler URL is the url to reach out for handling.
          
        
      
        

          

            1
            

            

              config_accessor(:handler_url, instance_accessor: false)
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

              # Set of session parameters
          
        
      
        

          

            2
            

            

              config_accessor(:session_params, instance_accessor: false) { {} }
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

              # url to connect to the RabbitMQ Server
          
        
      
        

          

            1
            

            

              config_accessor(:url, instance_accessor: false)
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

              # heartbeat: heartbeat for the connection. Defaults to nil.
          
        
      
        

          

            1
            

            

              config_accessor(:heartbeat, instance_accessor: false)
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

              # on_error: callable object that handles errors during processing the
          
        
      
        

          

            
            

            

              #       message.
          
        
      
        

          

            1
            

            

              config_accessor(:on_error, instance_accessor: false) do
          
        
      
        

          

            1
            

            

                ErrorHandler::LogErrorHandler
          
        
      
        

          

            
            

            

              end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

              # Instruct the subscriber to keep the subscriber alive even if the
          
        
      
        

          

            
            

            

              # queue is empty.
          
        
      
        

          

            
            

            

              # Default: true
          
        
      
        

          

            2
            

            

              config_accessor(:subscriber_keep_alive, instance_accessor: false) { false }
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            2
            

            

              config_accessor(:handler_params, instance_accessor: false) { {} }
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

              def self.run
          
        
      
        

          

            1
            

            

                subscriber = RabbitMQSubscriber.new(**config)
          
        
      
        

          

            1
            

            

                subscriber.run
          
        
      
        

          

            
            

            

              end
          
        
      
        

          

            
            

            

            end
          
        
      
    
  

    

      
        

          

            
            

            

            # frozen_string_literal: true
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

            module ContextRequestSubscriber
          
        
      
        

          

            1
            

            

              module ErrorHandler
          
        
      
        

          

            
            

            

                # :nodoc:
          
        
      
        

          

            1
            

            

                class LogErrorHandler
          
        
      
        

          

            1
            

            

                  def initialize(logger = nil)
          
        
      
        

          

            7
            

            

                    @logger = (logger || ContextRequestSubscriber.logger)
          
        
      
        

          

            
            

            

                  end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                  def call(error, _info, _properties, _payload)
          
        
      
        

          

            3
            

            

                    @logger.error("Error recieved during processing the message. \
          
        
      
        

          

            
            

            

            Error #{error}.")
          
        
      
        

          

            3
            

            

                    @logger.error(error.backtrace.join("\n")) if error.backtrace
          
        
      
        

          

            
            

            

                  end
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

                # :nodoc:
          
        
      
        

          

            1
            

            

                class LogAndRaiseErrorHandler < LogErrorHandler
          
        
      
        

          

            1
            

            

                  def call(error, info, properties, payload)
          
        
      
        

          

            2
            

            

                    super(error, info, properties, payload)
          
        
      
        

          

            2
            

            

                    raise error
          
        
      
        

          

            
            

            

                  end
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

              end
          
        
      
        

          

            
            

            

            end
          
        
      
    
  

    

      
        

          

            
            

            

            # frozen_string_literal: true
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

            require 'context_request_subscriber/handler/json_api_handler'
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

            # :nodoc:
          
        
      
        

          

            1
            

            

            module Handler
          
        
      
        

          

            
            

            

            end
          
        
      
    
  

    

      
        

          

            
            

            

            # frozen_string_literal: true
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

            require 'json_api_client'
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

            module ContextRequestSubscriber
          
        
      
        

          

            1
            

            

              module Handler
          
        
      
        

          

            1
            

            

                module JsonApiHandler
          
        
      
        

          

            
            

            

                  # :nodoc:
          
        
      
        

          

            1
            

            

                  class Base < JsonApiClient::Resource
          
        
      
        

          

            1
            

            

                    def initialize(params, **keys)
          
        
      
        

          

            2
            

            

                      @headers = keys.fetch(:handler_headers, {})
          
        
      
        

          

            2
            

            

                      self.class.site = keys.fetch(:site, nil)
          
        
      
        

          

            2
            

            

                      super(params)
          
        
      
        

          

            
            

            

                    end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                    def call
          
        
      
        

          

            1
            

            

                      self.class.with_headers(@headers) do
          
        
      
        

          

            1
            

            

                        save
          
        
      
        

          

            
            

            

                      end
          
        
      
        

          

            
            

            

                    end
          
        
      
        

          

            
            

            

                  end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

                  # :nodoc:
          
        
      
        

          

            1
            

            

                  class Context < Base; end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

                  # :nodoc:
          
        
      
        

          

            1
            

            

                  class Request < Base; end
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

              end
          
        
      
        

          

            
            

            

            end
          
        
      
    
  

    

      
        

          

            
            

            

            # frozen_string_literal: true
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

            module ContextRequestSubscriber
          
        
      
        

          

            
            

            

              # :nodoc:
          
        
      
        

          

            1
            

            

              module Processor
          
        
      
        

          

            1
            

            

                autoload :Base, 'context_request_subscriber/processor/base'
          
        
      
        

          

            1
            

            

                autoload :Request, 'context_request_subscriber/processor/request'
          
        
      
        

          

            1
            

            

                autoload :Context, 'context_request_subscriber/processor/context'
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

                # Class to execute the processing of the messages.
          
        
      
        

          

            
            

            

                # Base on the message_type a processor is instantiated and provides
          
        
      
        

          

            
            

            

                # processing.
          
        
      
        

          

            1
            

            

                class Executor
          
        
      
        

          

            1
            

            

                  attr_accessor :handler_params
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                  def initialize(logger, **keys)
          
        
      
        

          

            8
            

            

                    @logger = logger
          
        
      
        

          

            8
            

            

                    @handler_params = keys[:handler_params] || {}
          
        
      
        

          

            
            

            

                  end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                  def run(_delivery_info, properties, payload)
          
        
      
        

          

            8
            

            

                    @logger.debug("#{properties[:type]} processing message.")
          
        
      
        

          

            8
            

            

                    @logger.debug("   payload: #{payload}")
          
        
      
        

          

            8
            

            

                    @logger.debug("   properties: #{properties}")
          
        
      
        

          

            8
            

            

                    payload = JSON.parse(payload)
          
        
      
        

          

            7
            

            

                    process(properties[:type], properties[:headers]&.dig('version'),
          
        
      
        

          

            
            

            

                            payload)
          
        
      
        

          

            
            

            

                  rescue JSON::ParserError
          
        
      
        

          

            1
            

            

                    @logger.error("Could not parse message payload \
          
        
      
        

          

            
            

            

            with payload #{payload}. Ignoring.")
          
        
      
        

          

            1
            

            

                    nil
          
        
      
        

          

            
            

            

                  end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                  def process(name, version = nil, payload = {})
          
        
      
        

          

            7
            

            

                    processor = Processor.processor_class(name, version)
          
        
      
        

          

            
            

            

                      &.new(@logger, handler_params: handler_params)
          
        
      
        

          

            7
            

            

                    if processor
          
        
      
        

          

            6
            

            

                      processor.call(payload)
          
        
      
        

          

            
            

            

                    else
          
        
      
        

          

            1
            

            

                      @logger.error("Invalid processor type or \
          
        
      
        

          

            
            

            

            processor type with name #{name} cannot be found")
          
        
      
        

          

            
            

            

                    end
          
        
      
        

          

            
            

            

                  end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

                  # Method can be overwriten to communicate with the MQ and ack the message
          
        
      
        

          

            
            

            

                  # if processing was successful.
          
        
      
        

          

            
            

            

                  # Default is to do nothing because the setup is auto_ack.
          
        
      
        

          

            1
            

            

                  def ack(_channel); end
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                def self.processor_class(name, version = nil)
          
        
      
        

          

            7
            

            

                  version = "V#{version}" if version
          
        
      
        

          

            
            

            

                  [self.name, version, name.tr('.', '_').camelize].compact.join('::')
          
        
      
        

          

            7
            

            

                                                                  .constantize
          
        
      
        

          

            
            

            

                rescue NameError
          
        
      
        

          

            1
            

            

                  nil
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

              end
          
        
      
        

          

            
            

            

            end
          
        
      
    
  

    

      
        

          

            
            

            

            # frozen_string_literal: true
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

            module ContextRequestSubscriber
          
        
      
        

          

            1
            

            

              module Processor
          
        
      
        

          

            
            

            

                # :nodoc:
          
        
      
        

          

            1
            

            

                class Base
          
        
      
        

          

            1
            

            

                  def initialize(logger, **keys)
          
        
      
        

          

            8
            

            

                    @logger = logger
          
        
      
        

          

            8
            

            

                    @handler_params = keys[:handler_params] || {}
          
        
      
        

          

            
            

            

                  end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                  def call(payload)
          
        
      
        

          

            7
            

            

                    if (handler = ContextRequestSubscriber.handlers[type_name])
          
        
      
        

          

            2
            

            

                      handler.new(payload, **@handler_params).call
          
        
      
        

          

            
            

            

                    else
          
        
      
        

          

            5
            

            

                      @logger.error("Could not find handler for \
          
        
      
        

          

            
            

            

            message type #{type_name}")
          
        
      
        

          

            
            

            

                    end
          
        
      
        

          

            
            

            

                  end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                  private
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                  def type_name
          
        
      
        

          

            12
            

            

                    self.class.name.split('::').last.underscore.downcase.to_sym
          
        
      
        

          

            
            

            

                  end
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

              end
          
        
      
        

          

            
            

            

            end
          
        
      
    
  

    

      
        

          

            
            

            

            # frozen_string_literal: true
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

            module ContextRequestSubscriber
          
        
      
        

          

            1
            

            

              module Processor
          
        
      
        

          

            1
            

            

                class Context < Base
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

              end
          
        
      
        

          

            
            

            

            end
          
        
      
    
  

    

      
        

          

            
            

            

            # frozen_string_literal: true
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

            module ContextRequestSubscriber
          
        
      
        

          

            1
            

            

              module Processor
          
        
      
        

          

            1
            

            

                class Request < Base
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

              end
          
        
      
        

          

            
            

            

            end
          
        
      
    
  

    

      
        

          

            
            

            

            # frozen_string_literal: true
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

            require 'bunny'
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

            module ContextRequestSubscriber
          
        
      
        

          

            
            

            

              # The subscriber to attach to the exchange to process the
          
        
      
        

          

            
            

            

              # ContextRequest data.
          
        
      
        

          

            
            

            

              #
          
        
      
        

          

            
            

            

              # ErrorHandler is a Callable object that gets called for any exception
          
        
      
        

          

            
            

            

              # during the execution of the processor.
          
        
      
        

          

            
            

            

              # The call method gets passed the following information:
          
        
      
        

          

            
            

            

              #   @exception the exception caught
          
        
      
        

          

            
            

            

              #   @delivery_info the delivery_info object
          
        
      
        

          

            
            

            

              #   @properties properties passed in the message (headers, ..)
          
        
      
        

          

            
            

            

              #   @payload the payload of the message
          
        
      
        

          

            
            

            

              # The ErrorHandler has to decide what to do with the message. If none
          
        
      
        

          

            
            

            

              # is given the error will be ignored!
          
        
      
        

          

            1
            

            

              class RabbitMQSubscriber
          
        
      
        

          

            1
            

            

                DEFAULT_QUEUE_OPTS = {
          
        
      
        

          

            
            

            

                  exclusive: false, durable: true
          
        
      
        

          

            
            

            

                }.freeze
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                DEFAULT_SESSION_PARAMS = {
          
        
      
        

          

            
            

            

                  threaded: true, automatically_recover: true
          
        
      
        

          

            
            

            

                }.freeze
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

                # @processor the processor to be used to handle the data.
          
        
      
        

          

            
            

            

                # @config the configurables supported as follows
          
        
      
        

          

            
            

            

                #    session_params: set of session params
          
        
      
        

          

            
            

            

                #    url: url to connect to the RabbitMQ Server
          
        
      
        

          

            
            

            

                #    heartbeat: heartbeat for the connection. Defaults to nil.
          
        
      
        

          

            
            

            

                #    exchange_name: the exchange connected to the queue
          
        
      
        

          

            
            

            

                #    exchange_options: the options used to create the exchange if not
          
        
      
        

          

            
            

            

                #      existing.
          
        
      
        

          

            
            

            

                #    exchange_type: type of the exchange, defaults to topic.
          
        
      
        

          

            
            

            

                #    queue_name: the queue where the data is received from.
          
        
      
        

          

            
            

            

                #    queue_durable: if the queue is durable. Default: true
          
        
      
        

          

            
            

            

                #    queue_auto_delete: if the queue gets autodeleted.
          
        
      
        

          

            
            

            

                #    queue_exclusive: default false.
          
        
      
        

          

            
            

            

                #    routing_key: the routing_key used. Default nil.
          
        
      
        

          

            
            

            

                #    on_error: callable object that handles errors during processing the
          
        
      
        

          

            
            

            

                #       message.
          
        
      
        

          

            1
            

            

                def initialize(**config)
          
        
      
        

          

            5
            

            

                  @error_handler = config[:on_error].new(config[:logger])
          
        
      
        

          

            5
            

            

                  @logger = config[:logger]
          
        
      
        

          

            5
            

            

                  @executor = Processor::Executor.new(config[:logger],
          
        
      
        

          

            
            

            

                                                      config.slice(:handler_params))
          
        
      
        

          

            5
            

            

                  config_connection(config)
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            5
            

            

                  config_exchange(config)
          
        
      
        

          

            5
            

            

                  config_queue(config)
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                def run
          
        
      
        

          

            5
            

            

                  exchange, queue = setup_queue
          
        
      
        

          

            5
            

            

                  @consumer = queue.subscribe(manual_ack: true,
          
        
      
        

          

            
            

            

                                              block: false) do |info, properties, payload|
          
        
      
        

          

            5
            

            

                    consume(info, properties, payload)
          
        
      
        

          

            
            

            

                  end
          
        
      
        

          

            4
            

            

                  join(exchange)
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                def join(exchange)
          
        
      
        

          

            4
            

            

                  @join_workpool && exchange.channel.work_pool.join
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                private
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                def config_connection(config)
          
        
      
        

          

            5
            

            

                  @session_params = config.fetch(:session_params, {})
          
        
      
        

          

            
            

            

                                          .merge(DEFAULT_SESSION_PARAMS)
          
        
      
        

          

            5
            

            

                  @url = config[:url]
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                def config_queue(config)
          
        
      
        

          

            5
            

            

                  @queue_name = config[:queue_name]
          
        
      
        

          

            5
            

            

                  @queue_bindings = config.slice(:routing_key)
          
        
      
        

          

            5
            

            

                  @queue_opts = queue_opts(config)
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                def config_exchange(config)
          
        
      
        

          

            5
            

            

                  @exchange_type = config.fetch(:exchange_type, 'topic')
          
        
      
        

          

            5
            

            

                  @exchange_name = config.fetch(:exchange_name,
          
        
      
        

          

            
            

            

                                                ContextRequestSubscriber.exchange_name)
          
        
      
        

          

            5
            

            

                  @exchange_opts = config.fetch(:exchange_options, {})
          
        
      
        

          

            5
            

            

                  @join_workpool = config.fetch(:subscriber_join, true)
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

                # setup the whole message queue settings and return the queue object.
          
        
      
        

          

            
            

            

                # Can be overwriten by ContextRequestMiddleware.fetch_exchange_callback
          
        
      
        

          

            1
            

            

                def setup_queue
          
        
      
        

          

            5
            

            

                  callback = ContextRequestSubscriber.fetch_queue_callback
          
        
      
        

          

            5
            

            

                  if callback
          
        
      
        

          

            1
            

            

                    exchange, queue = callback.call(self)
          
        
      
        

          

            
            

            

                  else
          
        
      
        

          

            4
            

            

                    channel = create_channel(@session_params)
          
        
      
        

          

            4
            

            

                    exchange = fetch_exchange(channel, @exchange_type, @exchange_name,
          
        
      
        

          

            
            

            

                                              @exchange_opts)
          
        
      
        

          

            4
            

            

                    queue = bind_queue(channel, exchange)
          
        
      
        

          

            
            

            

                  end
          
        
      
        

          

            5
            

            

                  [exchange, queue]
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

                # Default behaviour is that all messages are consumed by just
          
        
      
        

          

            
            

            

                # passing the information to the processor. If any part fails
          
        
      
        

          

            
            

            

                # the application needs to gracefully handle because the message
          
        
      
        

          

            
            

            

                # is acked automatically.
          
        
      
        

          

            
            

            

                # To change that behaviour overwrite the fetch_queue_callback
          
        
      
        

          

            
            

            

                # and extend the Processor to handle the ack method.
          
        
      
        

          

            1
            

            

                def consume(info, properties, payload)
          
        
      
        

          

            5
            

            

                  @executor.run(info, properties, payload)
          
        
      
        

          

            4
            

            

                  @executor.ack(@channel)
          
        
      
        

          

            
            

            

                rescue StandardError => e
          
        
      
        

          

            1
            

            

                  handle_error(e, info, properties, payload)
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                def create_channel(options = {})
          
        
      
        

          

            4
            

            

                  connection = Bunny.new(@url, options)
          
        
      
        

          

            4
            

            

                  connection.start
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            4
            

            

                  connection.create_channel
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            
            

            

                # return the exchange Bunny object and do the whole setup around it.
          
        
      
        

          

            1
            

            

                def fetch_exchange(channel, exchange_type, exchange_name, exchange_opts)
          
        
      
        

          

            4
            

            

                  channel.exchanges[exchange_name] ||
          
        
      
        

          

            
            

            

                    bunny_exchange(channel, exchange_type, exchange_name, exchange_opts)
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                def bind_queue(channel, exchange)
          
        
      
        

          

            4
            

            

                  queue = channel.queue(@queue_name, @queue_opts)
          
        
      
        

          

            4
            

            

                  queue.bind(exchange, @queue_bindings)
          
        
      
        

          

            4
            

            

                  queue
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                def handle_error(error, info, properties, payload)
          
        
      
        

          

            1
            

            

                  @error_handler&.call(error, info, properties, payload)
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                def bunny_exchange(channel, type, name, opts)
          
        
      
        

          

            1
            

            

                  Bunny::Exchange.new(channel, type, name, opts)
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

            
          
        
      
        

          

            1
            

            

                def queue_opts(config)
          
        
      
        

          

            5
            

            

                  opts = config.slice(:queue_durable, :queue_auto_delete, :queue_exclusive)
          
        
      
        

          

            5
            

            

                  opts.dup.each do |k, _v|
          
        
      
        

          

            5
            

            

                    opts[k.to_s.sub(/^queue_/, '').to_sym] = opts.delete(k)
          
        
      
        

          

            
            

            

                  end
          
        
      
        

          

            5
            

            

                  DEFAULT_QUEUE_OPTS.merge(opts)
          
        
      
        

          

            
            

            

                end
          
        
      
        

          

            
            

            

              end
          
        
      
        

          

            
            

            

            end