Merge pull request #2116 from osela/messagepackeventstream-issue

Fix MessagePackEventStream issue with Enumerable methods

lib/fluent/event.rb

@@ -254,19 +254,9 @@ def slice(index, num)
     end
 
     def each(unpacker: nil, &block)
-      if @unpacked_times
-        @unpacked_times.each_with_index do |time, i|
-          block.call(time, @unpacked_records[i])
-        end
-      else
-        @unpacked_times = []
-        @unpacked_records = []
-        (unpacker || Fluent::MessagePackFactory.msgpack_unpacker).feed_each(@data) do |time, record|
-          @unpacked_times << time
-          @unpacked_records << record
-          block.call(time, record)
-        end
-        @size = @unpacked_times.size
+      ensure_unpacked!(unpacker: unpacker)
+      @unpacked_times.each_with_index do |time, i|
+        block.call(time, @unpacked_records[i])
       end
       nil
     end

test/test_event.rb

@@ -401,6 +401,22 @@ def setup
         i += 1
       }
     end
+
+    # `any?` represents an Enumerable method which calls `each` internally
+    test 'size_after_any' do
+      @es.any?
+
+      assert_equal 2, @es.size
+    end
+
+    # `any?` represents an Enumerable method which calls `each` internally
+    test 'each_after_any' do
+      @es.any?
+
+      count = 0
+      @es.each { |time, record| count += 1 }
+      assert_equal 2, count
+    end
   end
 
   class CompressedMessagePackEventStreamTest < ::Test::Unit::TestCase