Implementation on AMQP in Ruby

Question

I need comments on the below code:

Thread.new {EM.run do
              IpamAgent::Amqp.run
            end}



module IpamAgent
  class Amqp
    class << self
      def run
        begin
          $connection = AMQP.connect(RMQ_CONFIGURATIONS)
          $connection.on_tcp_connection_loss do |conn, settings|
            Rails.logger.info  "<<<<<<<<<<<<<<<< [network failure] Trying to reconnect...>>>>>>>>>>>>>>>>>>>>>>>>"
            conn.reconnect(false, 2)
          end
          Rails.logger.info "<<<<<<<<<<<<<<<<<<<<<<<<AMQP listening>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>"
          worker   = IpamAgent::MessageHandler.new
          worker.start
          Rails.logger.info "<<<<<<<<<<<<<<<<<<<<<<<<Message handler started>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>"

        rescue Exception => e
          Rails.logger.info "<<<<<<<<<<<<<<<<<<<<<<<<Message handler Exception>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>"
          Rails.logger.info "[error] Could not handle event of type #{e.inspect}"
          Rails.logger.info "<<<<<<<<<<<<<<<<<<<<<<<<Message handler Exception>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>"
        end
      end
    end
  end
end




    module IpamAgent

      class MessageHandler
        attr_accessor :ns_exchange, :sc_exchange, :ns_queue, :sc_queue, :service_location, :service_version, :service_name, :message_sequence_id, :presto_exchange, :presto_queue

        def initialize
          # Profile the code
          @ns_exchange = CONFIGURATIONS["ns_exchange"]
          @sc_exchange = CONFIGURATIONS["sc_exchange"]
          @presto_exchange =  CONFIGURATIONS["presto_exchange"]

          @ns_queue = CONFIGURATIONS["ns_queue"]
          @sc_queue = CONFIGURATIONS["sc_queue"]
          @presto_queue =  CONFIGURATIONS["presto_queue"]
        end

        # Create the channels, exchanges and queues
        def start
          ch1  = AMQP::Channel.new($connection)
          ch2  = AMQP::Channel.new($connection)
          ch3  = AMQP::Channel.new($connection)

          @ns_x = ch1.direct(ns_exchange, :durable => true)
          @ns_queue   = ch1.queue(ns_queue, :auto_delete => false)
          @ns_queue.bind(@ns_x, :routing_key => @ns_queue.name).subscribe(:ack => true, &method(:handle_ns_message))

          @sc_x = ch2.topic(sc_exchange, :durable => true)
          @sc_queue   = ch2.queue(sc_queue, :auto_delete => false)
          @sc_queue.bind(@sc_x, :routing_key => "#").subscribe(:ack => true, &method(:handle_sc_message))

          @presto_x = ch3.direct(presto_exchange, :durable => true)
          @presto_queue   = ch3.queue(presto_queue, :auto_delete => false)
          @presto_queue.bind(@presto_x, :routing_key => @presto_queue.name).subscribe(:ack => true, &method(:handle_presto_message))
        end

        # Handle the messages from Network service component
        def handle_ns_message(headers, payload)
          message_headers = JSON.parse(headers.to_json)["headers"]
          payload = eval(payload)
          headers.ack
          Rails.logger.info ">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>MESSAGE FROM NS<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<"
          Rails.logger.info message_headers
          Rails.logger.info payload
          Rails.logger.info ">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>MESSAGE FROM NS<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<"

          tenant_detail = IpamAgent::TenantDetail.where(service_instance: payload["orgId"]).first
          if(payload && payload.keys.include?(:responseCode))
            new_tenant_detail = IpamAgent::TenantDetail.create(message: ({"header" => message_headers, "payload" => payload}), status: "waiting_for_sc", service_instance: payload["orgId"])
            if tenant_detail && tenant_detail.service_group
              publish_sgid_to_presto(tenant_detail)
            else
              get_sgid_from_sc(new_tenant_detail)
            end
          else
            Rails.logger.info("Payload: #{payload}, routing key is #{message_headers}")
          end
        end

        # Retrieve the Service Group ID from Service controller
        def get_sgid_from_sc(tenant_detail)
          message = tenant_detail.get_sc_message
          Rails.logger.info(">>>>>>>>>>>>>>>>>>>>PUBLISHING TO SC<<<<<<<<<<<<<<<<<<<<<<<")
          Rails.logger.info(message)
          Rails.logger.info(">>>>>>>>>>>>>>>>>>>>PUBLISHING TO SC<<<<<<<<<<<<<<<<<<<<<<<")
          @sc_x.publish(message.last,  :routing_key => @sc_queue.name, :headers => message.first, :mandatory => true)
        end

        # Handle the messages from Service controller
        def handle_sc_message(headers, payload)
          message_headers = JSON.parse(headers.to_json)
          headers.ack
          payload = eval(payload)
          Rails.logger.info(">>>>>>>>>>>>>>>>>>>>MESSAGE FROM SC<<<<<<<<<<<<<<<<<<<<<<<")
          Rails.logger.info(message_headers)
          Rails.logger.info(payload)
          Rails.logger.info(">>>>>>>>>>>>>>>>>>>>MESSAGE FROM SC<<<<<<<<<<<<<<<<<<<<<<<")
          if(payload && payload["serviceInstanceGroupId"])
            tenant_detail = IpamAgent::TenantDetail.find_or_save(payload)
            publish_sgid_to_presto(tenant_detail)
          end
        end

        # Shovel the NS request to PMP with Service Group ID
        def publish_sgid_to_presto(tenant_detail)
          tenant_details = TenantDetail.where(service_instance: tenant_detail.service_instance, status: "waiting_for_sc")
          tenant_details.each do |sc|
            sc.update(status: "success")
            message = sc.get_pmp_message
            Rails.logger.info(">>>>>>>>>>>>>>>>>>>>PUBLISHING TO PRESTO<<<<<<<<<<<<<<<<<<<<<<<")
            Rails.logger.info(message.first)
            Rails.logger.info(message.last)
            Rails.logger.info(">>>>>>>>>>>>>>>>>>>>PUBLISHING TO PRESTO<<<<<<<<<<<<<<<<<<<<<<<")
            @presto_x.publish(message.last,  :routing_key => @presto_queue.name, :headers => message.first, :mandatory => true)
          end
        end

        # Receive the message from PMP presto and publsih it to Network Service
        def handle_presto_message(headers, payload)
          message_headers = JSON.parse(headers.to_json)
          payload = eval(payload)
          Rails.logger.info(">>>>>>>>>>>>>>>>>>>>MESSAGE FROM PRESTO<<<<<<<<<<<<<<<<<<<<<<<")
          Rails.logger.info(headers.to_json)
          Rails.logger.info(payload)
          Rails.logger.info(">>>>>>>>>>>>>>>>>>>>MESSAGE FROM PRESTO<<<<<<<<<<<<<<<<<<<<<<<")
          headers.ack
          @ns_x.publish(payload,  :routing_key =>@ns_queue.name, :headers => headers, :mandatory => true) if (message_headers && payload)
        end
      end

    end

