Summary of worker handled messages in async example

examples/async/reader.rb

@@ -8,37 +8,44 @@
 puts 'Press enter to start and enter to finish'
 $stdin.gets
 
-finish_mutex   = Mutex.new
 reader         = NSQ.create_reader(:nsqd_tcp_addresses => '127.0.0.1:4150')
 
 x_subscriber   = reader.subscribe('test_xy', 'x', :max_in_flight => x_worker_count)
 y_subscriber   = reader.subscribe('test_xy', 'y', :max_in_flight => y_worker_count)
 z_subscriber   = reader.subscribe('test_z',  'z', :max_in_flight => z_worker_count)
 
-threads = []
-[[x_subscriber, x_worker_count, 'x'], [y_subscriber, y_worker_count, 'y'], [z_subscriber, z_worker_count, 'z']].each do |subscriber, count, char|
-  threads += count.times.map do |i|
-    Thread.new(i, subscriber, char) do |i, subscriber, char|
-      message_count = 0
+class MyThread < Thread
+  attr_accessor :message_count
+  def initialize(index, subscriber, char)
+    @index = index
+    super do |i, subscriber, char|
+      @message_count = 0
       subscriber.run do |message|
-        print char
         eval message.body
-        message_count += 1
-      end
-      if ARGV[0] == '-v'
-        finish_mutex.synchronize { puts '%s[%02d] handled %4d messages' % [char, i, message_count]}
-      else
-        print char.upcase
+        print char
+        @message_count += 1
       end
+      print char.upcase
     end
   end
 end
+
+threads = {}
+[[x_subscriber, x_worker_count, 'x'], [y_subscriber, y_worker_count, 'y'], [z_subscriber, z_worker_count, 'z']].each do |subscriber, count, char|
+  threads[char] = count.times.map do |i|
+    MyThread.new(i, subscriber, char)
+  end
+end
+
 main_thread = Thread.new do
   reader.run
 end
 $stdin.gets
 puts 'Exiting...'
 reader.stop
 main_thread.join
-threads.each(&:join)
+threads.each_value { |arr| arr.each(&:join) }
 puts
+threads.each do |char, arr|
+  puts "#{char} -  #{arr.map(&:message_count).join(' ')}"
+end