gistfile1.rb

def consume(*types, &block)
  filter_types(types)
  @t.buf_consumer(buf_consumer)
  begin
    while(true) do
      @t.sleep # the tricky little trickster
      work = @t.work(probe_consumer, rec_consumer(block))
      if (@done || work > 0)
        break
      end
    end
  ensure
    @t.stop
    @t.work(probe_consumer)
  end
end