Merge pull request #174 from datastax/RUBY-220

RUBY-220 - Plumbing to support custom types: serializing and deserial…

Author: stamhankar999

lib/cassandra.rb

@@ -66,6 +66,7 @@ module Cassandra
     :connections_per_remote_node,
     :consistency,
     :credentials,
+    :custom_types,
     :datacenter,
     :futures_factory,
     :heartbeat_interval,
@@ -801,6 +802,7 @@ def self.validate_and_massage_options(options)
 require 'cassandra/executors'
 require 'cassandra/future'
 require 'cassandra/cluster'
+require 'cassandra/custom_data'
 require 'cassandra/driver'
 require 'cassandra/host'
 require 'cassandra/session'

lib/cassandra/cluster/connector.rb

@@ -134,7 +134,8 @@ def do_connect(host)
                                            @connection_options.compressor,
                                            @connection_options.heartbeat_interval,
                                            @connection_options.idle_timeout,
-                                           @connection_options.requests_per_connection)
+                                           @connection_options.requests_per_connection,
+                                           @connection_options.custom_type_handlers)
         end.flat_map do |connection|
           # connection is a CqlProtocolHandler
           f = request_options(connection)

lib/cassandra/cluster/options.rb

@@ -24,7 +24,7 @@ class Options
 
       attr_reader :auth_provider, :compressor, :connect_timeout, :credentials,
                   :heartbeat_interval, :idle_timeout, :port, :schema_refresh_delay,
-                  :schema_refresh_timeout, :ssl
+                  :schema_refresh_timeout, :ssl, :custom_type_handlers
       attr_boolean :protocol_negotiable, :synchronize_schema, :nodelay
 
       attr_accessor :protocol_version
@@ -45,7 +45,8 @@ def initialize(logger,
                      schema_refresh_delay,
                      schema_refresh_timeout,
                      nodelay,
-                     requests_per_connection)
+                     requests_per_connection,
+                     custom_types)
         @logger                 = logger
         @protocol_version       = protocol_version
         @credentials            = credentials
@@ -60,6 +61,10 @@ def initialize(logger,
         @schema_refresh_delay   = schema_refresh_delay
         @schema_refresh_timeout = schema_refresh_timeout
         @nodelay                = nodelay
+        @custom_type_handlers   = {}
+        custom_types.each do |type_klass|
+          @custom_type_handlers[type_klass.type] = type_klass
+        end
 
         @connections_per_local_node  = connections_per_local_node
         @connections_per_remote_node = connections_per_remote_node

lib/cassandra/cluster/schema/cql_type_parser.rb

@@ -72,6 +72,9 @@ def lookup_type(node, types)
             Cassandra::Types.tuple(*node.children.map { |t| lookup_type(t, types)})
           when 'empty'             then
             Cassandra::Types.custom('org.apache.cassandra.db.marshal.EmptyType')
+          when /\A'/ then
+            # Custom type.
+            Cassandra::Types.custom(node.name[1..-2])
           else
             types.fetch(node.name) do
               raise IncompleteTypeError, "unable to lookup type #{node.name.inspect}"

lib/cassandra/cluster/schema/fetchers.rb

@@ -836,7 +836,7 @@ def create_aggregate(aggregate_data, functions)
             initial_state  = Util.encode_object(
               Protocol::Coder.read_value_v4(
                 Protocol::CqlByteBuffer.new.append_bytes(aggregate_data['initcond']),
-                state_type))
+                state_type, nil))
 
             # The state-function takes arguments: first the stype, then the args of the aggregate.
             state_function = functions.get(aggregate_data['state_func'],

lib/cassandra/custom_data.rb

@@ -0,0 +1,51 @@
+# encoding: utf-8
+
+#--
+# Copyright 2013-2016 DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#++
+
+# Use this module to mark domain object classes as custom type implementations for custom-type
+# columns in C*. This module has no logic of its own, but indicates that the marked class has
+# certain methods.
+# @private
+module Cassandra::CustomData
+  def self.included base
+    base.send :include, InstanceMethods
+    base.extend ClassMethods
+  end
+
+  module ClassMethods
+    # @return [Cassandra::Types::Custom] the custom type that this class represents.
+    def type
+      raise NotImplementedError, "#{self.class} must implement the :type class method"
+    end
+
+    # Deserialize the given data into an instance of this domain object class.
+    # @param data [String] byte-array representation of a column value of this custom type.
+    # @return An instance of the domain object class.
+    # @raise [Cassandra::Errors::DecodingError] upon failure.
+    def deserialize(data)
+      raise NotImplementedError, "#{self.class} must implement the :deserialize class method"
+    end
+  end
+
+  module InstanceMethods
+    # Serialize this domain object into a byte array to send to C*.
+    # @return [String] byte-array representation of this domain object.
+    def serialize
+      raise NotImplementedError, "#{self.class} must implement the :serialize instance method"
+    end
+  end
+end

lib/cassandra/driver.rb

@@ -139,10 +139,12 @@ def self.let(name, &block)
         schema_refresh_delay,
         schema_refresh_timeout,
         nodelay,
-        requests_per_connection
+        requests_per_connection,
+        custom_types
       )
     end
 
