Class: Fluent::SQSOutput

Inherits:
BufferedOutput
  • Object
show all
Includes:
SetTagKeyMixin, SetTimeKeyMixin
Defined in:
lib/fluent/plugin/out_sqs.rb

Instance Method Summary collapse

Instance Method Details

#configure(conf) ⇒ Object

config_param :buffer_queue_limit, :integer, :default => 10



24
25
26
 
# File 'lib/fluent/plugin/out_sqs.rb', line 24

def configure(conf)
    super
end

#format(tag, time, record) ⇒ Object 



45
46
47
48
49
50
51
 
# File 'lib/fluent/plugin/out_sqs.rb', line 45

def format(tag, time, record)
    if @include_tag then
        record[@tag_property_name] = tag
    end

    record.to_msgpack
end

#shutdown ⇒ Object 



41
42
43
 
# File 'lib/fluent/plugin/out_sqs.rb', line 41

def shutdown
    super
end

#start ⇒ Object 



28
29
30
31
32
33
34
35
36
37
38
39
 
# File 'lib/fluent/plugin/out_sqs.rb', line 28

def start
    super
    
    AWS.config(
        :access_key_id => @aws_key_id,
        :secret_access_key => @aws_sec_key)

    @sqs = AWS::SQS.new(
        :sqs_endpoint => @sqs_endpoint)
    @queue = @sqs.queues.create(@queue_name)
    
end

#write(chunk) ⇒ Object 



53
54
55
56
57
58
59
60
61
62
63
 
# File 'lib/fluent/plugin/out_sqs.rb', line 53

def write(chunk)
    records = []
    chunk.msgpack_each {|record| records << { :message_body => record.to_json, :delay_seconds => @delay_seconds } }
    until records.length <= 0 do
        begin
            @queue.batch_send(records.slice!(0..9))
        rescue => e
            $stderr.puts e
        end
    end
end