Add "targetRole" field to the envelope to ignore not acceptable messages

droonga · Apr 28, 2015 · 02f10e3 · 02f10e3
1 parent 66a855b
commit 02f10e3
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 0 deletions.
diff --git a/bin/droonga-engine-join b/bin/droonga-engine-join
@@ -234,6 +234,7 @@ module Droonga
 
         :messages_per_second => @options["records-per-second"],
         :progress_interval_seconds => @options["progress-interval-seconds"],
+        :target_role => NodeRole::ABSORB_DESTINATION,
 
         :client_options   => {
           :backend => :coolio,

diff --git a/lib/droonga/data_absorber_client.rb b/lib/droonga/data_absorber_client.rb
@@ -57,6 +57,7 @@ def initialize(params)
                                DEFAULT_MESSAGES_PER_SECOND
       @progress_interval_seconds = @params[:progress_interval_seconds] ||
                                      DEFAULT_PROGRESS_INTERVAL_SECONDS
+      @target_role = @params[:target_role]
 
       @host    = @params[:host]    || DEFAULT_HOST
       @port    = @params[:port]    || DEFAULT_PORT
@@ -91,6 +92,7 @@ def run
           "dataset" => @source_dataset,
           "messagesPerSecond" => @messages_per_second,
           "progressIntervalSeconds" => @progress_interval_seconds,
+          "targetRole" => @target_role,
         },
       }
       client = Droonga::Client.new(destination_client_options)

diff --git a/lib/droonga/plugins/system/absorb_data.rb b/lib/droonga/plugins/system/absorb_data.rb
@@ -94,6 +94,7 @@ def do_absorb(&block)
               @dumper_error_message = @dumper.run(dump_options) do |message|
                 begin
                   message["dataset"] = current_dataset
+                  message["targetRole"] = target_role if target_role
                   message["xSender"] = "system.absorb-data"
                   @messenger.forward(message,
                                      "to"   => my_node_name,
@@ -335,6 +336,10 @@ def current_dataset
             @request.dataset
           end
 
+          def target_role
+            @request.request["targetRole"]
+          end
+
           def prepare_progress_interval_seconds
             interval_seconds = @request.request["progressIntervalSeconds"] ||
                                  DEFAULT_PROGRESS_INTERVAL_SECONDS