Answer 1

Don't use global variables
Using $connection as a holder for your AMQP connection is a bad idea. connection being such a generic name, it might be used somewhere else for something completely different, and you may end up breaking your code with someone else's.

A better idea would be to use class variables - this way you get your namespace for free:

module IpamAgent
  class Amqp
    class << self
      attr_reader connection

      def run
        begin
          @connection = AMQP.connect(RMQ_CONFIGURATIONS)
          connection.on_tcp_connection_loss do |conn, settings|
            # ...
end

And now your usage will be:

def start
  ch1 = AMQP::Channel.new(IpamAgent::Amqp.connection)
  # ...
end

Don't change the meaning of a variable
When you set a variable with one type, for one usage, but then set the same variable with another type, for a different usage - you are confusing yourself, and future code-readers, and make your code very brittle. This goes double for instance variables, and triple for instance variables exposed to the outside - what do you think a user will expect from this code:

  worker = IpamAgent::MessageHandler.new
  queue = worker.ns_queue
  worker.start
  same_queue = worker.ns_queue

Avoid noisy code, and noisy logs
Your code is filled with < and > characters, which are meant to highlight sections in your log files, but since it is used so much, your log file will be filled with marquees, and will be painful to read (just like you code right now, only much worse).

You should differentiate between events in the log using log levels - exceptions and errors should be logged using Rails.logger.error, production-level events should be logged using Rails.logger.info. All other events and debug data should be logged only using Rails.logger.debug, and maybe not at all.

Variable naming
Using names such as ch, ns, sc, etc. is very discouraged, unless it is part of your domain's nomenclature. For example, ns for a lot of people coming from XML, means namespace. Be verbal - use full names (network_service).

If it needs a comment - you probably have to rename it or refactor it
You have many methods with comments explaining what they do. Comments are generally a bad idea, since they tend to "rot" - when you change the code, most often than not, you neglect to maintain the comments (For example what does # Profile the code refer to?).

There are two types of method comments in your code - those which repeat the name of the method, and don't add much (handle_network_service_message is enough, you don't need to say # Handle the messages from Network service component...), and those which explain things which are not explained by the method name.

Delete the ones which repeat the name of the method.

The other kind hints that you probably need to either rename the method, or break it down to its parts. You could either rename start to create_channels, or (probably better) have a create_channel method which start will call 3 times - it will make your code DRYier.