Class: Fluent::RPC::Server

Inherits:
Object
  • Object
show all
Defined in:
lib/fluent/rpc.rb

Instance Method Summary collapse

Constructor Details

#initialize(endpoint, log) ⇒ Server

Returns a new instance of Server.

Raises:



22
23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/fluent/rpc.rb', line 22

def initialize(endpoint, log)
  m = endpoint.match(/^\[?(?<host>[0-9a-zA-Z:\-\.]+)\]?:(?<port>[0-9]+)$/)
  raise Fluent::ConfigError, "Invalid rpc_endpoint: #{endpoint}" unless m
  @bind = m[:host]
  @port = m[:port]
  @log = log

  @server = WEBrick::HTTPServer.new(
    BindAddress: @bind,
    Port: @port,
    Logger: WEBrick::Log.new(STDERR, WEBrick::Log::FATAL),
    AccessLog: [],
  )
end

Instance Method Details

#mount(path, servlet, *args) ⇒ Object 



37
38
39
40
 
# File 'lib/fluent/rpc.rb', line 37

def mount(path, servlet, *args)
  @server.mount(path, servlet, *args)
  @log.debug "register #{path} RPC servlet"
end

#mount_proc(path, &block) ⇒ Object 



42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/fluent/rpc.rb', line 42

def mount_proc(path, &block)
  @server.mount_proc(path) { |req, res|
    begin
      code, header, response = block.call(req, res)
    rescue => e
      @log.warn "failed to handle RPC request", path: path, error: e.to_s
      @log.warn_backtrace e.backtrace

      code = 500
      body = {
        'message '=> 'Internal Server Error',
        'error' => "#{e}",
        'backtrace'=> e.backtrace,
      }
    end

    code = 200 if code.nil?
    header = {'Content-Type' => 'application/json'} if header.nil?
    body = if response.nil?
             '{"ok":true}'
           else
             response.body['ok'] = code == 200
             response.body.to_json
           end

    res.status = code
    header.each_pair { |k, v|
      res[k] = v
    }
    res.body = body
  }
  @log.debug "register #{path} RPC handler"
end

#shutdown ⇒ Object 



83
84
85
86
87
88
89
90
91
92
 
# File 'lib/fluent/rpc.rb', line 83

def shutdown
  if @server
    @server.shutdown
    @server = nil
  end
  if @thread
    @thread.join
    @thread = nil
  end
end

#start ⇒ Object 



76
77
78
79
80
81
 
# File 'lib/fluent/rpc.rb', line 76

def start
  @log.debug "listening RPC http server on http://#{@bind}:#{@port}/"
  @thread = Thread.new {
    @server.start
  }
end