Enable consumers to spawn more than one processes

Gemfile.lock

@@ -1,8 +1,9 @@
 PATH
   remote: .
   specs:
-    sevak (0.5.3)
+    sevak (0.6.0)
       bunny (~> 2.7)
+      highline
 
 GEM
   remote: https://rubygems.org/

README.md

@@ -22,7 +22,7 @@ To install this plugin:
 # Installation
 
     gem install sevak
-    
+
 # Make sure the rabbitmq server is running
 
 You can either install and run the rabbitmq server from the appropriate package for your os or you can run the preconfigured docker image for local testing.
@@ -31,7 +31,7 @@ The image can be found here.
 
 https://hub.docker.com/r/deepakkumarnd/sevak/
 
-You can run the rabbitmq by doing the following step 
+You can run the rabbitmq by doing the following step
 
     docker pull deepakkumarnd/sevak
     docker run -d --name rabbitmq_test -p 15672:15672 -p 5672:5672 deepakkumarnd/sevak
@@ -98,3 +98,13 @@ https://hub.docker.com/r/deepakkumarnd/sevak/
 
 docker pull deepakkumarnd/sevak
 docker run -d --name rabbitmq_test -p 15672:15672 -p 5672:5672 deepakkumarnd/sevak
+
+### Enable consumers to spawn more than one processes
+
+We can also run the consumers as follows:
+
+$ RAILS_ENV={env} REPLICA={n} bundle exec rake consumer_name
+
+Where n is a number, the above command should spin up n number of consumer process.
+
+In effect REPLICA={n} should fork the consumer n times.

Rakefile

@@ -1,6 +1,5 @@
 require 'bundler/gem_tasks'
 require 'sevak'
-require 'highline'
 
 # run tests by running command `rake test`
 require 'rake/testtask'
@@ -14,72 +13,5 @@ end
 
 # Load all rake tasks in the tasks folder
 
-tasks = FileList["tasks/*.rake"]
-
+tasks = FileList["tasks/**/*.rake"]
 tasks.each { |task| load(task) }
-
-namespace :consumer do
-
-  desc 'Interactively create a new consumer'
-  task :new do
-
-    cli = HighLine.new
-
-    consumer_name = cli.ask('Input consumer name(eg. push_alert) :') { |q| q.validate = /\A[a-z]+[a-z_]+[a-z]\z/}
-    queue_name    = cli.ask('Input queue name(eg. myqueue)') { |q| q.validate = /\A[a-z]+[a-z.]+[a-z]\z/}
-
-    to_class_name = Proc.new do |name|
-      name.split('_').map(&:capitalize).join
-    end
-
-text1 = <<-CODE
-
-module Sevak
-
-  class #{to_class_name.call(consumer_name)}Consumer < Consumer
-
-    queue_name "#{queue_name}"
-
-    def run(payload)
-      puts "Consumer running"
-    end
-  end
-
-end
-
-CODE
-
-    if File.exists?("app/consumers/#{consumer_name}_consumer.rb")
-      puts "Already exists file: app/consumers/#{consumer_name}_consumer.rb"
-      ans = cli.ask('Do you want to overwrite it ? (y/n)') { |q| q.validate = /Y|N|y|n/}
-      exit(-1) if ans.downcase == 'n'
-    end
-
-    file = File.open("app/consumers/#{consumer_name}_consumer.rb", 'w+')
-    file.write(text1)
-    file.close
-
-text2 = <<-CODE
-namespace :sevak do
-
-  desc "Run the #{consumer_name} worker"
-  task :#{consumer_name} do
-    consumer = Sevak::#{to_class_name.call(consumer_name)}Consumer.new
-    consumer.start
-  end
-end
-
-CODE
-
-    if File.exists?("lib/tasks/#{consumer_name}.rake")
-      puts "Already exists file: lib/tasks/#{consumer_name}.rake"
-      ans = cli.ask('Do you want to overwrite it ? (y/n)') { |q| q.validate = /Y|N|y|n/}
-      exit(-1) if ans.downcase == 'n'
-    end
-
-    file = File.open("lib/tasks/#{consumer_name}.rake", 'w+')
-    file.write(text2)
-    file.close
-
-  end
-end

lib/sevak.rb

@@ -115,3 +115,4 @@ def self.root
 require 'sevak/autoscale'
 require 'sevak/consumer'
 require 'sevak/publisher'
+require 'sevak/railtie' if defined?(Rails)

lib/sevak/autoscale.rb

@@ -26,7 +26,6 @@ def fork_process
         end
 
         pids.push(pid)
-
       rescue => e
         log("Unable to fork process #{e.message}")
       end
@@ -75,8 +74,20 @@ def calculate_average_load
       end
     end
 
+    def replicas_required
+      ENV['REPLICA']
+    end
+
     def start_master_worker
-      fork_process
+      if replicas_required
+        replicas_required.to_i.times do
+          fork_process
+        end
+
+        Process.waitall
+      else
+        fork_process
+      end
 
       loop do
         avg_load = calculate_average_load
@@ -90,7 +101,7 @@ def start_master_worker
         end
 
         sleep 5
-      end
+      end unless ENV['REPLICA']
     rescue => e
       cleanup
     ensure

lib/sevak/consumer.rb

@@ -4,16 +4,15 @@ module Sevak
   # run method. The run method should implement the business logic.
 
   class ConsumerBase
-
     include Core
     include Autoscale
 
     DEFAULT_PREFETCH_COUNT = 10
 
     # class methods
-      def self.queue_name(name='default')
-        @queue_name ||= name
-      end
+    def self.queue_name(name='default')
+      @queue_name ||= name
+    end
     # end of class methods
 
     def initialize
