Class: Fluent::MessagePackEventStream

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

Direct Known Subclasses

CompressedMessagePackEventStream

Instance Method Summary collapse

Methods inherited from EventStream

#==, #to_compressed_msgpack_stream, #to_msgpack_stream_forced_integer

Methods included from Plugin::Compressable

#compress, #decompress

Constructor Details

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

Keep cached_unpacker argument for existing plugins



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

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 



217
218
219
220
221
222
223
 
# File 'lib/fluent/event.rb', line 217

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

#each(unpacker: nil, &block) ⇒ Object 



256
257
258
259
260
261
262
 
# File 'lib/fluent/event.rb', line 256

def each(unpacker: nil, &block)
  ensure_unpacked!(unpacker: unpacker)
  @unpacked_times.each_with_index do |time, i|
    block.call(time, @unpacked_records[i])
  end
  nil
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


213
214
215
 
# File 'lib/fluent/event.rb', line 213

def empty?
  @data.empty?
end

#ensure_unpacked!(unpacker: nil) ⇒ Object 



236
237
238
239
240
241
242
243
244
245
246
247
 
# File 'lib/fluent/event.rb', line 236

def ensure_unpacked!(unpacker: nil)
  return if @unpacked_times && @unpacked_records
  @unpacked_times = []
  @unpacked_records = []
  (unpacker || Fluent::MessagePackFactory.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) 


232
233
234
 
# File 'lib/fluent/event.rb', line 232

def repeatable?
  true
end

#size ⇒ Object 



225
226
227
228
229
230
 
# File 'lib/fluent/event.rb', line 225

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.



251
252
253
254
 
# File 'lib/fluent/event.rb', line 251

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, packer: nil) ⇒ Object 



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

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