Class: Fluent::ExHttpInput

Inherits:
HttpInput
  • Object
show all
Defined in:
lib/fluent/plugin/in_http_ex.rb

Defined Under Namespace

Classes: ExHandler, KeepaliveManager

Instance Method Summary collapse

Instance Method Details

#on_request(path_info, params) ⇒ 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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
 
# File 'lib/fluent/plugin/in_http_ex.rb', line 61

def on_request(path_info, params)
  $log.debug "remote_addr: #{params["REMOTE_ADDR"]}, path_info: #{path_info}"
  begin
    path = path_info[1..-1] # remove /
    resource, tag = path.split('/')
    tag ||= resource

    if chunk = params['chunked']
      record = chunk 
      
    elsif js = params['json']
      record = JSON.parse(js)

    elsif ms = params['x-msgpack'] || ms = params['msgpack']
      record = MessagePack::unpack(ms)

    else
      raise "'json' or 'msgpack' parameter is required"
    end

    time = params['time']
    time = time.to_i
    if time == 0
      time = Engine.now
    end

  rescue
    return ["400 Bad Request", {'Content-type'=>'text/plain'}, "400 Bad Request\n#{$!}\n"]
  end

  # TODO server error
  begin
    case params["resource"]
    when :js
      record.each do |v|
        line = begin
          JSON.parse(v)
        rescue TypeError
          v #hash
        end
        Engine.emit(tag, time, line)
      end

    when :ms
      record.each {|line| Engine.emit(tag, time, line) }

    else
      Engine.emit(tag, time, record)
    end

  rescue
    return ["500 Internal Server Error", {'Content-type'=>'text/plain'}, "500 Internal Server Error\n#{$!}\n"]
  end

  return ["200 OK", {'Content-type'=>'text/plain'}, ""]
end

#start ⇒ Object 



42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
 
# File 'lib/fluent/plugin/in_http_ex.rb', line 42

def start
  $log.debug "listening http on #{@bind}:#{@port}"
  lsock = TCPServer.new(@bind, @port)

  detach_multi_process do
    Input.new.start
    @km = KeepaliveManager.new(@keepalive_timeout)
    @lsock = Coolio::TCPServer.new(lsock, nil, ExHandler, @km, method(:on_request),
                                   @body_size_limit, nil, $log,
                                   @cors_allow_origins)

    @loop = Coolio::Loop.new
    @loop.attach(@km)
    @loop.attach(@lsock)

    @thread = Thread.new(&method(:run))
  end
end