Merge pull request #173 from datastax/3.0.1

stamhankar999 · stamhankar999 · commit 1f6534318f4f · 2016-05-29T15:09:30.000-07:00
Assorted 3.0.1 internal fixes.
diff --git a/Gemfile.lock b/Gemfile.lock
@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    cassandra-driver (3.0.0)
+    cassandra-driver (3.0.1.beta.1)
       ione (~> 1.2)
 
 GEM
diff --git a/lib/cassandra.rb b/lib/cassandra.rb
@@ -293,28 +293,7 @@ def self.cluster(options = {})
   # @return [Cassandra::Future<Cassandra::Cluster>] a future resolving to the
   #   cluster instance.
   def self.cluster_async(options = {})
-    options = validate_and_massage_options(options)
-    hosts = []
-
-    Array(options.fetch(:hosts, '127.0.0.1')).each do |host|
-      case host
-      when ::IPAddr
-        hosts << host
-      when ::String # ip address or hostname
-        Resolv.each_address(host) do |ip|
-          hosts << ::IPAddr.new(ip)
-        end
-      else
-        raise ::ArgumentError, ":hosts must be String or IPAddr, #{host.inspect} given"
-      end
-    end
-
-    if hosts.empty?
-      raise ::ArgumentError,
-            ":hosts #{options[:hosts].inspect} could not be resolved to any ip address"
-    end
-
-    hosts.shuffle!
+    options, hosts = validate_and_massage_options(options)
   rescue => e
     futures = options.fetch(:futures_factory) { return Future::Error.new(e) }
     futures.error(e)
@@ -664,14 +643,14 @@ def self.validate_and_massage_options(options)
 
       case address_resolution
       when :none
-        # do nothing
+      # do nothing
       when :ec2_multi_region
         options[:address_resolution_policy] =
           AddressResolution::Policies::EC2MultiRegion.new
       else
         raise ::ArgumentError,
               ':address_resolution must be either :none or :ec2_multi_region, ' \
-                  "#{address_resolution.inspect} given"
+                "#{address_resolution.inspect} given"
       end
     end
 
@@ -757,7 +736,31 @@ def self.validate_and_massage_options(options)
         end
       end
     end
-    options
+
+    # Get host addresses.
+    hosts = []
+
+    Array(options.fetch(:hosts, '127.0.0.1')).each do |host|
+      case host
+      when ::IPAddr
+        hosts << host
+      when ::String # ip address or hostname
+        Resolv.each_address(host) do |ip|
+          hosts << ::IPAddr.new(ip)
+        end
+      else
+        raise ::ArgumentError, ":hosts must be String or IPAddr, #{host.inspect} given"
+      end
+    end
+
+    if hosts.empty?
+      raise ::ArgumentError,
+            ":hosts #{options[:hosts].inspect} could not be resolved to any ip address"
+    end
+
+    hosts.shuffle!
+
+    [options, hosts]
   end
 
   # @private
diff --git a/lib/cassandra/auth.rb b/lib/cassandra/auth.rb
@@ -34,7 +34,7 @@ module Auth
     #
     # @see Cassandra::Auth::Providers
     class Provider
-      # @!method create_authenticator(authentication_class, protocol_version)
+      # @!method create_authenticator(authentication_class, host)
       #
       # Create a new authenticator object. This method will be called once per
       # connection that requires authentication. The auth provider can create
@@ -45,6 +45,8 @@ class Provider
       #
       # @param authentication_class [String] the authentication class used by
       #   the server.
+      # @param host [Cassandra::Host] the node to whom we're authenticating.
+      #
       # @return [Cassandra::Auth::Authenticator, nil] an object with an
       #   interface matching {Cassandra::Auth::Authenticator} or `nil` if the
       #   authentication class is not supported.
diff --git a/lib/cassandra/cluster/connector.rb b/lib/cassandra/cluster/connector.rb
@@ -154,7 +154,7 @@ def do_connect(host)
                 supported_cql_versions.first :
                 '3.1.0'
 
-            startup_connection(connection, cql_version, compression)
+            startup_connection(host, connection, cql_version, compression)
           end
           f.fallback do |error|
             case error
