Merge pull request #298 from basho/features/lrb/riak-client-2.6.0

Riak Ruby Client 2.6.0

Author: lukebakken

.rubocop_todo.yml

@@ -37,7 +37,6 @@ Lint/AssignmentInCondition:
     - 'lib/riak/client/beefcake/object_methods.rb'
     - 'lib/riak/client/beefcake/time_series_list_operator.rb'
     - 'lib/riak/client/beefcake_protobuffs_backend.rb'
-    - 'lib/riak/multiget.rb'
 
 # Offense count: 4
 # Cop supports --auto-correct.

Makefile

@@ -32,6 +32,7 @@ help:
 	@echo ' test             - Run unit & integration tests '
 	@echo ' unit-test        - Run unit tests               '
 	@echo ' integration-test - Run integration tests        '
+	@echo ' timeseries-test  - Run timeseries tests         '
 	@echo ' security-test    - Run security tests           '
 	@echo '-------------------------------------------------'
 	@echo ''
@@ -56,6 +57,10 @@ integration-test:
 	@cp -f $(TCY).example $(TCY)
 	@bundle exec rake spec:integration
 
+timeseries-test:
+	@cp -f $(TCY).example $(TCY)
+	@bundle exec rake spec:time_series
+
 security-test:
 	@cp -f $(TCY).example $(TCY) && \
 		echo 'authentication:' >> $(TCY) && \
@@ -64,7 +69,7 @@ security-test:
 		echo "  ca_file: $(CA_CERT)" >> $(TCY)
 	@bundle exec rake spec:security
 
-test: lint integration-test
+test: lint unit-test integration-test
 
 gemspec_validate:
 	@bundle exec rake gemspec

lib/riak/bucket.rb

@@ -114,7 +114,7 @@ def get(key, options = {})
     # @return [Hash<String, Riak::RObject>] hash of keys to objects
     def get_many(keys)
       pairs = keys.map{|k| [self, k]}
-      results = Multiget.get_all @client, pairs
+      results = Multiget.perform @client, pairs
       results.keys.inject(Hash.new) do |mem, var|
         mem[var[1]] = results[var]
         mem

lib/riak/client.rb

@@ -18,6 +18,7 @@
 require 'riak/bucket_properties'
 require 'riak/bucket_type'
 require 'riak/multiget'
+require 'riak/multiexist'
 require 'riak/secondary_index'
 require 'riak/search'
 require 'riak/stamp'
@@ -69,7 +70,10 @@ class Client
     attr_reader :protobuffs_pool
 
     # @return [Integer] The number of threads for multiget requests
-    attr_reader :multiget_threads
+    attr_reader :multi_threads
+
+    # @deprecated use multi_threads
+    alias_method :multiget_threads, :multi_threads
 
     # @return [Hash] The authentication information this client will use.
     attr_reader :authentication
@@ -123,7 +127,7 @@ def initialize(options = {})
 
       self.protobuffs_backend = options[:protobuffs_backend] || :Beefcake
       self.client_id          = options[:client_id]          if options[:client_id]
-      self.multiget_threads   = options[:multiget_threads]
+      self.multi_threads      = options[:multi_threads] || options[:multiget_threads]
       @authentication         = options[:authentication] && options[:authentication].symbolize_keys
       self.max_retries        = options[:max_retries]        || 2
       @connect_timeout        = options[:connect_timeout]
@@ -195,20 +199,23 @@ def choose_node(nodes = self.nodes)
     # If set to nil, defaults to twice the number of nodes.
     # @param [Integer] count The number of threads to use.
     # @raise [ArgumentError] when a non-nil, non-positive-Integer count is given
-    def multiget_threads=(count)
+    def multi_threads=(count)
       if count.nil?
-        @multiget_threads = nodes.length * 2
+        @multi_threads = nodes.length * 2
         return
       end
 
       if count.is_a?(Integer) && count > 0
-        @multiget_threads = count
+        @multi_threads = count
         return
       end
 
-      raise ArgumentError, t("invalid_multiget_thread_count")
+      raise ArgumentError, t("invalid_multiget_thread_count") # TODO: rename to invalid_multi_thread_count
     end
 
+    # @deprecated use multi_threads=
+    alias_method :multiget_threads=, :multi_threads=
+
     # Set the client ID for this client. Must be a string or Fixnum value 0 =<
     # value < MAX_CLIENT_ID.
     # @param [String, Fixnum] value The internal client ID used by Riak to route responses
