Class: Fluent::IRCOutput

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

Defined Under Namespace

Classes: IRCConnection, TimerWatcher

Constant Summary collapse

COMMAND_MAP = 
{
  'priv_msg' => :priv_msg,
  'privmsg'  => :priv_msg,
  'notice'   => :notice,
}

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ IRCOutput

Returns a new instance of IRCOutput.



50
51
52
53
 
# File 'lib/fluent/plugin/out_irc.rb', line 50

def initialize
  super
  require 'irc_parser'
end

Instance Attribute Details

#conn ⇒ Object (readonly)

for test



48
49
50
 
# File 'lib/fluent/plugin/out_irc.rb', line 48

def conn
  @conn
end

Instance Method Details

#configure(conf) ⇒ Object 



55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/fluent/plugin/out_irc.rb', line 55

def configure(conf)
  super

  begin
    @message % (['1'] * @out_keys.length)
  rescue ArgumentError
    raise Fluent::ConfigError, "string specifier '%s' and out_keys specification mismatch"
  end

  if @channel_keys
    begin
      @channel % (['1'] * @channel_keys.length)
    rescue ArgumentError
      raise Fluent::ConfigError, "string specifier '%s' and channel_keys specification mismatch"
    end
  end
  @channel = '#'+@channel

  if @command_keys
    begin
      @command % (['1'] * @command_keys.length)
    rescue ArgumentError
      raise Fluent::ConfigError, "string specifier '%s' and command_keys specification mismatch"
    end
  else
    unless @command = COMMAND_MAP[@command]
      raise Fluent::ConfigError, "command must be one of #{COMMAND_MAP.keys.join(', ')}"
    end
  end

  @send_queue = []
end

#emit(tag, es, chain) ⇒ Object 



118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
 
# File 'lib/fluent/plugin/out_irc.rb', line 118

def emit(tag, es, chain)
  chain.next

  if @conn.closed?
    log.warn "out_irc: connection is closed. try to reconnect"
    @conn = create_connection
  end

  es.each do |time,record|
    if @send_queue.size >= @send_queue_limit
      log.warn "out_irc: send queue size exceeded send_queue_limit(#{@send_queue_limit}), discards"
      break
    end

    filter_record(tag, time, record)

    command = build_command(record)
    channel = build_channel(record)
    message = build_message(record)

    log.debug { "out_irc: push {command:\"#{command}\", channel:\"#{channel}\", message:\"#{message}\"}" }
    @send_queue.push([command, channel, message])
  end
end

#on_timer ⇒ Object 



143
144
145
146
147
148
 
# File 'lib/fluent/plugin/out_irc.rb', line 143

def on_timer
  return if @send_queue.empty?
  command, channel, message = @send_queue.shift
  log.info { "out_irc: send {command:\"#{command}\", channel:\"#{channel}\", message:\"#{message}\"}" }
  @conn.send_message(command, channel, message)
end

#run ⇒ Object 



111
112
113
114
115
116
 
# File 'lib/fluent/plugin/out_irc.rb', line 111

def run
  @loop.run(@blocking_timeout)
rescue => e
  log.error "unexpected error", :error => e, :error_class => e.class
  log.error_backtrace
end

#shutdown ⇒ Object 



103
104
105
106
107
108
109
 
# File 'lib/fluent/plugin/out_irc.rb', line 103

def shutdown
  super
  @loop.watchers.each { |w| w.detach }
  @loop.stop
  @conn.close
  @thread.join
end

#start ⇒ Object 



88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/fluent/plugin/out_irc.rb', line 88

def start
  super

  begin
    @loop = Coolio::Loop.new
    @conn = create_connection
    @timer = TimerWatcher.new(@send_interval, true, log, &method(:on_timer))
    @loop.attach(@timer)
    @thread = Thread.new(&method(:run))
  rescue => e
    puts e
    raise Fluent::ConfigError, "failed to connect IRC server #{@host}:#{@port}"
  end
end