@@ -200,7 +200,7 @@ def do_connect(host)
         end
       end
 
-      def startup_connection(connection, cql_version, compression)
+      def startup_connection(host, connection, cql_version, compression)
         connection.send_request(Protocol::StartupRequest.new(cql_version, compression),
                                 @execution_options.timeout).flat_map do |r|
           case r
@@ -213,12 +213,9 @@ def startup_connection(connection, cql_version, compression)
                 Ione::Future.failed(cannot_authenticate_error)
               end
             else
-              authenticator = @connection_options.create_authenticator(
-                r.authentication_class)
+              authenticator = @connection_options.create_authenticator(r.authentication_class, host)
               if authenticator
-                challenge_response_cycle(connection,
-                                         authenticator,
-                                         authenticator.initial_response)
+                challenge_response_cycle(connection, authenticator, authenticator.initial_response)
               else
                 Ione::Future.failed(cannot_authenticate_error)
               end
@@ -283,7 +280,7 @@ def challenge_response_cycle(connection, authenticator, token)
           case r
           when Protocol::AuthChallengeResponse
             token = authenticator.challenge_response(r.token)
-            challenge_response_cycle(pending_connection, authenticator, token)
+            challenge_response_cycle(connection, authenticator, token)
           when Protocol::AuthSuccessResponse
             begin
               authenticator.authentication_successful(r.token)
diff --git a/lib/cassandra/cluster/options.rb b/lib/cassandra/cluster/options.rb
@@ -78,8 +78,17 @@ def compression
         @compressor && @compressor.algorithm
       end
 
-      def create_authenticator(authentication_class)
-        @auth_provider && @auth_provider.create_authenticator(authentication_class)
+      def create_authenticator(authentication_class, host)
+        if @auth_provider
+          # Auth providers should take an auth-class and host, but they used to not, so for backward compatibility
+          # we figure out if this provider does, and if so send both args, otherwise just send the auth-class.
+
+          if @auth_provider.method(:create_authenticator).arity == 1
+            @auth_provider.create_authenticator(authentication_class)
+          else
+            @auth_provider.create_authenticator(authentication_class, host)
+          end
+        end
       end
 
       def connections_per_local_node
diff --git a/lib/cassandra/driver.rb b/lib/cassandra/driver.rb
@@ -92,23 +92,25 @@ def self.let(name, &block)
                                      schema_fetcher)
     end
 
+    let(:cluster_klass) { Cluster }
+
     let(:cluster) do
-      Cluster.new(logger,
-                  io_reactor,
-                  executor,
-                  control_connection,
-                  cluster_registry,
-                  cluster_schema,
-                  cluster_metadata,
-                  execution_options,
-                  connection_options,
-                  load_balancing_policy,
-                  reconnection_policy,
-                  retry_policy,
-                  address_resolution_policy,
-                  connector,
-                  futures_factory,
-                  timestamp_generator)
+      cluster_klass.new(logger,
+                        io_reactor,
+                        executor,
+                        control_connection,
+                        cluster_registry,
+                        cluster_schema,
+                        cluster_metadata,
+                        execution_options,
+                        connection_options,
+                        load_balancing_policy,
+                        reconnection_policy,
+                        retry_policy,
+                        address_resolution_policy,
+                        connector,
+                        futures_factory,
+			timestamp_generator)
     end
 
     let(:execution_options) do
diff --git a/lib/cassandra/result.rb b/lib/cassandra/result.rb
@@ -76,8 +76,8 @@ def next_page(options = nil)
     #
     # @note `:paging_state` option will be ignored.
     #
-    # @return [Cassandra::Future<Cassandra::Result, nil>] `nil` if last
-    #   page
+    # @return [Cassandra::Future<Cassandra::Result>] a future that resolves to a new Result if there is a new page,
+    #   `nil` otherwise.
     #
     # @see Cassandra::Session#execute
     def next_page_async(options = nil)
diff --git a/lib/cassandra/session.rb b/lib/cassandra/session.rb
@@ -89,20 +89,13 @@ def execute_async(statement, options = nil)
 
       case statement
       when ::String