@@ -275,7 +282,7 @@ def get_preflist(bucket, key, type = nil, options = {  })
 
     # Get multiple objects in parallel.
     def get_many(pairs)
-      Multiget.get_all self, pairs
+      Multiget.perform self, pairs
     end
 
     # Get an object. See Bucket#get

lib/riak/multi.rb

@@ -0,0 +1,101 @@
+require 'riak/client'
+require 'riak/bucket'
+
+module Riak
+  # Coordinates a parallel operation for multiple keys.
+  class Multi
+    include Util::Translation
+
+    # @return [Riak::Client] the associated client
+    attr_reader :client
+
+    # @return [Array<Bucket, String>] fetch_list an {Array} of {Bucket} and {String} keys
+    attr_reader :keys
+
+    # @return [Hash<fetch_list_entry, RObject] result_hash a {Hash} of {Bucket} and {String} key pairs to {RObject} instances
+    attr_accessor :result_hash
+
+    # @return [Integer] The number of threads to use
+    attr_accessor :thread_count
+
+    # Perform a Riak Multi operation.
+    # @param [Client] client the {Riak::Client} that will perform the operation
+    # @param [Array<Bucket, String>] keys an {Array} of {Bucket} and {String} keys to work with
+    # @return [Hash<key, RObject] result_hash a {Hash} of {Bucket} and {String} key pairs to {RObject} instances
+    def self.perform(client, keys)
+      multi = new client, keys
+      multi.perform
+      multi.results
+    end
+
+    # Create a Riak Multi operation.
+    # @param [Client] client the {Riak::Client} that will perform the multiget
+    # @param [Array<Bucket, String>] keys an {Array} of {Bucket} and {String} keys to work on
+    # @raise [ArgumentError] when a non-positive-Integer count is given for threads
+    def initialize(client, keys)
+      raise ArgumentError, t('client_type', :client => client.inspect) unless client.is_a? Riak::Client
+      raise ArgumentError, t('array_type', :array => keys.inspect) unless keys.is_a? Array
+
+      self.thread_count = client.multi_threads
+      validate_keys keys
+      @client = client
+      @keys = keys.uniq
+      self.result_hash = {}
+      @finished = false
+    end
+
+    # Starts the parallelized operation
+    def perform
+      queue = keys.dup
+      queue_mutex = Mutex.new
+      result_mutex = Mutex.new
+
+      @threads = 1.upto(thread_count).map do |_node|
+        Thread.new do
+          loop do
+            pair = queue_mutex.synchronize do
+              queue.shift
+            end
+
+            break if pair.nil?
+
+            found = work(*pair)
+            result_mutex.synchronize do
+              result_hash[pair] = found
+            end
+          end
+        end
+      end
+    end
+
+    def results
+      wait_for_finish
+      result_hash
+    end
+
+    def finished?
+      @finished ||= @threads && @threads.none?(&:alive?)
+    end
+    alias :finished :finished? # deprecated
+
+    def wait_for_finish
+      return if finished?
+      @threads.each(&:join)
+      @finished = true
+    end
+
+    private
+
+    def work(_bucket, _key)
+      raise NotImplementedError
+    end
+
+    def validate_keys(keys)
+      erroneous = keys.detect do |bucket, key|
+        !bucket.is_a?(Bucket) || !key.is_a?(String)
+      end
+      return unless erroneous
+      raise ArgumentError, t('fetch_list_type', problem: erroneous) # TODO: should be keys_type
+    end
+  end
+end

lib/riak/multiexist.rb

@@ -0,0 +1,14 @@
+require 'riak/client'
+require 'riak/bucket'
+require 'riak/multi'
+
+module Riak
+  # Coordinates a parallel exist? operation for multiple keys.
+  class Multiexist < Multi
+    private
+
+    def work(bucket, key)
+      bucket.exists?(key)
+    end
+  end
+end

lib/riak/multiget.rb

@@ -1,122 +1,22 @@
 require 'riak/client'
 require 'riak/bucket'
+require 'riak/multi'
 
 module Riak
