Method process_message has a Cognitive Complexity of 90 (exceeds 5 allowed). Consider refactoring.

Open

      def process_message(message, options = {})
        unless message.is_a?(Hash)
          message =
            begin
              JSON.parse(message)

• Read up
  Read up

Method enqueue has a Cognitive Complexity of 54 (exceeds 5 allowed). Consider refactoring.

Open

      def enqueue(message, &block)
        message = message.with_indifferent_access
        task_description = message[:task_description]
        auto_retry = message[:auto_retry].presence || Setup::Task.auto_retry_enum.first
        scheduler = message.delete(:scheduler)

• Read up
  Read up

Method start_consumer has a Cognitive Complexity of 31 (exceeds 5 allowed). Consider refactoring.

Open

      def start_consumer
        if init
          new_rabbit_consumer = RabbitConsumer.create(channel: "#{connection.host}:#{connection.local_port} (#{channel.id})",
                                                      tag: channel.generate_consumer_tag(Cenit.rabbit_mq_queue))
          new_consumer = queue.subscribe(consumer_tag: new_rabbit_consumer.tag, manual_ack: true) do |delivery_info, properties, body|

• Read up
  Read up

Method lookup_messages has a Cognitive Complexity of 30 (exceeds 5 allowed). Consider refactoring.

Open

      def lookup_messages(opts = {})
        channel_mutex.lock
        if channel && !channel.closed?
          dispatched_ids = []
          tenant_tasks = {}

• Read up
  Read up

File rabbit.rb has 349 lines of code (exceeds 250 allowed). Consider refactoring.

Open

require 'json'
require 'openssl'
require 'bunny'

module Cenit

Method init has a Cognitive Complexity of 25 (exceeds 5 allowed). Consider refactoring.

Open

      def init
        channel_mutex.lock
        if ENV['SKIP_RABBIT_MQ'].to_b
          puts 'RabbitMQ SKIPPED'
          false

• Read up
  Read up

Method process_message has 74 lines of code (exceeds 25 allowed). Consider refactoring.

Open

      def process_message(message, options = {})
        unless message.is_a?(Hash)
          message =
            begin
              JSON.parse(message)

Method enqueue has 69 lines of code (exceeds 25 allowed). Consider refactoring.

Open

      def enqueue(message, &block)
        message = message.with_indifferent_access
        task_description = message[:task_description]
        auto_retry = message[:auto_retry].presence || Setup::Task.auto_retry_enum.first
        scheduler = message.delete(:scheduler)

Method lookup_messages has 49 lines of code (exceeds 25 allowed). Consider refactoring.

Open

      def lookup_messages(opts = {})
        channel_mutex.lock
        if channel && !channel.closed?
          dispatched_ids = []
          tenant_tasks = {}

Method start_consumer has 45 lines of code (exceeds 25 allowed). Consider refactoring.

Open

      def start_consumer
        if init
          new_rabbit_consumer = RabbitConsumer.create(channel: "#{connection.host}:#{connection.local_port} (#{channel.id})",
                                                      tag: channel.generate_consumer_tag(Cenit.rabbit_mq_queue))
          new_consumer = queue.subscribe(consumer_tag: new_rabbit_consumer.tag, manual_ack: true) do |delivery_info, properties, body|

Method detask has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

      def detask(message)
        report = nil
        case task = message.delete(:task)
        when Class
          task_class = task

• Read up
  Read up

Consider simplifying this complex logical expression.

Open

              if scheduler || publish_at || async_message
                if (token = message[:token])
                  TaskToken.where(token: token).delete_all
                end
                message[:task_id] = task.id.to_s

Method init has 32 lines of code (exceeds 25 allowed). Consider refactoring.

Open

      def init
        channel_mutex.lock
        if ENV['SKIP_RABBIT_MQ'].to_b
          puts 'RabbitMQ SKIPPED'
          false

Method detask has 27 lines of code (exceeds 25 allowed). Consider refactoring.

Open

      def detask(message)
        report = nil
        case task = message.delete(:task)
        when Class
          task_class = task

Avoid deeply nested control flow statements.

Open

                if (token = message[:token])
                  TaskToken.where(token: token).delete_all
                end

Avoid deeply nested control flow statements.

Open

                if delayed
                  Setup::DelayedMessage.create(message: message, publish_at: publish_at, scheduler: scheduler)
                else
                  begin
                    channel_mutex.lock

Avoid deeply nested control flow statements.

Open

                    if task
                      if (rabbit_consumer = options[:rabbit_consumer] || RabbitConsumer.where(tag: options[:consumer_tag]).first)
                        rabbit_consumer.update(executor_id: tenant.id, task_id: task.id)
                      end
                      task.execute(execution_id: execution_id)

Avoid deeply nested control flow statements.

Open

                  if (resume_interval = task.resume_interval)
                    message[:publish_at] = Time.now + resume_interval
                  end

Avoid deeply nested control flow statements.

Open

                  if task
                    task.notify(ex)
                  else
                    Setup::SystemNotification.create(message: "Can not execute task for message: #{message}")
                  end