-        @client.query(Statements::Simple.new(statement,
-                                             options.arguments,
-                                             options.type_hints,
-                                             options.idempotent?),
-                      options)
-      when Statements::Simple
-        @client.query(statement, options)
-      when Statements::Prepared
-        @client.execute(statement.bind(options.arguments), options)
-      when Statements::Bound
-        @client.execute(statement, options)
-      when Statements::Batch
-        Util.assert_not_empty(statement.statements) { 'batch cannot be empty' }
-        @client.batch(statement, options)
+        Statements::Simple.new(statement,
+                               options.arguments,
+                               options.type_hints,
+                               options.idempotent?).accept(@client,
+                                                           options)
+      when Statement
+        statement.accept(@client, options)
       else
         @futures.error(::ArgumentError.new("unsupported statement #{statement.inspect}"))
       end
diff --git a/lib/cassandra/statement.rb b/lib/cassandra/statement.rb
@@ -23,5 +23,10 @@ module Statement
     def idempotent?
       !!@idempotent
     end
+
+    # @private
+    def accept(client, options)
+      raise NotImplementedError, "#{self.class} must implement :accept method"
+    end
   end
 end
diff --git a/lib/cassandra/statements/batch.rb b/lib/cassandra/statements/batch.rb
@@ -113,6 +113,12 @@ def add(statement, options = nil)
         self
       end
 
+      # @private
+      def accept(client, options)
+        Util.assert_not_empty(statements) { 'batch cannot be empty' }
+        client.batch(self, options)
+      end
+
       # Determines whether or not the statement is safe to retry on timeout
       # Batches are idempotent only when all statements in a batch are.
       # @return [Boolean] whether the statement is safe to retry on timeout
diff --git a/lib/cassandra/statements/bound.rb b/lib/cassandra/statements/bound.rb
@@ -46,6 +46,11 @@ def initialize(cql,
         @idempotent      = idempotent
       end
 
+      # @private
+      def accept(client, options)
+        client.execute(self, options)
+      end
+
       # @return [String] a CLI-friendly bound statement representation
       def inspect
         "#<#{self.class.name}:0x#{object_id.to_s(16)} @cql=#{@cql.inspect} " \
diff --git a/lib/cassandra/statements/prepared.rb b/lib/cassandra/statements/prepared.rb
@@ -155,6 +155,11 @@ def execution_info
                                           nil)
       end
 
+      # @private
+      def accept(client, options)
+        client.execute(bind(options.arguments), options)
+      end
+
       # @return [String] a CLI-friendly prepared statement representation
       def inspect
         "#<#{self.class.name}:0x#{object_id.to_s(16)} @cql=#{@cql.inspect}>"
diff --git a/lib/cassandra/statements/simple.rb b/lib/cassandra/statements/simple.rb
@@ -93,6 +93,11 @@ def initialize(cql, params = nil, type_hints = nil, idempotent = false)
         @idempotent   = idempotent
       end
 
+      # @private
+      def accept(client, options)
+        client.query(self, options)
+      end
+
       # @return [String] a CLI-friendly simple statement representation
       def inspect
         "#<#{self.class.name}:0x#{object_id.to_s(16)} @cql=#{@cql.inspect} " \
diff --git a/lib/cassandra/statements/void.rb b/lib/cassandra/statements/void.rb
@@ -23,6 +23,11 @@ module Statements
     class Void
       include Statement
 
+      # @private
+      def accept(client, options)
+        nil
+      end
+
       # Returns nothing
       # @return [nil] there is no cql for the void statement
       def cql
diff --git a/lib/cassandra/version.rb b/lib/cassandra/version.rb
@@ -17,5 +17,5 @@
 #++
 
 module Cassandra
-  VERSION = '3.0.0'.freeze
+  VERSION = '3.0.1.beta.1'.freeze
 end
diff --git a/spec/cassandra/session_spec.rb b/spec/cassandra/session_spec.rb
@@ -35,7 +35,7 @@ module Cassandra
             statement = double('simple statement')
 
             expect(Statements::Simple).to receive(:new).once.with(cql, EMPTY_LIST, EMPTY_LIST, false).and_return(statement)
