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



22
23
24
 
# File 'lib/fluent/plugin/out_sqs.rb', line 22

def configure(conf)
    super
end

#format(tag, time, record) ⇒ Object 



43
44
45
 
# File 'lib/fluent/plugin/out_sqs.rb', line 43

def format(tag, time, record)
    record.to_msgpack
end

#shutdown ⇒ Object 



39
40
41
 
# File 'lib/fluent/plugin/out_sqs.rb', line 39

def shutdown
    super
end

#start ⇒ Object 



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

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 



47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/fluent/plugin/out_sqs.rb', line 47

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