Class: Fluent::MessagePackEventStream

Inherits:
EventStream show all
Defined in:
lib/fluent/event.rb

Instance Method Summary collapse

Methods inherited from EventStream

#==, #to_msgpack_stream_forced_integer

Methods included from Fluent::MessagePackFactory::Mixin

#msgpack_factory, #msgpack_packer, #msgpack_unpacker

Constructor Details

#initialize(data, cached_unpacker = nil, size = 0, unpacked_times: nil, unpacked_records: nil) ⇒ MessagePackEventStream

Keep cached_unpacker argument for existing plugins



196
197
198
199
200
201
 
# File 'lib/fluent/event.rb', line 196

def initialize(data, cached_unpacker = nil, size = 0, unpacked_times: nil, unpacked_records: nil)
  @data = data
  @size = size
  @unpacked_times = unpacked_times
  @unpacked_records = unpacked_records
end

Instance Method Details

#dup ⇒ Object 



207
208
209
210
211
212
213
 
# File 'lib/fluent/event.rb', line 207

def dup
  if @unpacked_times
    MessagePackEventStream.new(@data.dup, nil, @size, unpacked_times: @unpacked_times, unpacked_records: @unpacked_records.map(&:dup))
  else
    MessagePackEventStream.new(@data.dup, nil, @size)
  end
end

#each(&block) ⇒ Object 



246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
 
# File 'lib/fluent/event.rb', line 246

def each(&block)
  if @unpacked_times
    @unpacked_times.each_with_index do |time, i|
      block.call(time, @unpacked_records[i])
    end
  else
    @unpacked_times = []
    @unpacked_records = []
    msgpack_unpacker.feed_each(@data) do |time, record|
      @unpacked_times << time
      @unpacked_records << record
      block.call(time, record)
    end
    @size = @unpacked_times.size
  end
  nil
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


203
204
205
 
# File 'lib/fluent/event.rb', line 203

def empty?
  @data.empty?
end

#ensure_unpacked! ⇒ Object 



226
227
228
229
230
231
232
233
234
235
236
237
 
# File 'lib/fluent/event.rb', line 226

def ensure_unpacked!
  return if @unpacked_times && @unpacked_records
  @unpacked_times = []
  @unpacked_records = []
  msgpack_unpacker.feed_each(@data) do |time, record|
    @unpacked_times << time
    @unpacked_records << record
  end
  # @size should be updated always right after unpack.
  # The real size of unpacked objects are correct, rather than given size.
  @size = @unpacked_times.size
end

#repeatable? ⇒ Boolean

Returns:

  • (Boolean) 


222
223
224
 
# File 'lib/fluent/event.rb', line 222

def repeatable?
  true
end

#size ⇒ Object 



215
216
217
218
219
220
 
# File 'lib/fluent/event.rb', line 215

def size
  # @size is unbelievable always when @size == 0
  # If the number of events is really zero, unpacking events takes very short time.
  ensure_unpacked! if @size == 0
  @size
end

#slice(index, num) ⇒ Object

This method returns MultiEventStream, because there are no reason to surve binary serialized by msgpack.



241
242
243
244
 
# File 'lib/fluent/event.rb', line 241

def slice(index, num)
  ensure_unpacked!
  MultiEventStream.new(@unpacked_times.slice(index, num), @unpacked_records.slice(index, num))
end

#to_msgpack_stream(time_int: false) ⇒ Object 



264
265
266
267
 
# File 'lib/fluent/event.rb', line 264

def to_msgpack_stream(time_int: false)
  # time_int is always ignored because @data is always packed binary in this class
  @data
end