adding initial version. check out readme and tests for example usages

.gitignore

@@ -0,0 +1,4 @@
+*.gem
+.bundle
+Gemfile.lock
+pkg/*

Gemfile

@@ -0,0 +1,9 @@
+source "http://rubygems.org"
+
+# Specify your gem's dependencies in queue_bundle.gemspec
+gemspec
+
+group :dev do
+  gem 'rake'
+
+end

LICENSE

@@ -0,0 +1,19 @@
+Copyright (c) 2011 Edgenet Inc
+
+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,37 @@
+The QueueBundle presents itself as a standard queue to those who put(value) stuff in.
+the value is then hashed based on some simple hashing algorithm's,  or a custom one can be provided.
+based on the result of that hashing it is slotted into an output queue
+
+threads can then query for results from their queue by specifying their index.
+
+
+Install
+=======
+    sudo gem install queue_bundle
+
+Usage
+=====
+
+    #A simple approach for usimg would be
+    queue = QueueBundle.new(5, :key_lookup => :id)
+    threads = []
+    0.upto(4).each {|worker_id|
+      threads << Thread.new do 
+        loop do
+          break if queue.closed? && queue.empty?(worker_id)
+          work = queue.pop(worker_id)   #in this case this would be blocking
+        end
+      end
+    }
+    0.upto(50).each {|work_id|
+      queue.push({id: work_id, name: "task"})
+    }
+    # sets a flag that queue is closed so threads know that no more work is coming
+    queue.close    
+
+    threads.each do |t|
+      begin
+        t.join
+      rescue Interrupt
+      end
+    end

Rakefile

@@ -0,0 +1,9 @@
+require 'bundler/gem_tasks'
+require 'rake/testtask'
+task :default => :test
+
+
+  Rake::TestTask.new do |t|
+    t.test_files = FileList['test/test_*.rb']
+    t.verbose = true
+  end

lib/queue_bundle.rb

@@ -0,0 +1,214 @@
+require "queue_bundle/version"
+
+require 'thread'
+
+class QueueNotFound < StandardError; end
+
+class QueueBundle
+
+
+
+  attr_reader :hash_algorithm
+  def initialize(size, options = {})
+    @mutex = Mutex.new
+    @queues = []
+    @size = size
+    @closed = false
+    @hash_algorithm = options[:hash_algorithm] || :simple_hash
+    @key_lookup = options[:key_lookup] || :id
+    1.upto(@size) {|i|
+      @queues << Queue.new
+    }
+  end
+  def is_queue?
+    true
+  end
+  def simple_hash(key)
+    key.hash % @size
+  end
+  def simple_index(key)
+    key
+  end
+
+ # def resize(new_size)
+ #   
+ # end
+
+  def closed?
+    @closed
+  end
+  def empty?(index = nil)
+    self.length(index) == 0
+  end
+  def reassign(index)
+    @mutex.synchronize do
+      dest_queue = next_queue(index)
+      if !dest_queue.nil?
+        old_queue = @queues[index]
+        @queues[index] = :reassigned
+        if !old_queue.empty?
+          begin
+            while !old_queue.empty?
+              dest_queue.push(old_queue.pop(true))
+            end
+          rescue ThreadError
+
+          end
+
+        end
+        true
+      else
+        false
+      end
+    end
+  end
+
+  def close()
+    @closed = true
+  end
+  def push(obj, key = nil)
+    key ||= get_key(obj)
+    queue = get_queue(key)
+
+    if queue.nil?
+      raise QueueNotFound.new("Queue not found for #{key}")
+    else
+      queue.push(obj)
+    end
+  end
+  alias_method :enq, :push
+  def pop(index, non_block = false)
+    queue = @queues[index]
+
+    if queue.nil? 
+      raise QueueNotFound.new("Queue not found for #{key}")
+    elsif queue == :reassigned
+      raise QueueNotFound.new("Queue reassigned")
+    else
+      queue.pop(non_block)
+    end
+  end
+  alias_method :deq, :pop
+  def clear(index = nil)
+    if index.nil?
+      active_queues.each {|queue| queue.clear}
+    else
+      queue = @queues[index]   #doesn't switch to next queue so that it will error out
+
+      if queue.nil? 
+        raise QueueNotFound.new("Queue not found for #{key}")
+      elsif queue == :reassigned
+        raise QueueNotFound.new("Queue reassigned")
+      else
+        queue.clear()
+      end
+    end
+  end
+  def num_waiting(index = nil)
+    if index.nil?
+      active_queues.map {|queue| queue.num_waiting}.inject{|sum,x| sum + x }
+    else
+      queue = @queues[index]
+
+      if queue.nil? 
+        raise QueueNotFound.new("Queue not found for #{key}")
+      elsif queue == :reassigned
+        raise QueueNotFound.new("Queue reassigned")
+      else
+        queue.num_waiting
+      end
+    end
+  end
+  def empty?(index = nil)
+    if index.nil?
+      active_queues.inject{|total,y|  (!total || !y) ? false : true }  # if it sees a false then it is not empty
+    else
+      queue = @queues[index]
+
+      if queue.nil? 
+        raise QueueNotFound.new("Queue not found for #{key}")
+      elsif queue == :reassigned
+        raise QueueNotFound.new("Queue reassigned")
+      else
+        queue.empty?
+      end
+    end
+  end
+
+  def length(index = nil)
+    if index.nil?
+      active_queues.map {|queue| queue.length}.inject{|sum,x| sum + x }
+    else
+      queue = @queues[index]
+
+      if queue.nil? 
+        raise QueueNotFound.new("Queue not found for #{key}")
+      elsif queue == :reassigned
+        raise QueueNotFound.new("Queue reassigned")
+      else
+        queue.length
+      end
+    end
+  end
+
+  alias_method :size, :length
+
+  def lengths
+    length_hash = {}
+    active_queues.each_index {|queue_index| 
+      length_hash[queue_index] = @queues[queue_index].length
+    }
+    length_hash
+  end
+  alias_method :sizes, :lengths
+  private
+
+    def get_key(obj)
+      key = nil
+      if @key_lookup.kind_of?(Proc)
+        key = @key_lookup.call obj
+      elsif @key_lookup.kind_of?(Symbol)
+        if obj.kind_of?(Hash)
+          key = obj[@key_lookup]
+        else
+          key = obj.send(@key_lookup)
+        end
+      end
+      key
+    end
+    def get_queue(key)
+      queue_index = nil
+      if @hash_algorithm.kind_of?(Proc)
+        queue_index = @hash_algorithm.call key
+      elsif @hash_algorithm.kind_of?(Symbol)
+        queue_index = self.send(@hash_algorithm, key)
+      end
+      queue = @queues[queue_index]
+      if queue == :reassigned
+        queue = next_queue(queue_index)
+      end
+
+      queue
+    end
+    def active_queues
+      @queues.select {|queue| queue != :reassigned }
+    end
+    def next_queue(index)
+      current_index = index
+      while n = next_index(current_index) 
+        break if n == index
+        if @queues[n] != :reassigned
+          return @queues[n]
+        end
+        current_index = n
+      end
+      nil
+    end
+    def next_index(index)
+      n = index + 1
+      if n >= @size
+        n = 0
+      end
+      n
+    end
+end

lib/queue_bundle/version.rb

@@ -0,0 +1,3 @@
+class QueueBundle
+  VERSION = "0.0.1"
+end

queue_bundle.gemspec

@@ -0,0 +1,20 @@
+# -*- encoding: utf-8 -*-
+$:.push File.expand_path("../lib", __FILE__)
+require "queue_bundle/version"
+
+Gem::Specification.new do |s|
+  s.name        = "queue_bundle"
+  s.version     = QueueBundle::VERSION
+  s.authors     = ["Pete Brumm"]
+  s.email       = ["pbrumm@edgenet.com"]
+  s.homepage    = ""
+  s.summary     = %q{Provides a way to split queue put's into multiple seperate queue pulls}
+  s.description = %q{Allows queue work to be distributed to seperate threads that need a consistent end point.   Allows you to provide a hashing algorithm for which thread handles the work.}
+
+  s.rubyforge_project = "queue_bundle"
+
+  s.files         = Dir['bin/*'] + Dir['lib/*'] + Dir['lib/**/*.rb'] + Dir['test/**/*.rb']
+  s.test_files    = Dir['test/**/*.rb']
+ # s.executables   = `git ls-files -- bin/*`.split("\n").map{ |f| File.basename(f) }
+  s.require_paths = ["lib"]
+end

test/helper.rb

@@ -0,0 +1,10 @@
+require 'rubygems'
+require 'test/unit'
+
+$LOAD_PATH.unshift(File.join(File.dirname(__FILE__), '..', 'lib'))
+$LOAD_PATH.unshift(File.dirname(__FILE__))
+
+require 'queue_bundle'
+
+class Test::Unit::TestCase
+end