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.



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

def initialize(endpoint, log)
  bind, port = endpoint.split(':')
  @bind = bind
  @port = 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 



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

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

#mount_proc(path, &block) ⇒ Object 



41
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
 
# File 'lib/fluent/rpc.rb', line 41

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 



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

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

#start ⇒ Object 



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

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