+    let(:custom_types)      { [] }
     let(:port)                      { 9042 }
     let(:protocol_version)          { nil }
     let(:connect_timeout)           { 10 }

lib/cassandra/protocol.rb

@@ -28,6 +28,13 @@ module Formats
 
       BYTES_FORMAT = 'C*'.freeze
       TWO_INTS_FORMAT = 'NN'.freeze
+
+      # All of the formats above are big-endian (e.g. network-byte-order). Some payloads (custom types) may have
+      # little-endian components.
+
+      DOUBLE_FORMAT_LE = 'E'.freeze
+      INT_FORMAT_LE = 'V'.freeze
+      SHORT_FORMAT_LE = 'v'.freeze
     end
 
     module Constants

lib/cassandra/protocol/coder.rb

@@ -97,6 +97,7 @@ def write_value_v4(buffer, value, type)
         when :bigint, :counter then write_bigint(buffer, value)
         when :blob             then write_blob(buffer, value)
         when :boolean          then write_boolean(buffer, value)
+        when :custom           then write_custom(buffer, value, type)
         when :decimal          then write_decimal(buffer, value)
         when :double           then write_double(buffer, value)
         when :float            then write_float(buffer, value)
@@ -221,19 +222,19 @@ def read_type_v4(buffer)
         end
       end
 
-      def read_values_v4(buffer, column_metadata)
+      def read_values_v4(buffer, column_metadata, custom_type_handlers)
         ::Array.new(buffer.read_int) do |_i|
           row = ::Hash.new
 
           column_metadata.each do |(_, _, column, type)|
-            row[column] = read_value_v4(buffer, type)
+            row[column] = read_value_v4(buffer, type, custom_type_handlers)
           end
 
           row
         end
       end
 
-      def read_value_v4(buffer, type)
+      def read_value_v4(buffer, type, custom_type_handlers)
         case type.kind
         when :ascii            then read_ascii(buffer)
         when :bigint, :counter then read_bigint(buffer)
@@ -253,11 +254,12 @@ def read_value_v4(buffer, type)
         when :smallint         then read_smallint(buffer)
         when :time             then read_time(buffer)
         when :date             then read_date(buffer)
+        when :custom           then read_custom(buffer, type, custom_type_handlers)
         when :list
           return nil unless read_size(buffer)
 
           value_type = type.value_type
-          ::Array.new(buffer.read_signed_int) { read_value_v4(buffer, value_type) }
+          ::Array.new(buffer.read_signed_int) { read_value_v4(buffer, value_type, custom_type_handlers) }
         when :map
           return nil unless read_size(buffer)
 
@@ -266,7 +268,7 @@ def read_value_v4(buffer, type)
           value      = ::Hash.new
 
           buffer.read_signed_int.times do
