# File lib/fluent/plugin/in_http.rb, line 235 def initialize(io, km, callback, body_size_limit, format, log, cors_allow_origins) super(io) @km = km @callback = callback @body_size_limit = body_size_limit @next_close = false @format = format @log = log @cors_allow_origins = cors_allow_origins @idle = 0 @km.add(self) @remote_port, @remote_addr = *Socket.unpack_sockaddr_in(io.getpeername) rescue nil end
# File lib/fluent/plugin/in_http.rb, line 389 def closing? @next_close end
# File lib/fluent/plugin/in_http.rb, line 322 def on_body(chunk) if @body.bytesize + chunk.bytesize > @body_size_limit unless closing? send_response_and_close("413 Request Entity Too Large", {}, "Too large") end return end @body << chunk end
# File lib/fluent/plugin/in_http.rb, line 254 def on_close @km.delete(self) end
# File lib/fluent/plugin/in_http.rb, line 258 def on_connect @parser = Http::Parser.new(self) end
# File lib/fluent/plugin/in_http.rb, line 275 def on_headers_complete(headers) expect = nil size = nil if @parser.http_version == [1, 1] @keep_alive = true else @keep_alive = false end @env = {} @content_type = "" headers.each_pair {|k,v| @env["HTTP_#{k.gsub('-','_').upcase}"] = v case k when /Expect/ expect = v when /Content-Length/ size = v.to_i when /Content-Type/ @content_type = v when /Connection/ if v =~ /close/ @keep_alive = false elsif v =~ /Keep-alive/ @keep_alive = true end when /Origin/ @origin = v when /X-Forwarded-For/ # For multiple X-Forwarded-For headers. Use first header value. v = v.first if v.is_a?(Array) @remote_addr = v.split(",").first end } if expect if expect == '100-continue' if !size || size < @body_size_limit send_response_nobody("100 Continue", {}) else send_response_and_close("413 Request Entity Too Large", {}, "Too large") end else send_response_and_close("417 Expectation Failed", {}, "") end end end
# File lib/fluent/plugin/in_http.rb, line 271 def on_message_begin @body = '' end
# File lib/fluent/plugin/in_http.rb, line 332 def on_message_complete return if closing? # CORS check # ========== # For every incoming request, we check if we have some CORS # restrictions and white listed origins through @cors_allow_origins. unless @cors_allow_origins.nil? unless @cors_allow_origins.include?(@origin) send_response_and_close("403 Forbidden", {'Connection' => 'close'}, "") return end end @env['REMOTE_ADDR'] = @remote_addr if @remote_addr uri = URI.parse(@parser.request_url) params = WEBrick::HTTPUtils.parse_query(uri.query) if @format != 'default' params[EVENT_RECORD_PARAMETER] = @body elsif @content_type =~ /^application\/x-www-form-urlencoded/ params.update WEBrick::HTTPUtils.parse_query(@body) elsif @content_type =~ /^multipart\/form-data; boundary=(.+)/ boundary = WEBrick::HTTPUtils.dequote($1) params.update WEBrick::HTTPUtils.parse_form_data(@body, boundary) elsif @content_type =~ /^application\/json/ params['json'] = @body elsif @content_type =~ /^application\/msgpack/ params['msgpack'] = @body end path_info = uri.path params.merge!(@env) @env.clear code, header, body = *@callback.call(path_info, params) body = body.to_s header['Access-Control-Allow-Origin'] = @origin if !@cors_allow_origins.nil? && @cors_allow_origins.include?(@origin) if @keep_alive header['Connection'] = 'Keep-Alive' send_response(code, header, body) else send_response_and_close(code, header, body) end end
# File lib/fluent/plugin/in_http.rb, line 262 def on_read(data) @idle = 0 @parser << data rescue @log.warn "unexpected error", error: $!.to_s @log.warn_backtrace close end
# File lib/fluent/plugin/in_http.rb, line 380 def on_write_complete close if @next_close end
# File lib/fluent/plugin/in_http.rb, line 393 def send_response(code, header, body) header['Content-length'] ||= body.bytesize header['Content-type'] ||= 'text/plain' data = %Q[HTTP/1.1 #{code}\r\n] header.each_pair {|k,v| data << "#{k}: #{v}\r\n" } data << "\r\n" write data write body end
# File lib/fluent/plugin/in_http.rb, line 384 def send_response_and_close(code, header, body) send_response(code, header, body) @next_close = true end
# File lib/fluent/plugin/in_http.rb, line 407 def send_response_nobody(code, header) data = %Q[HTTP/1.1 #{code}\r\n] header.each_pair {|k,v| data << "#{k}: #{v}\r\n" } data << "\r\n" write data end
# File lib/fluent/plugin/in_http.rb, line 250 def step_idle @idle += 1 end