Kouhei Sutou
null+****@clear*****
Sat Sep 6 00:05:06 JST 2014
Kouhei Sutou 2014-09-06 00:05:06 +0900 (Sat, 06 Sep 2014) New Revision: dbed46075b94d312ecb9b3318094adf768bbb18d https://github.com/droonga/droonga-engine/commit/dbed46075b94d312ecb9b3318094adf768bbb18d Message: Shutdown clients after engine processes all requests Modified files: lib/droonga/command/droonga_engine_service.rb lib/droonga/fluent_message_receiver.rb Modified: lib/droonga/command/droonga_engine_service.rb (+1 -0) =================================================================== --- lib/droonga/command/droonga_engine_service.rb 2014-09-05 18:44:26 +0900 (ce16ca4) +++ lib/droonga/command/droonga_engine_service.rb 2014-09-06 00:05:06 +0900 (1c19c71) @@ -210,6 +210,7 @@ module Droonga @engine.stop_gracefully do shutdown_worker_process_agent shutdown_internal_message_receiver + @receiver.shutdown_clients logger.trace("loop watchers = #{@loop.watchers}") logger.trace("stop_gracefully: done") end Modified: lib/droonga/fluent_message_receiver.rb (+20 -6) =================================================================== --- lib/droonga/fluent_message_receiver.rb 2014-09-05 18:44:26 +0900 (c54a6d3) +++ lib/droonga/fluent_message_receiver.rb 2014-09-06 00:05:06 +0900 (5b998c9) @@ -55,6 +55,12 @@ module Droonga logger.trace("stop_immediately: done") end + def shutdown_clients + @clients.dup.each do |client| + client.close + end + end + private def start_heartbeat_receiver logger.trace("start_heartbeat_receiver: start") @@ -77,6 +83,9 @@ module Droonga client = Client.new(connection) do |tag, time, record| @on_message.call(tag, time, record) end + client.on_close = lambda do + @clients.delete(client) + end @clients << client end @loop.attach(@server) @@ -94,12 +103,6 @@ module Droonga logger.trace("shutdown_server: done") end - def shutdown_clients - @clients.each do |client| - client.close - end - end - def log_tag "fluent-message-receiver" end @@ -155,16 +158,27 @@ module Droonga class Client include Loggable + attr_accessor :on_close def initialize(io, &on_message) @io = io @on_message = on_message + @on_close = nil @unpacker = MessagePack::Unpacker.new + on_read = lambda do |data| feed(data) end @io.on_read do |data| on_read.call(data) end + + on_close = lambda do + @io = nil + @on_close.call if @on_close + end + @io.on_close do + on_close.call + end end def close -------------- next part -------------- HTML����������������������������...Download