-            value[read_value_v4(buffer, key_type)] = read_value_v4(buffer, value_type)
+            value[read_value_v4(buffer, key_type, custom_type_handlers)] = read_value_v4(buffer, value_type, custom_type_handlers)
           end
 
           value
@@ -277,7 +279,7 @@ def read_value_v4(buffer, type)
           value      = ::Set.new
 
           buffer.read_signed_int.times do
-            value << read_value_v4(buffer, value_type)
+            value << read_value_v4(buffer, value_type, custom_type_handlers)
           end
 
           value
@@ -295,7 +297,7 @@ def read_value_v4(buffer, type)
             values[field.name] = if length - buffer.length >= size
                                    nil
                                  else
-                                   read_value_v4(buffer, field.type)
+                                   read_value_v4(buffer, field.type, custom_type_handlers)
                                  end
           end
 
@@ -308,7 +310,7 @@ def read_value_v4(buffer, type)
 
           members.each do |member_type|
             break if buffer.empty?
-            values << read_value_v4(buffer, member_type)
+            values << read_value_v4(buffer, member_type, custom_type_handlers)
           end
 
           values.fill(nil, values.length, (members.length - values.length))
@@ -774,6 +776,15 @@ def read_boolean(buffer)
         read_size(buffer) && buffer.read(1) == Constants::TRUE_BYTE
       end
 
+      def read_custom(buffer, type, custom_type_handlers)
+        # Lookup the type-name to get the Class that can deserialize buffer data into a custom domain object.
+        unless custom_type_handlers && custom_type_handlers.key?(type)
+          raise Errors::DecodingError, %(Unsupported custom column type: #{type.name})
+        end
+        num_bytes = read_size(buffer)
+        custom_type_handlers[type].deserialize(buffer.read(num_bytes)) if num_bytes && num_bytes > 0
+      end
+
       def read_decimal(buffer)
         size = read_size(buffer)
         size && buffer.read_decimal(size)
@@ -860,6 +871,15 @@ def write_boolean(buffer, value)
         buffer.append(value ? Constants::TRUE_BYTE : Constants::FALSE_BYTE)
       end
 
+      def write_custom(buffer, value, type)
+        # Verify that the given type-name matches the value's cql type name.
+        if value.class.type != type
+          raise Errors::EncodingError, "type mismatch: value is a #{value.type} and column is a #{type}"
+        end
+
+        buffer.append_bytes(value.serialize)
+      end
+
       def write_decimal(buffer, value)
         buffer.append_bytes(CqlByteBuffer.new.append_decimal(value))
       end

lib/cassandra/protocol/cql_byte_buffer.rb

@@ -92,6 +92,13 @@ def read_double
               "Not enough bytes available to decode a double: #{e.message}", e.backtrace
       end
 
+      def read_double_le
+        read(8).unpack(Formats::DOUBLE_FORMAT_LE).first
+      rescue RangeError => e
+        raise Errors::DecodingError,
+              "Not enough bytes available to decode a double: #{e.message}", e.backtrace
+      end
+
       def read_float
         read(4).unpack(Formats::FLOAT_FORMAT).first
       rescue RangeError => e
@@ -108,13 +115,27 @@ def read_signed_int
               "Not enough bytes available to decode an int: #{e.message}", e.backtrace
       end
 
+      def read_unsigned_int_le
+        read(4).unpack(Formats::INT_FORMAT_LE).first
+      rescue RangeError => e
+        raise Errors::DecodingError,
+              "Not enough bytes available to decode an int: #{e.message}", e.backtrace
+      end
+
       def read_unsigned_short
         read_short
       rescue RangeError => e
         raise Errors::DecodingError,
               "Not enough bytes available to decode a short: #{e.message}", e.backtrace
       end
 
+      def read_unsigned_short_le
+        read(2).unpack(Formats::SHORT_FORMAT_LE).first
+      rescue RangeError => e
+        raise Errors::DecodingError,
+              "Not enough bytes available to decode a short: #{e.message}", e.backtrace
+      end
+
       def read_string
         length = read_unsigned_short
         string = read(length)