@@ -37,22 +36,21 @@ def message_count
     end
 
     def start
-      if config.autoscale
+      if config.autoscale || replicas_required
         start_master_worker
       else
         start_worker
       end
     end
 
     def start_worker
+      load_rails_environment if defined?(Rails)
+
       channel.prefetch(config.prefetch_count || DEFAULT_PREFETCH_COUNT)
 
       queue.subscribe(manual_ack: true, exclusive: false) do |delivery_info, metadata, payload|
         body = JSON.parse(payload)
 
-        # p delivery_info
-        # p metadata
-
         begin
           status = run(body)
         rescue => ex
@@ -90,10 +88,14 @@ def exception_details(e, payload = nil)
       "Sevak Exception: #{msg}"
     end
 
+    private
+
+    def load_rails_environment
+      require './config/environment'
+    end
   end
 
   class Consumer < ConsumerBase
-
     # Set the queue name for the consumer
     queue_name 'sevak.default'
 
@@ -104,4 +106,4 @@ def run(payload)
       :ok
     end
   end
-end
+end

lib/sevak/publisher.rb

@@ -30,21 +30,21 @@ def publish_exchange(queue_name, message, delay)
       exchange(queue_name).publish(message.to_json, headers: { 'x-delay' => delay })
     end
 
-  private
-
-      def self.publish_message(queue_name, message, delay = nil)
-        attempt = 0
-        begin
-          if delay.nil? 
-            instance.queue(queue_name).publish(message.to_json)
-          else
-            instance.publish_exchange(queue_name, message, delay.to_i)
-          end
-        rescue Bunny::ConnectionClosedError => e
-          attempt += 1
-          sleep(0.001)
-          attempt < 2 ? retry : raise
+    private
+
+    def self.publish_message(queue_name, message, delay = nil)
+      attempt = 0
+      begin
+        if delay.nil?
+          instance.queue(queue_name).publish(message.to_json)
+        else
+          instance.publish_exchange(queue_name, message, delay.to_i)
         end
+      rescue Bunny::ConnectionClosedError => e
+        attempt += 1
+        sleep(0.001)
+        attempt < 2 ? retry : raise
       end
+    end
   end
 end

lib/sevak/railtie.rb

@@ -0,0 +1,13 @@
+require 'rails'
+
+module Sevak
+  class Railtie < Rails::Railtie
+    railtie_name :sevak
+
+    rake_tasks do
+      path = File.join(File.expand_path('../..', __dir__), 'tasks/**/*.rake')
+      tasks = FileList[path]
+      tasks.each { |task| load(task) }
+    end
+  end
+end

lib/sevak/version.rb

@@ -1,3 +1,3 @@
 module Sevak
-  VERSION = '0.5.3'
+  VERSION = '0.6.0'
 end

sevak.gemspec

@@ -22,4 +22,5 @@ Gem::Specification.new do |spec|
 
   # spec.add_development_dependency 'bundler', '~> 1.9'
   # spec.add_development_dependency 'rake', '~> 10.0'
+  spec.add_runtime_dependency 'highline'
 end

tasks/consumer/new.rake

@@ -0,0 +1,69 @@
+require 'highline'
+
+namespace :consumer do
+
+  desc 'Interactively create a new consumer'
+  task :new do
+
+    cli = HighLine.new
+
+    consumer_name = cli.ask('Input consumer name(eg. push_alert) :') { |q| q.validate = /\A[a-z]+[a-z_]+[a-z]\z/}
+    queue_name    = cli.ask('Input queue name(eg. myqueue)') { |q| q.validate = /\A[a-z]+[a-z.]+[a-z]\z/}
+
+    to_class_name = Proc.new do |name|
+      name.split('_').map(&:capitalize).join
+    end
+
+text1 = <<-CODE
+
+module Sevak
+
+  class #{to_class_name.call(consumer_name)}Consumer < Consumer
+
+    queue_name "#{queue_name}"
+
+    def run(payload)
+      puts "Consumer running"
+    end
+  end
+
+end
+
+CODE
+
+    if File.exists?("app/consumers/#{consumer_name}_consumer.rb")
+      puts "Already exists file: app/consumers/#{consumer_name}_consumer.rb"
+      ans = cli.ask('Do you want to overwrite it ? (y/n)') { |q| q.validate = /Y|N|y|n/}
+      exit(-1) if ans.downcase == 'n'
+    end
+
+    file = File.open("app/consumers/#{consumer_name}_consumer.rb", 'w+')
+    file.write(text1)
+    file.close
+
+text2 = <<-CODE
+namespace :sevak do
+
+  desc "Run the #{consumer_name} worker"
+  task :#{consumer_name} do
+    autoload(:#{to_class_name.call(consumer_name)}Consumer, './app/consumers/#{consumer_name}_consumer.rb')
+    Sevak.config.autoscale = true if ENV['REPLICA']
+    consumer = Sevak::#{to_class_name.call(consumer_name)}Consumer.new
+    consumer.start
+  end
+end
+
+CODE
+
+    if File.exists?("lib/tasks/#{consumer_name}.rake")
+      puts "Already exists file: lib/tasks/#{consumer_name}.rake"
+      ans = cli.ask('Do you want to overwrite it ? (y/n)') { |q| q.validate = /Y|N|y|n/}
+      exit(-1) if ans.downcase == 'n'
+    end
+
+    file = File.open("lib/tasks/#{consumer_name}.rake", 'w+')
+    file.write(text2)
+    file.close
+
+  end
+end