Class: Fluent::ExHttpInput::ExHandler

Inherits:

Handler

• Object
• Handler
• Fluent::ExHttpInput::ExHandler

Defined in:

lib/fluent/plugin/in_http_ex.rb

Instance Method Summary

• #check_content_type(params, content_type, body, path_info) ⇒ Object
• #on_body(chunk) ⇒ Object
• #on_close ⇒ Object
• #on_headers_complete(headers) ⇒ Object
• #on_message_complete ⇒ Object
• #on_read_msgpack(data) ⇒ Object

Instance Method Details

#check_content_type(params, content_type, body, path_info) ⇒ Object

# File 'lib/fluent/plugin/in_http_ex.rb', line 238

def check_content_type(params, content_type, body, path_info)
  path = path_info[1..-1] # remove /
  resource, _ = path.split('/')
  case resource
  when /^js$/
    params["resource"] = :js
  when /^ms$/
    params["resource"] = :ms
  end

  if 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|x-msgpack)/
    params[$1] = body
  end

  params
rescue => ex
  $log.error ex
  $log.error ex.backtrace * "\n"
end

#on_body(chunk) ⇒ Object

# File 'lib/fluent/plugin/in_http_ex.rb', line 167

def on_body(chunk)
  if @chunked && @content_type =~ /application\/x-msgpack/i
    m = method(:on_read_msgpack)
    @u = MessagePack::Unpacker.new
    (class << self; self; end).module_eval do
      define_method(:on_body, m)
    end
    m.call(chunk)
  else
    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
end

#on_close ⇒ Object

# File 'lib/fluent/plugin/in_http_ex.rb', line 119

def on_close
  $log.debug "close #{@remote_addr}:#{@remote_port}"
  super
end

#on_headers_complete(headers) ⇒ Object

# File 'lib/fluent/plugin/in_http_ex.rb', line 124

def on_headers_complete(headers)
  expect = nil
  size = nil
  if @parser.http_version == [1, 1]
    @keep_alive = true
  else
    @keep_alive = false
  end
  @env = {}
  headers.each_pair {|k,v|
    @env["HTTP_#{k.gsub('-','_').upcase}"] = v
    case k
    when /Expect/i
      expect = v
    when /Content-Length/i
      size = v.to_i
    when /Content-Type/i
      @content_type = v
    when /Connection/i
      if v =~ /close/i
        @keep_alive = false
      elsif v =~ /Keep-alive/i
        @keep_alive = true
      end
    when /Transfer-Encoding/i
      if v =~ /chunked/i
        @chunked = true
      end
    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

#on_message_complete ⇒ Object

# File 'lib/fluent/plugin/in_http_ex.rb', line 200

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

  path_info, query_string = @parser.request_url.split("?", 2)
  params = WEBrick::HTTPUtils.parse_query(query_string)

  params = check_content_type(params, @content_type, @body, path_info)
  params.merge!(@env)
  @env.clear

  unless @chunked
    code, header, body = *@callback.call(path_info, params)
    body = body.to_s

    if @keep_alive
      header['Connection'] = 'Keep-Alive'
      send_response(code, header, body)
    else
      send_response_and_close(code, header, body)
    end
  else
    send_response("200 OK", {'Content-type'=>'text/plain', 'Connection'=>'Keep-Alive'}, "")
  end
end

#on_read_msgpack(data) ⇒ Object

# File 'lib/fluent/plugin/in_http_ex.rb', line 186

def on_read_msgpack(data)
  path_info, query_string = @parser.request_url.split("?", 2)
  params = WEBrick::HTTPUtils.parse_query(query_string)
  @u.feed_each(data) do |obj|
    params["chunked"] = obj
    params["REMOTE_ADDR"] = @remote_addr
    @callback.call(path_info, params)
  end
rescue
  $log.error "on_read_msgpack error: #{$!.to_s}"
  $log.error_backtrace
  close
end