SpecificReader.cs

Go to the documentation of this file.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.IO;
using Avro;
using Avro.IO;
using Avro.Generic;

namespace Avro.Specific
{
    public class SpecificReader<T> : DatumReader<T>
    {
        private readonly SpecificDefaultReader reader;

        public Schema WriterSchema { get { return reader.WriterSchema; } }

        public Schema ReaderSchema { get { return reader.ReaderSchema; } }

        public SpecificReader(Schema writerSchema, Schema readerSchema)
        {
            reader = new SpecificDefaultReader(writerSchema, readerSchema);
        }

        public SpecificReader(SpecificDefaultReader reader)
        {
            this.reader = reader;
        }

        public T Read(T reuse, Decoder dec)
        {
            return reader.Read(reuse, dec);
        }
    }

    public class SpecificDefaultReader : DefaultReader
    {
        private static IDictionary<string, Type> TypeName = new Dictionary<string, Type>();

        public SpecificDefaultReader(Schema writerSchema, Schema readerSchema) : base(writerSchema,readerSchema) 
        {
        }

        protected override object ReadRecord(object reuse, RecordSchema writerSchema, Schema readerSchema, Decoder dec)
        {
            RecordSchema rs = (RecordSchema)readerSchema;

            if (rs.Name == null)
                return base.ReadRecord(reuse, writerSchema, readerSchema, dec);

            ISpecificRecord rec = (reuse != null ? reuse : ObjectCreator.Instance.New(rs.Fullname, Schema.Type.Record)) as ISpecificRecord;
            object obj;
            foreach (Field wf in writerSchema)
            {
                try
                {
                    Field rf;
                    if (rs.TryGetField(wf.Name, out rf))
                    {
                        obj = rec.Get(rf.Pos);
                        rec.Put(rf.Pos, Read(obj, wf.Schema, rf.Schema, dec));
                    }
                    else
                        Skip(wf.Schema, dec);
                }
                catch (Exception ex)
                {
                    throw new AvroException(ex.Message + " in field " + wf.Name);
                }
            }
            
            var defaultStream = new MemoryStream();
            var defaultEncoder = new BinaryEncoder(defaultStream);
            var defaultDecoder = new BinaryDecoder(defaultStream);
            foreach (Field rf in rs)
            {
                if (writerSchema.Contains(rf.Name)) continue;

                defaultStream.Position = 0; // reset for writing
                Resolver.EncodeDefaultValue(defaultEncoder, rf.Schema, rf.DefaultValue);
                defaultStream.Flush();
                defaultStream.Position = 0; // reset for reading

                obj = rec.Get(rf.Pos);
                rec.Put(rf.Pos, Read(obj, rf.Schema, rf.Schema, defaultDecoder));
            }
            
            return rec;
        }

        protected override object ReadFixed(object reuse, FixedSchema writerSchema, Schema readerSchema, Decoder d)
        {
            FixedSchema rs = readerSchema as FixedSchema;
            if (rs.Size != writerSchema.Size)
            {
                throw new AvroException("Size mismatch between reader and writer fixed schemas. Writer: " + writerSchema +
                    ", reader: " + readerSchema);
            }

            SpecificFixed fixedrec = (reuse != null ? reuse : ObjectCreator.Instance.New(rs.Fullname, Schema.Type.Fixed)) as SpecificFixed;
            d.ReadFixed(fixedrec.Value);
            return fixedrec;
        }

        protected override object ReadEnum(object reuse, EnumSchema writerSchema, Schema readerSchema, Decoder dec)
        {
            EnumSchema rs = readerSchema as EnumSchema;
            return rs.Ordinal(writerSchema[dec.ReadEnum()]);
        }

        protected override object ReadArray(object reuse, ArraySchema writerSchema, Schema readerSchema, Decoder dec)
        {
            ArraySchema rs = readerSchema as ArraySchema;
            System.Collections.IList array;
            if (reuse != null)
            {
                array = reuse as System.Collections.IList;
                if (array == null)
                    throw new AvroException("array object does not implement non-generic IList");

                array.Clear();
            }
            else
                array = ObjectCreator.Instance.New(getTargetType(readerSchema), Schema.Type.Array) as System.Collections.IList;
            
            int i = 0;
            for (int n = (int)dec.ReadArrayStart(); n != 0; n = (int)dec.ReadArrayNext())
            {
                for (int j = 0; j < n; j++, i++)
                    array.Add(Read(null, writerSchema.ItemSchema, rs.ItemSchema, dec));
            }
            return array;
        }

        protected override object ReadMap(object reuse, MapSchema writerSchema, Schema readerSchema, Decoder d)
        {
            MapSchema rs = readerSchema as MapSchema;
            System.Collections.IDictionary map;
            if (reuse != null)
            {
                map = reuse as System.Collections.IDictionary;
                if (map == null)
                    throw new AvroException("map object does not implement non-generic IList");

                map.Clear();
            }
            else
                map = ObjectCreator.Instance.New(getTargetType(readerSchema), Schema.Type.Map) as System.Collections.IDictionary;

            for (int n = (int)d.ReadMapStart(); n != 0; n = (int)d.ReadMapNext())
            {
                for (int j = 0; j < n; j++)
                {
                    string k = d.ReadString();
                    map[k] = Read(null, writerSchema.ValueSchema, rs.ValueSchema, d);   // always create new map item
                }
            }
            return map;
        }

        protected virtual string getTargetType(Schema schema)
        {
            bool nEnum = false;
            string type = Avro.CodeGen.getType(schema, false, ref nEnum);
            if (schema.Tag == Schema.Type.Array)
            {
                type = type.Remove(0, 6);              // remove IList<
                type = type.Remove(type.Length - 1);   // remove >
            }
            else if (schema.Tag == Schema.Type.Map)
            {
                type = type.Remove(0, 19);             // remove IDictionary<string,
                type = type.Remove(type.Length - 1);   // remove >
            }
            return type;
        }
    }
}