Ensure date and time coercions are working correctly

lib/field_struct/avro_schema/avro_schema.rb

@@ -6,14 +6,26 @@ module AvroSchema
       float: :float,
       big_integer: :float,
       decimal: :float,
-      currency: :int,
       integer: :int,
       binary: :bytes,
       string: :string,
-      date: :int,
-      datetime: :long,
+      date: {
+        nil    => :int,
+        'date' => :int
+      },
+      datetime: {
+        nil                 => :int,
+        :date               => :int,
+        :'timestamp-millis' => :long,
+        :'timestamp-micros' => :long
+      },
       immutable_string: :string,
-      time: :long,
+      time: {
+        nil                 => :int,
+        :date               => :int,
+        :'timestamp-millis' => :long,
+        :'timestamp-micros' => :long
+      },
       boolean: :boolean,
       array: :array
     }.with_indifferent_access.freeze

lib/field_struct/avro_schema/builders/avro_builder.rb

@@ -198,7 +198,12 @@ def add_array_field_type_for(attr, hsh)
         if (type_ary = logical_type_tuple(attr, type))
           hsh[:items] = type_ary
         elsif (type_key = ACTIVE_MODEL_TYPES[type])
-          hsh[:items] = type_key.to_s
+          result = if type_key.is_a?(Hash)
+                     type_key[nil]
+                   else
+                     type_key
+                   end
+          hsh[:items] = result.to_s
         elsif type.field_struct?
           hsh[:items] = type.schema_record_name
         end
@@ -209,16 +214,31 @@ def add_single_field_type_for(attr, hsh)
         if (type_ary = logical_type_tuple(attr, type))
           hsh[:type] = type_ary.first.to_s
         elsif (type_key = ACTIVE_MODEL_TYPES[type])
-          hsh[:type] = type_key.to_s
+          result = if type_key.is_a?(Hash)
+                     type_key[nil]
+                   else
+                     type_key
+                   end
+          hsh[:type] = result.to_s
         elsif type.field_struct?
           hsh[:type] = type.schema_record_name
         end
       end
 
       def logical_type_tuple(attr, type)
         if !attr.avro.nil? && attr.avro.key?(:logical_type)
-          resolved_type = ACTIVE_MODEL_TYPES.key?(type) ? ACTIVE_MODEL_TYPES[type] : type
-          [resolved_type, attr.avro[:logical_type]]
+          logical_type = attr.avro[:logical_type]
+          resolved = resolve_type(type, logical_type)
+          [resolved, logical_type]
+        end
+      end
+
+      def resolve_type(type, logical_type)
+        resolved_type = ACTIVE_MODEL_TYPES.key?(type) ? ACTIVE_MODEL_TYPES[type] : type
+        if resolved_type.is_a?(Hash)
+          resolved_type[logical_type.to_s]
+        else
+          resolved_type
         end
       end
 

lib/field_struct/avro_schema/converters.rb

@@ -5,7 +5,6 @@
 require_relative 'value_converters/date_converter'
 require_relative 'value_converters/time_converter'
 require_relative 'value_converters/date_time_converter'
-require_relative 'value_converters/currency_converter'
 
 require_relative 'converters/to_avro'
 require_relative 'converters/from_avro'

lib/field_struct/avro_schema/converters/from_avro.rb

@@ -34,7 +34,7 @@ def convert_value(attr, value)
           when Array
             convert_array_value attr, value
           else
-            convert_simple_value attr.of || attr.type, value
+            convert_simple_value attr, value
           end
         end
 
@@ -47,16 +47,16 @@ def convert_array_value(attr, value)
             if attr.of.field_struct?
               attr.of.from_avro_hash(x).to_hash
             else
-              convert_simple_value attr.of, x
+              convert_simple_value attr, x
             end
           end
         end
 
-        def convert_simple_value(type, value)
-          converter = ValueConverters::Registry.find type
+        def convert_simple_value(attr, value)
+          converter = ValueConverters::Registry.find attr.of || attr.type
           return value unless converter
 
-          converter.from_avro value
+          converter.from_avro value, attr.avro&.fetch(:logical_type)
         end
       end
 

lib/field_struct/avro_schema/converters/to_avro.rb

@@ -37,7 +37,7 @@ def convert_simple_value(attr, value)
           converter = ValueConverters::Registry.find attr.of || attr.type
           return value unless converter
 
-          converter.to_avro value
+          converter.to_avro value, attr.avro&.fetch(:logical_type)
         end
       end
 

lib/field_struct/avro_schema/value_converters/base.rb

@@ -10,19 +10,20 @@ def handles(*values)
             @values || []
           end
 
-          def to_avro(value)
-            new(value).to_avro
+          def to_avro(value, logical_type)
+            new(value, logical_type).to_avro
           end
 
-          def from_avro(value)
-            new(value).from_avro
+          def from_avro(value, logical_type)
+            new(value, logical_type).from_avro
           end
         end
 
         attr_reader :value
 
-        def initialize(value)
-          @value = value
+        def initialize(value, logical_type)
+          @value        = value
+          @logical_type = logical_type
         end
 
         def to_avro

lib/field_struct/avro_schema/value_converters/registry.rb

@@ -12,12 +12,12 @@ def all
 
         def register(*klasses)
           klasses.each do |klass|
-            klass.handles.each { |type| all[type.to_s] = klass }
+            klass.handles.each { |type| all[type] = klass }
           end
         end
 
         def find(type)
-          all[type.to_s]
+          all[type]
         end
       end
     end

lib/field_struct/avro_schema/value_converters/time_converter.rb

@@ -7,14 +7,30 @@ module ValueConverters
       class TimeConverter < Base
         handles :time
 
+        EPOCH = Date.new(1970, 1, 1)
+
         def to_avro
-          value.utc.strftime('%s%3N').to_i
+          case @logical_type
+          when nil, 'date'
+            (value.utc.to_date - EPOCH).to_i
+          when 'timestamp-millis'
+            value.utc.strftime('%s%3N').to_i
+          when 'timestamp-micros'
+            value.utc.strftime('%s%6N').to_i
+          end
         end
 
         def from_avro
           return value if value.is_a?(Time)
 
-          Time.use_zone('UTC') { Time.zone.at value / 1_000.0 }
+          case @logical_type
+          when nil, 'date'
+            Time.use_zone('UTC') { Time.zone.at(value * 86400) }
+          when 'timestamp-millis'
+            Time.use_zone('UTC') { Time.zone.at value / 1_000.0 }
+          when 'timestamp-micros'
+            Time.use_zone('UTC') { Time.zone.at value / 1_000_000.0 }
+          end
         end
       end
 

spec/kafka_spec.rb

@@ -5,6 +5,7 @@
 RSpec.describe FieldStruct::AvroSchema::Kafka do
   let(:known_events) do
     {
+      'Coercions::Examples::TestStruct' => Coercions::Examples::TestStruct,
       'CustomNamespace::CustomRecordName' => CustomNamespace::CustomRecordName,
       'ExampleApp::Examples::Friend' => ExampleApp::Examples::Friend,
       'Examples::Base' => Examples::Base,