-  # Coordinates a parallel fetch operation for multiple values.
-  class Multiget
-    include Util::Translation
-
-    # @return [Riak::Client] the associated client
-    attr_reader :client
-
-    # @return [Array<Bucket, String>] fetch_list an {Array} of {Bucket} and {String} keys to fetch
-    attr_reader :fetch_list
-
-    # @return [Hash<fetch_list_entry, RObject] result_hash a {Hash} of {Bucket} and {String} key pairs to {RObject} instances
-    attr_accessor :result_hash
-
-    # @return [Boolean] finished if the fetch operation has completed
-    attr_reader :finished
-
-    # @return [Integer] The number of threads to use
-    attr_accessor :thread_count
-
-    # Perform a Riak Multiget operation.
-    # @param [Client] client the {Riak::Client} that will perform the multiget
-    # @param [Array<Bucket, String>] fetch_list an {Array} of {Bucket} and {String} keys to fetch
-    # @return [Hash<fetch_list_entry, RObject] result_hash a {Hash} of {Bucket} and {String} key pairs to {RObject} instances
-    def self.get_all(client, fetch_list)
-      multi = new client, fetch_list
-      multi.fetch
-      multi.results
-    end
-
-    # Create a Riak Multiget operation.
-    # @param [Client] client the {Riak::Client} that will perform the multiget
-    # @param [Array<Bucket, String>] fetch_list an {Array} of {Bucket} and {String} keys to fetch
-    def initialize(client, fetch_list)
-      raise ArgumentError, t('client_type', :client => client.inspect) unless client.is_a? Riak::Client
-      raise ArgumentError, t('array_type', :array => fetch_list.inspect) unless fetch_list.is_a? Array
-
-      validate_fetch_list fetch_list
-      @client, @fetch_list = client, fetch_list.uniq
-      self.result_hash = Hash.new
-      @finished = false
-      self.thread_count = client.multiget_threads
-    end
-
-    # Starts the parallelized fetch operation
-    # @raise [ArgumentError] when a non-positive-Integer count is given
-    def fetch
-      queue = fetch_list.dup
-      queue_mutex = Mutex.new
-      result_mutex = Mutex.new
-
-      unless thread_count.is_a?(Integer) && thread_count > 0
-        raise ArgumentError, t("invalid_multiget_thread_count")
-      end
-
-      @threads = 1.upto(thread_count).map do |_node|
-        Thread.new do
-          loop do
-            pair = queue_mutex.synchronize do
-              queue.shift
-            end
-
-            break if pair.nil?
-
-            found = attempt_fetch(*pair)
-            result_mutex.synchronize do
-              result_hash[pair] = found
-            end
-          end
-        end
-      end
-    end
-
-    def results
-      wait_for_finish
-      result_hash
-    end
-
-    def finished?
-      set_finished_for_thread_liveness
-      finished
-    end
-
-    def wait_for_finish
-      return if finished?
-      @threads.each {|t| t.join }
-      @finished = true
+  # Coordinates a parallel fetch operation for multiple keys.
+  class Multiget < Multi
+    # @deprecated use perform
+    class << self
+      alias_method :get_all, :perform
     end
 
     private
 
-    def attempt_fetch(bucket, key)
+    def work(bucket, key)
       bucket[key]
     rescue Riak::FailedRequest => e
       raise e unless e.not_found?
       nil
     end
-
-    def set_finished_for_thread_liveness
-      return if @finished # already done
-
-      all_dead = @threads.none? {|t| t.alive? }
-      return unless all_dead # still working
-
-      @finished = true
-      return
-    end
-
-    def validate_fetch_list(fetch_list)
-      return unless erroneous = fetch_list.detect do |e|
-        bucket, key = e
-        next true unless bucket.is_a? Bucket
-        next true unless key.is_a? String
-      end
-
-      raise ArgumentError, t('fetch_list_type', :problem => erroneous)
-    end
   end
 end

riak-client.gemspec

@@ -18,7 +18,7 @@ Gem::Specification.new do |gem|
   gem.add_development_dependency 'rake', '~> 10.1'
   gem.add_development_dependency 'rspec', '~> 3.0'
   gem.add_development_dependency 'rubocop', '~> 0.40.0'
-  gem.add_development_dependency 'simplecov', '~> 0.10'
+  gem.add_development_dependency 'single_cov', '0.5.7'
   gem.add_development_dependency 'yard', '~> 0.8'
 
   gem.add_runtime_dependency 'beefcake', '~> 1.1'

spec/integration/riak/bucket_types_spec.rb

@@ -2,7 +2,6 @@
 require 'riak'
 
 describe 'Bucket Types', test_client: true, integration: true do
-
   describe 'nested bucket types API' do
     let(:bucket_type){ test_client.bucket_type 'yokozuna' }