Distribute from outside of Planner.

lib/droonga/dispatcher.rb

@@ -212,7 +212,8 @@ def farm_path(route)
 
     def process_input_message(message)
       adapted_message = @adapter_runner.adapt_input(message)
-      @planner.process(adapted_message["type"], adapted_message)
+      plan = @planner.process(adapted_message["type"], adapted_message)
+      @planner.distribute(plan)
     rescue Droonga::Pluggable::UnknownPlugin => error
       raise UnknownCommand.new(error.command, message["dataset"])
     end

lib/droonga/planner_plugin.rb

@@ -36,14 +36,14 @@ def scatter(message, options={})
       planner.scatter
       planner.key = options[:key]
       planner.reduce(options[:reduce])
-      distribute(planner.plan)
+      planner.plan
     end
 
     def broadcast(message, options={})
       planner = DistributedCommandPlanner.new(message)
       planner.broadcast(:write => options[:write])
       planner.reduce(options[:reduce])
-      distribute(planner.plan)
+      planner.plan
     end
 
     private

lib/droonga/pluggable.rb

@@ -42,9 +42,10 @@ def process(command, *arguments)
       $log.trace("#{log_tag}: process: start: <#{command}>",
                  :plugin => plugin.class)
       raise UnknownPlugin.new(command) if plugin.nil?
-      plugin.process(command, *arguments)
+      result = plugin.process(command, *arguments)
       $log.trace("#{log_tag}: process: done: <#{command}>",
                  :plugin => plugin.class)
+      result
     end
 
     private

lib/droonga/plugin/planner/search.rb

@@ -25,7 +25,7 @@ class SearchPlanner < Droonga::PlannerPlugin
     command :search
     def search(message)
       planner = DistributedSearchPlanner.new(message)
-      distribute(planner.plan)
+      planner.plan
     end
   end
 end

test/unit/plugin/planner/test_search.rb

@@ -56,7 +56,7 @@ def test_distribute
       },
     }
 
-    @plugin.process("search", envelope)
+    @planner.distribute(@plugin.process("search", envelope))
 
     message = []