Class: Fluent::NamedPipeInput

Inherits:
Input
  • Object
show all
Defined in:
lib/fluent/plugin/in_namedpipe.rb

Instance Method Summary collapse

Constructor Details

#initialize ⇒ NamedPipeInput 



9
10
11
12
 
# File 'lib/fluent/plugin/in_namedpipe.rb', line 9

def initialize
    super
    @path = nil
end

Instance Method Details

#configure(conf) ⇒ Object 



19
20
21
22
23
24
25
26
27
28
29
30
31
 
# File 'lib/fluent/plugin/in_namedpipe.rb', line 19

def configure(conf)
    super

    if !File.exists?(@path)
        raise ConfigError,"File not found #{@path}"
    end

    if @tag.nil? 
        raise ConfigError,"tag is empty"
    end

    configure_parser(conf)
end

#configure_parser(conf) ⇒ Object 



33
34
35
36
 
# File 'lib/fluent/plugin/in_namedpipe.rb', line 33

def configure_parser(conf)
    @parser = TextParser.new
    @parser.configure(conf)
end

#configure_tag ⇒ Object

TODO: Not yet used



39
40
41
42
43
44
45
46
47
48
 
# File 'lib/fluent/plugin/in_namedpipe.rb', line 39

def configure_tag
    if @tag.index('*')
        @tag_prefix, @tag_suffix = @tag.split('*')
        @tag_suffix ||= ''
    else
        @tag_prefix = nil
        @tag_suffix = nil
    end

end

#parse_line(line) ⇒ Object 



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

def parse_line(line)
    return @parser.parse(line)
end

#run ⇒ Object 



61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/fluent/plugin/in_namedpipe.rb', line 61

def run
    until @finished
        begin
            lines = @pipe.gets
            if lines.nil?
                sleep @receive_interval
                next
            end

            lines = lines.split("\n")
            lines.each { |line| 
                time, record = parse_line(line)
                if time && record
                    Engine.emit(@tag,time,record)
                else
                    log.warn "Pattern not match: #{line.inspect}"
                end
            }
        rescue
            log.error "unexpected error", :error=>$!.to_s
            log.error_backtrace
        end
    end
end

#shutdown ⇒ Object 



86
87
88
89
90
91
 
# File 'lib/fluent/plugin/in_namedpipe.rb', line 86

def shutdown
    super
    @finished = true
    @thread.join
    @pipe.close
end

#start ⇒ Object 



54
55
56
57
58
59
 
# File 'lib/fluent/plugin/in_namedpipe.rb', line 54

def start
    super
    @pipe = open(@path,"r")
    @finished = false
    @thread = Thread.new(&method(:run))
end