Class: Tros::IO::DatumWriter

Inherits:
Object
  • Object
show all
Defined in:
lib/tros/io.rb

Overview

DatumWriter for generic ruby objects

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(writers_schema = nil) ⇒ DatumWriter

Returns a new instance of DatumWriter.



528
529
530
 
# File 'lib/tros/io.rb', line 528

def initialize(writers_schema=nil)
  @writers_schema = writers_schema
end

Instance Attribute Details

#writers_schema ⇒ Object

Returns the value of attribute writers_schema.



527
528
529
 
# File 'lib/tros/io.rb', line 527

def writers_schema
  @writers_schema
end

Instance Method Details

#write(datum, encoder) ⇒ Object 



532
533
534
 
# File 'lib/tros/io.rb', line 532

def write(datum, encoder)
  write_data(writers_schema, datum, encoder)
end

#write_array(writers_schema, datum, encoder) ⇒ Object 



571
572
573
574
575
576
577
578
579
 
# File 'lib/tros/io.rb', line 571

def write_array(writers_schema, datum, encoder)
  if datum.size > 0
    encoder.write_long(datum.size)
    datum.each do |item|
      write_data(writers_schema.items, item, encoder)
    end
  end
  encoder.write_long(0)
end

#write_data(writers_schema, datum, encoder) ⇒ Object 



536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
 
# File 'lib/tros/io.rb', line 536

def write_data(writers_schema, datum, encoder)
  unless Schema.validate(writers_schema, datum)
    raise AvroTypeError.new(writers_schema, datum)
  end

  # function dispatch to write datum
  case writers_schema.type_sym
  when :null;    encoder.write_null(datum)
  when :boolean; encoder.write_boolean(datum)
  when :string;  encoder.write_string(datum)
  when :int;     encoder.write_int(datum)
  when :long;    encoder.write_long(datum)
  when :float;   encoder.write_float(datum)
  when :double;  encoder.write_double(datum)
  when :bytes;   encoder.write_bytes(datum)
  when :fixed;   write_fixed(writers_schema, datum, encoder)
  when :enum;    write_enum(writers_schema, datum, encoder)
  when :array;   write_array(writers_schema, datum, encoder)
  when :map;     write_map(writers_schema, datum, encoder)
  when :union;   write_union(writers_schema, datum, encoder)
  when :record, :error, :request;  write_record(writers_schema, datum, encoder)
  else
    raise AvroError.new("Unknown type: #{writers_schema.type}")
  end
end

#write_enum(writers_schema, datum, encoder) ⇒ Object 



566
567
568
569
 
# File 'lib/tros/io.rb', line 566

def write_enum(writers_schema, datum, encoder)
  index_of_datum = writers_schema.symbols.index(datum)
  encoder.write_int(index_of_datum)
end

#write_fixed(writers_schema, datum, encoder) ⇒ Object 



562
563
564
 
# File 'lib/tros/io.rb', line 562

def write_fixed(writers_schema, datum, encoder)
  encoder.write(datum)
end

#write_map(writers_schema, datum, encoder) ⇒ Object 



581
582
583
584
585
586
587
588
589
590
 
# File 'lib/tros/io.rb', line 581

def write_map(writers_schema, datum, encoder)
  if datum.size > 0
    encoder.write_long(datum.size)
    datum.each do |k,v|
      encoder.write_string(k)
      write_data(writers_schema.values, v, encoder)
    end
  end
  encoder.write_long(0)
end

#write_record(writers_schema, datum, encoder) ⇒ Object 



601
602
603
604
605
 
# File 'lib/tros/io.rb', line 601

def write_record(writers_schema, datum, encoder)
  writers_schema.fields.each do |field|
    write_data(field.type, datum[field.name], encoder)
  end
end

#write_union(writers_schema, datum, encoder) ⇒ Object

Raises:



592
593
594
595
596
597
598
599
 
# File 'lib/tros/io.rb', line 592

def write_union(writers_schema, datum, encoder)
  index = writers_schema.schemas.find_index { |s| Schema.validate_strictly(s, datum) }
  index ||= writers_schema.schemas.find_index { |s| Schema.validate(s, datum) }
  raise AvroTypeError.new(writers_schema, datum) if index.nil?

  encoder.write_long(index)
  write_data(writers_schema.schemas[index], datum, encoder)
end