Class: Fluent::KinesisStreamsOutput

Inherits:
KinesisOutput
  • Object
show all
Includes:
Fluent::KinesisHelper::API::BatchRequest
Defined in:
lib/fluent/plugin/out_kinesis_streams.rb

Constant Summary collapse

RequestType =
:streams
BatchRequestLimitCount =
500
BatchRequestLimitSize =
5 * 1024 * 1024

Constants included from Fluent::KinesisHelper::API

Fluent::KinesisHelper::API::MaxRecordSize

Instance Method Summary collapse

Methods included from Fluent::KinesisHelper::API::BatchRequest

included, #size_of_values

Methods inherited from KinesisOutput

fluentd_v0_12?, #multi_workers_ready?

Methods included from Fluent::KinesisHelper::API

included

Methods included from Fluent::KinesisHelper::Client

#client, included

Instance Method Details

#configure(conf) ⇒ Object



29
30
31
32
# File 'lib/fluent/plugin/out_kinesis_streams.rb', line 29

def configure(conf)
  super
  @key_formatter = key_formatter_create
end

#format(tag, time, record) ⇒ Object



34
35
36
37
38
39
40
# File 'lib/fluent/plugin/out_kinesis_streams.rb', line 34

def format(tag, time, record)
  format_for_api do
    data = @data_formatter.call(tag, time, record)
    key = @key_formatter.call(record)
    [data, key]
  end
end

#write(chunk) ⇒ Object



42
43
44
45
46
47
48
49
50
51
52
# File 'lib/fluent/plugin/out_kinesis_streams.rb', line 42

def write(chunk)
  write_records_batch(chunk) do |batch|
    records = batch.map{|(data, partition_key)|
      { data: data, partition_key: partition_key }
    }
    client.put_records(
      stream_name: @stream_name,
      records: records,
    )
  end
end