Tasks for Symofony2 project to start/stop/restart rabbitmq consumers

consumers.yml

# config/consumers.yml
# {consumer_name}: {how many processes to start}
consumers:
  my_consumer_1: 1
  my_consumer_2: 1

deploy.rb

# config/deploy.rb

# ...

#
# Project config
#

set :consumers_config_path, 'config/consumers.yml'

# ...

rabbitmq_consumers.rake

# lib/capistrano/tasks/rabbitmq_consumers.rake
namespace :project do
  namespace :consumer do

    desc 'Run all the project consumers'
    task :start_all do
      on roles(:app) do

        # Get consumers config for this release
        consumers_config_path = "#{release_path}/" + fetch(:consumers_config_path)

        if test("[ -f #{consumers_config_path} ]")
          consumers_config = YAML::load(capture("cat #{consumers_config_path}"))
          consumers_config['consumers'].each do |consumer, processes|
            invoke 'project:consumer:start', consumer, processes
            Rake::Task['project:consumer:start'].reenable
          end
        else
          info 'There is no consumers config in this release. Consumers will not be started.'
        end

      end
    end

    desc 'Stop all the project consumers'
    task :stop_all do
      on roles(:app) do

        # Get consumers config for this release
        consumers_config_path = "#{release_path}/" + fetch(:consumers_config_path)

        if test("[ -f #{consumers_config_path} ]")
          consumers_config = YAML::load(capture("cat #{consumers_config_path}"))
          consumers_config['consumers'].each do |consumer, processes|
            invoke 'project:consumer:stop', consumer
            Rake::Task['project:consumer:stop'].reenable
          end
        else
          info 'There is no consumers config in this release. Consumers will not be stopped.'
        end

      end
    end

    desc 'Restart all the project consumers'
    task :restart_all do
      on roles(:app) do

        # Get consumers config for this release
        consumers_config_path = "#{release_path}/" + fetch(:consumers_config_path)

        if test("[ -f #{consumers_config_path} ]")
          consumers_config = YAML::load(capture("cat #{consumers_config_path}"))
          consumers_config['consumers'].each do |consumer, processes|
            invoke 'project:consumer:stop', consumer
            Rake::Task['project:consumer:stop'].reenable

            invoke 'project:consumer:start', consumer, processes
            Rake::Task['project:consumer:start'].reenable
          end
        else
          info 'There is no consumers config in this release. Consumers will not be stopped nor started.'
        end

      end
    end

    task :start, :consumer_name, :processes, :params do |t, args|
      on roles(:app) do
        within release_path do

          processes = args[:processes] || 1
          params = (args[:params] || '') + fetch(:symfony_console_flags)

          # Run the process
          for i in 1..processes
            # NOTES:
            # 'nohup'               do not stop the process after SSH session will be closed
            # ' > /dev/null 2>&1'   capistrano will not wait for the exit status and deploy will not get stuck
            # '&'                   to start process in background
            # 'sleep 1'             the hack to make 'nohup' working
            execute :nohup, 'php', fetch(:symfony_console_path), 'rabbitmq:consumer', args[:consumer_name], params, ' > /dev/null 2>&1 & sleep 1'
          end

          info "Consumer #{args[:consumer_name]} started (#{processes} processes)."

          # Debug
          processes_found = capture :ps, "ax | grep 'rabbitmq:consumer #{args[:consumer_name]}' | grep -v grep | wc -l"
          debug "Consumer '#{args[:consumer_name]}': Expecting for #{processes} processes, #{processes_found} processes are running."

        end
      end
    end

    task :stop, :consumer_name do |t, args|
      on roles(:app) do
        within release_path do

          # Due to a bug in a php-amqplib (https://github.com/videlalvaro/php-amqplib/issues/352),
          # only SIGKILL and SIGQUIT will stop the process.
          signal = '9'
          processes_found = capture :ps, "ax | grep 'rabbitmq:consumer #{args[:consumer_name]}' | grep -v grep | wc -l"

          if processes_found == '0'
            info "Consumer #{args[:consumer_name]} is not running. Don't stop it."
          else
            # Kill all processes of the consumer
            execute :kill, '-s', signal, "`ps aux | less | grep 'rabbitmq:consumer #{args[:consumer_name]}' | grep -v grep | awk '{print $2}'`"
            info "Consumer #{args[:consumer_name]} stopped (#{processes_found} processes)."
          end

        end
      end
    end

  end
end