Install
===

    gem install producer_consumer

Use
===

    ProducerConsumer.produce 4 do |p|
      # 4 threads will produce items. When they're done, they call p.finish.
      some_item or p.finish
    end.consume 10 do |item|
      # 10 threads will consume items produced by the above block.
      puts item
    end.run

Install
===

    gem install producer_consumer

Use
===

ProducerConsumer is a model for distributing work concurrently. A producer runs
a block over and over again, yielding a list of objects. Those objects are
handed to consumers. Producers run in n distinct threads, and consumers run in
m distinct threads. When a producer is done, it may optionally call .finish on
its block argument to signal that it has completed its work. #run will wait for
all producers to finish, and terminate the consumers when all items have been
processed.
    
    ProducerConsumer.produce 4 do |p|
      # 4 threads will produce items. When they're done, they call p.finish.
      some_item or p.finish
    end.consume 10 do |item|
      # 10 threads will consume items produced by the above block.
      puts item
    end.run

Here's an example which reads keys from a file and performs deletes against
Riak, across 40 concurrent connections, including status messages.

    #!/usr/bin/env ruby

    require 'rubygems'
    require 'riak'
    require 'producer_consumer'

    f = File.open(ARGV.first)
    c = 0

    ProducerConsumer.produce do |p|
      begin
        key = f.readline.chomp

        puts "<- #{c}" if c % 100 == 0
        [c += 1, k.chomp]
      rescue
        p.finish
      end
    end.consume(40) do |c, key|
      puts "-> #{c}" if c % 100 == 0
      
      r = (Thread.current['riak'] ||= Riak::Client.new(:host => '127.0.0.1', :protocol => 'pbc'))
      
      begin
        r['my_bucket'].delete key, :w => 1, :dw => 0
      rescue Riak::ProtobuffsFailedRequest
      end
    end.run

Expand readme and include a real-world example.
Expand readme and include a real-world example.