-            expect(client).to receive(:query).once.with(statement, session_options).and_return(promise)
+            expect(statement).to receive(:accept).once.with(client, session_options).and_return(promise)
             expect(session.execute_async(cql)).to eq(promise)
           end
         end
@@ -48,7 +48,7 @@ module Cassandra
             statement = double('simple statement')
 
             expect(Statements::Simple).to receive(:new).once.with(cql, [1], [], false).and_return(statement)
-            expect(client).to receive(:query).once.with(statement, session_options.override(arguments: [1])).and_return(promise)
+            expect(statement).to receive(:accept).once.with(client, session_options.override(arguments: [1])).and_return(promise)
             expect(session.execute_async(cql, arguments: [1])).to eq(promise)
           end
         end
@@ -61,7 +61,7 @@ module Cassandra
             statement = double('simple statement')
 
             expect(Statements::Simple).to receive(:new).once.with(cql, [1], [:int], false).and_return(statement)
-            expect(client).to receive(:query).once.with(statement, session_options.override(arguments: [1], type_hints: [:int])).and_return(promise)
+            expect(statement).to receive(:accept).once.with(client, session_options.override(arguments: [1], type_hints: [:int])).and_return(promise)
             expect(session.execute_async(cql, arguments: [1], type_hints: [:int])).to eq(promise)
           end
         end
@@ -75,7 +75,7 @@ module Cassandra
             statement = double('simple statement')
 
             expect(Statements::Simple).to receive(:new).once.with(cql, EMPTY_LIST, EMPTY_LIST, false).and_return(statement)
-            expect(client).to receive(:query).once.with(statement, session_options.override(options)).and_return(promise)
+            expect(statement).to receive(:accept).once.with(client, session_options.override(options)).and_return(promise)
             expect(session.execute_async(cql, options)).to eq(promise)
           end
         end
diff --git a/spec/cassandra_spec.rb b/spec/cassandra_spec.rb
@@ -8,8 +8,10 @@
 # noinspection RubyClassModuleNamingConvention
 module C
   class << self
-    # noinspection RubyResolve
-    alias :validate :validate_and_massage_options
+    def validate(*args)
+      # Actually, we only care about option-validation, and the method returns options + hosts.
+      validate_and_massage_options(*args).first
+    end
   end
 end
 

Original file line number	Diff line number	Diff line change
`@@ -34,7 +34,7 @@ module Auth`
`34`	`34`	`#`
`35`	`35`	`# @see Cassandra::Auth::Providers`
`36`	`36`	`class Provider`
`37`		`- # @!method create_authenticator(authentication_class, protocol_version)`
	`37`	`+ # @!method create_authenticator(authentication_class, host)`
`38`	`38`	`#`
`39`	`39`	`# Create a new authenticator object. This method will be called once per`
`40`	`40`	`# connection that requires authentication. The auth provider can create`
`@@ -45,6 +45,8 @@ class Provider`
`45`	`45`	`#`
`46`	`46`	`# @param authentication_class [String] the authentication class used by`
`47`	`47`	`# the server.`
	`48`	`+ # @param host [Cassandra::Host] the node to whom we're authenticating.`
	`49`	`+ #`
`48`	`50`	`# @return [Cassandra::Auth::Authenticator, nil] an object with an`
`49`	`51`	# interface matching {Cassandra::Auth::Authenticator} or `nil` if the
`50`	`52`	`# authentication class is not supported.`
Original file line number	Diff line number	Diff line change
`@@ -76,8 +76,8 @@ def next_page(options = nil)`
`76`	`76`	`#`
`77`	`77`	# @note `:paging_state` option will be ignored.
`78`	`78`	`#`
`79`		- # @return [Cassandra::Future<Cassandra::Result, nil>] `nil` if last
`80`		`- # page`
	`79`	`+ # @return [Cassandra::Future<Cassandra::Result>] a future that resolves to a new Result if there is a new page,`
	`80`	+ # `nil` otherwise.
`81`	`81`	`#`
`82`	`82`	`# @see Cassandra::Session#execute`
`83`	`83`	`def next_page_async(options = nil)`