Navigation Menu

Skip to content

Commit

Permalink
Start and shutdown Forwarder under EngineNode correctly
Browse files Browse the repository at this point in the history
Conflicts:
	lib/droonga/engine_node.rb
  • Loading branch information
piroor committed Jan 6, 2015
1 parent 6639f6b commit 936acaa
Show file tree
Hide file tree
Showing 3 changed files with 41 additions and 7 deletions.
28 changes: 22 additions & 6 deletions lib/droonga/cluster.rb
Expand Up @@ -33,20 +33,36 @@ def initialize(loop)
@state = nil
@on_change = nil

@file_observer = FileObserver.new(loop, Path.cluster_state)
@file_observer.on_change = lambda do
reload
end

reload
end

def start_observe
return if @file_observer
@file_observer = FileObserver.new(@loop, Path.cluster_state)
@file_observer.on_change = lambda do
reload
end
@file_observer.start
end

def stop_observe
return unless @file_observer
@file_observer.stop
@file_observer = nil
end

def start
engine_nodes.each do |node|
node.start
end
start_observe
end

def shutdown
stop_observe
engine_nodes.each do |node|
node.shutdown
end
end

def reload
Expand Down Expand Up @@ -85,7 +101,7 @@ def forward(message, destination)
receiver_node_name = receiver.match(/\A[^:]+:\d+\/[^.]+/).to_s
@engine_nodes.each do |node|
if node.name == receiver_node_name
node.forwarder.forward(message, destination)
node.forward(message, destination)
return true
end
end
Expand Down
4 changes: 3 additions & 1 deletion lib/droonga/engine.rb
Expand Up @@ -52,7 +52,7 @@ def start
@on_ready.call if @on_ready
end
@state.start
@cluster.start_observe
@cluster.start
@node_metadata_observer.start
@dispatcher.start
logger.trace("start: done")
Expand All @@ -66,6 +66,7 @@ def stop_gracefully
logger.trace("stop_gracefully/on_finish: start")
save_last_processed_message_timestamp
@dispatcher.stop_gracefully do
@cluster.shutdown
@state.shutdown
yield
end
Expand All @@ -88,6 +89,7 @@ def stop_immediately
@cluster.stop_observe
@node_metadata_observer.stop
@dispatcher.stop_immediately
@cluster.shutdown
@state.shutdown
logger.trace("stop_immediately: done")
end
Expand Down
16 changes: 16 additions & 0 deletions lib/droonga/engine_node.rb
Expand Up @@ -28,6 +28,22 @@ def initialize(name, state, sender_role, loop)
@forwarder = Forwarder.new(loop, :buffering => true)
end

def start
logger.trace("start: start")
@forwarder.start
logger.trace("start: done")
end

def shutdown
logger.trace("shutdown: start")
@forwarder.shutdown
logger.trace("shutdown: done")
end

def forward(message, destination)
@forwarder.forward(message, destination)
end

def live?
@state.nil? or @state["live"]
end
Expand Down

0 comments on commit 936acaa

Please sign in to comment.