Accept "--records-per-second" option for droonga-engine-absorb-data a…

…nd droonga-engine-join
droonga · Nov 27, 2014 · c7d2e03 · c7d2e03
1 parent ae14394
commit c7d2e03
Show file tree

Hide file tree

Showing 3 changed files with 27 additions and 2 deletions.
diff --git a/bin/droonga-engine-absorb-data b/bin/droonga-engine-absorb-data
@@ -57,6 +57,7 @@ class AbsorbDataCommand
     options.tag     = Droonga::CatalogGenerator::DEFAULT_TAG
     options.dataset = Droonga::CatalogGenerator::DEFAULT_DATASET
     options.remote  = true
+    options.messages_per_second = Droonga::DataAbsorber::DEFAULT_MESSAGES_PER_SECOND
     parser = OptionParser.new
     parser.version = Droonga::Engine::VERSION
 
@@ -93,6 +94,12 @@ class AbsorbDataCommand
               "(#{options.dataset})") do |dataset|
       options.dataset = dataset
     end
+    parser.on("--records-per-second=N", Integer,
+              "Maximum number of records per second to be absorbed.",
+              "'#{Droonga::Client::RateLimiter::NO_LIMIT}' means no limit.",
+              "(#{options.messages_per_second})") do |n|
+      options.messages_per_second = n
+    end
 
     parser.parse!(ARGV)
     @options = options
@@ -134,6 +141,7 @@ class AbsorbDataCommand
       :destination_host => @options.destination_host,
       :port             => @options.port,
       :tag              => @options.tag,
+      :messages_per_second => @options.messages_per_second,
     }
     Droonga::DataAbsorber.new(absorber_options)
   end
@@ -145,7 +153,8 @@ class AbsorbDataCommand
                        "source"  => @options.source_host,
                        "port"    => @options.port,
                        "tag"     => @options.tag,
-                       "dataset" => @options.dataset)
+                       "dataset" => @options.dataset,
+                       "messages_per_second" => @options.messages_per_second)
     last_progress = ""
     while true
       sleep(3)

diff --git a/bin/droonga-engine-join b/bin/droonga-engine-join
@@ -64,6 +64,11 @@ class JoinCommand
       option.on(:tag=,
                 "Tag name of the soruce cluster to be connected.",
                 :default => Droonga::CatalogGenerator::DEFAULT_TAG)
+      option.on("records-per-second=",
+                :as => Integer,
+                "Maximum number of records per second to be copied. " +
+                  "'#{Droonga::Client::RateLimiter::NO_LIMIT}' means no limit.",
+                :default => Droonga::DataAbsorber::DEFAULT_MESSAGES_PER_SECOND)
     end
     @options = options
   rescue Slop::MissingOptionError => error
@@ -98,6 +103,7 @@ class JoinCommand
       :destination_host => @options[:host],
       :port             => @options[:port],
       :tag              => @options[:tag],
+      :messages_per_second => @options["records-per-second"],
     }
     Droonga::DataAbsorber.new(absorber_options)
   end

diff --git a/lib/droonga/command/remote.rb b/lib/droonga/command/remote.rb
@@ -122,6 +122,10 @@ def dataset_name
           @params["dataset"]
         end
 
+        def messages_per_second
+          @params["messages_per_second"]
+        end
+
         def valid_params?
           have_required_params? and
             valid_node?(source_node) and
@@ -227,7 +231,8 @@ def absorb_data
                               :source_host      => source_host,
                               :destination_host => joining_host,
                               :port             => port,
-                              :tag              => tag)
+                              :tag              => tag,
+                              :messages_per_second => messages_per_second)
           status.delete(:absorbing)
           sleep(1)
         end
@@ -265,6 +270,7 @@ def process
                               :destination_host => host,
                               :port             => port,
                               :tag              => tag,
+                              :messages_per_second => messages_per_second,
                               :client           => "droonga-send")
           status.delete(:absorbing)
         end
@@ -285,6 +291,10 @@ def port
         def tag
           @tag ||= @params["tag"]
         end
+
+        def messages_per_second
+          @messages_per_second ||= @params["messages_per_second"]
+        end
       end
 
       class ModifyReplicasBase < Base