Initial import.

Author: Malte Rohde

COPYING

@@ -0,0 +1,21 @@
+The MIT License (MIT)
+
+Copyright (c) 2014 FlavourSys Technology GmbH
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

README.md

@@ -0,0 +1,77 @@
+# Sidekiq::QueuePause
+
+This gem adds a `.pause` functionality to Sidekiq queues. Sidekiq Pro has this feature, so please consider upgrading if you can.
+
+## Usage
+
+Initializer:
+
+```ruby
+Sidekiq.configure_server do |config|
+  Sidekiq.options[:fetch] = Sidekiq::QueuePause::PausingFetch
+
+  # Optionally, you may set some unique key identifying the
+  # Sidekiq process you want to control. This (server) process will
+  # only be paused/unpaused when the function gets called with
+  # the corresponding key. See below.
+  Sidekiq::QueuePause.process_key = 'foo'
+
+  # You may also pass in a Proc which is then evaluated when needed.
+  Sidekiq::QueuePause.process_key { SomeClass.some_method }
+
+  # Optionally, you may configure the sleep period (in seconds) after the
+  # queue lock has been checked. By default, the fetcher will sleep for
+  # Sidekiq::Fetcher::TIMEOUT, i.e. the same time that the redis fetch
+  # command may take.
+  Sidekiq::QueuePause.retry_after = 5
+end
+```
+
+Pausing a queue:
+
+```ruby
+Sidekiq::QueuePause.pause(:example)
+```
+
+```ruby
+# With process key:
+Sidekiq::QueuePause.pause(:example, 'foo')
+```
+
+```ruby
+# On a queue object:
+example = Sidekiq::Queue.new(:example)
+example.pause
+```
+
+```ruby
+# On a queue object with process key:
+example.pause('foo')
+```
+
+Unpause:
+
+```ruby
+Sidekiq::QueuePause.unpause(:example)
+```
+
+etc.
+
+Getting pause status:
+
+```ruby
+Sidekiq::QueuePause.paused?(:example)
+# => true/false
+```
+
+etc.
+
+Unpause all queues/processes:
+
+```ruby
+Sidekiq::QueuePause.unpause_all
+```
+
+## License
+
+The gem is licensed under the [MIT-License](COPYING).

lib/sidekiq-queue-pause.rb

@@ -0,0 +1,84 @@
+require 'celluloid'
+require 'sidekiq'
+require 'sidekiq/fetch'
+
+module Sidekiq
+  module QueuePause
+    PREFIX = 'queue_pause'
+
+    class << self
+      attr_accessor :retry_after
+      attr_writer   :process_key
+
+      def process_key(&block)
+        if block_given?
+          @process_key = block
+        else
+          @process_key.is_a?(Proc) ? @process_key.call : @process_key
+        end
+      end
+
+      def pause(queue, pkey = nil)
+        Sidekiq.redis { |it| it.set rkey(queue, pkey), true }
+      end
+
+      def unpause(queue, pkey = nil)
+        Sidekiq.redis { |it| it.del rkey(queue, pkey) }
+      end
+
+      def paused?(queue, pkey = nil)
+        Sidekiq.redis { |it| it.exists rkey(queue, pkey) }
+      end
+
+      def unpause_all
+        Sidekiq.redis { |it| it.keys("#{PREFIX}:*").each { |k| it.del k } }
+      end
+
+      private
+
+      def rkey(queue, pkey)
+        pkey ? "#{PREFIX}:#{queue}:#{pkey}" : "#{PREFIX}:#{queue}"
+      end
+    end
+
+    class PausingFetch < Sidekiq::BasicFetch
+      def retrieve_work
+        qcmd = unpaused_queues_cmd
+
+        if qcmd.size > 1
+          retrieve_work_for_queues qcmd
+        else
+          sleep(Sidekiq::QueuePause.retry_after || Sidekiq::Fetcher::TIMEOUT)
+          nil
+        end
+      end
+
+      def retrieve_work_for_queues(qcmd)
+        work = Sidekiq.redis { |conn| conn.brpop(*qcmd) }
+        UnitOfWork.new(*work) if work
+      end
+
+      def unpaused_queues_cmd
+        queues = queues_cmd
+        queues.reject do |q|
+          q != Sidekiq::Fetcher::TIMEOUT &&
+            Sidekiq::QueuePause.paused?(q.gsub('queue:', ''), Sidekiq::QueuePause.process_key)
+        end
+      end
+    end
+  end
+
+  class Queue
+    def pause(pkey = nil)
+      Sidekiq::QueuePause.pause(name, pkey)
+    end
+
+    def unpause(pkey = nil)
+      Sidekiq::QueuePause.unpause(name, pkey)
+    end
+
+    def paused?(pkey = nil)
+      Sidekiq::QueuePause.paused?(name, pkey)
+    end
+  end
+end

sidekiq-queue-pause.gemspec

@@ -0,0 +1,15 @@
+Gem::Specification.new do |s|
+  s.name          = 'sidekiq-queue-pause'
+  s.version       = '0.0.1'
+  s.summary       = 'Pause a Sidekiq queue'
+  s.description   = 'Let\'s you pause/unpause individual sidekiq queues.'
+  s.license       = 'MIT'
+  s.authors       = ['FlavourSys Technology GmbH']
+  s.email         = ['technology@flavoursys.com']
+  s.homepage      = 'http://github.com/FlavourSys/sidekiq-queue-pause'
+
+  s.require_paths = ['lib']
+  s.files         = Dir['lib/**/*rb']
+  
+  s.add_dependency 'sidekiq', '~> 3.1'
+end