GenericReader.cs

Go to the documentation of this file.

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

namespace Avro.Generic
{
    public delegate T Reader<T>();

    public sealed class GenericReader<T> : DatumReader<T>
    {
        private readonly DefaultReader reader;

        public GenericReader(Schema writerSchema, Schema readerSchema)
            : this(new DefaultReader(writerSchema, readerSchema))
        {
        }

        public GenericReader(DefaultReader reader)
        {
            this.reader = reader;
        }

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

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

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

    public class DefaultReader
    {
        public Schema ReaderSchema { get; private set; }
        public Schema WriterSchema { get; private set; }


        public DefaultReader(Schema writerSchema, Schema readerSchema)
        {
            this.ReaderSchema = readerSchema;
            this.WriterSchema = writerSchema;
        }

        public T Read<T>(T reuse, Decoder decoder)
        {
            if (!ReaderSchema.CanRead(WriterSchema))
                throw new AvroException("Schema mismatch. Reader: " + ReaderSchema + ", writer: " + WriterSchema);

            return (T)Read(reuse, WriterSchema, ReaderSchema, decoder);
        }

        public object Read(object reuse, Schema writerSchema, Schema readerSchema, Decoder d)
        {
            if (readerSchema.Tag == Schema.Type.Union && writerSchema.Tag != Schema.Type.Union)
            {
                readerSchema = findBranch(readerSchema as UnionSchema, writerSchema);
            }
            /*
            if (!readerSchema.CanRead(writerSchema))
            {
                throw new AvroException("Schema mismatch. Reader: " + readerSchema + ", writer: " + writerSchema);
            }
            */
            switch (writerSchema.Tag)
            {
                case Schema.Type.Null:
                    return ReadNull(readerSchema, d);
                case Schema.Type.Boolean:
                    return Read<bool>(writerSchema.Tag, readerSchema, d.ReadBoolean);
                case Schema.Type.Int:
                    {
                        int i = Read<int>(writerSchema.Tag, readerSchema, d.ReadInt);
                        switch (readerSchema.Tag)
                        {
                            case Schema.Type.Long:
                                return (long)i;
                            case Schema.Type.Float:
                                return (float)i;
                            case Schema.Type.Double:
                                return (double)i;
                            default:
                                return i;
                        }
                    }
                case Schema.Type.Long:
                    {
                        long l = Read<long>(writerSchema.Tag, readerSchema, d.ReadLong);
                        switch (readerSchema.Tag)
                        {
                            case Schema.Type.Float:
                                return (float)l;
                            case Schema.Type.Double:
                                return (double)l;
                            default:
                                return l;
                        }
                    }
                case Schema.Type.Float:
                    {
                        float f = Read<float>(writerSchema.Tag, readerSchema, d.ReadFloat);
                        switch (readerSchema.Tag)
                        {
                            case Schema.Type.Double:
                                return (double)f;
                            default:
                                return f;
                        }
                    }
                case Schema.Type.Double:
                    return Read<double>(writerSchema.Tag, readerSchema, d.ReadDouble);
                case Schema.Type.String:
                    return Read<string>(writerSchema.Tag, readerSchema, d.ReadString);
                case Schema.Type.Bytes:
                    return Read<byte[]>(writerSchema.Tag, readerSchema, d.ReadBytes);
                case Schema.Type.Error:
                case Schema.Type.Record:
                    return ReadRecord(reuse, (RecordSchema)writerSchema, readerSchema, d);
                case Schema.Type.Enumeration:
                    return ReadEnum(reuse, (EnumSchema)writerSchema, readerSchema, d);
                case Schema.Type.Fixed:
                    return ReadFixed(reuse, (FixedSchema)writerSchema, readerSchema, d);
                case Schema.Type.Array:
                    return ReadArray(reuse, (ArraySchema)writerSchema, readerSchema, d);
                case Schema.Type.Map:
                    return ReadMap(reuse, (MapSchema)writerSchema, readerSchema, d);
                case Schema.Type.Union:
                    return ReadUnion(reuse, (UnionSchema)writerSchema, readerSchema, d);
                default:
                    throw new AvroException("Unknown schema type: " + writerSchema);
            }
        }

        protected virtual object ReadNull(Schema readerSchema, Decoder d)
        {
            d.ReadNull();
            return null;
        }

        protected S Read<S>(Schema.Type tag, Schema readerSchema, Reader<S> reader)
        {
            return reader();
        }

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

            object rec = CreateRecord(reuse, rs);
            foreach (Field wf in writerSchema)
            {
                try
                {
                    Field rf;
                    if (rs.TryGetFieldAlias(wf.Name, out rf))
                    {
                        object obj = null;
                        TryGetField(rec, wf.Name, rf.Pos, out obj);
                        AddField(rec, wf.Name, 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

                object obj = null;
                TryGetField(rec, rf.Name, rf.Pos, out obj);
                AddField(rec, rf.Name, rf.Pos, Read(obj, rf.Schema, rf.Schema, defaultDecoder));
            }

            return rec;
        }

        protected virtual object CreateRecord(object reuse, RecordSchema readerSchema)
        {
            GenericRecord ru = (reuse == null || !(reuse is GenericRecord) || !(reuse as GenericRecord).Schema.Equals(readerSchema)) ?
                new GenericRecord(readerSchema) :
                reuse as GenericRecord;
            return ru;
        }

        protected virtual bool TryGetField(object record, string fieldName, int fieldPos, out object value)
        {
            return (record as GenericRecord).TryGetValue(fieldName, out value);
        }

        protected virtual void AddField(object record, string fieldName, int fieldPos, object fieldValue)
        {
            (record as GenericRecord).Add(fieldName, fieldValue);
        }

        protected virtual object ReadEnum(object reuse, EnumSchema writerSchema, Schema readerSchema, Decoder d)
        {
            EnumSchema es = readerSchema as EnumSchema;
            return CreateEnum(reuse, readerSchema as EnumSchema, writerSchema[d.ReadEnum()]);
        }

        protected virtual object CreateEnum(object reuse, EnumSchema es, string symbol)
        {
            if (reuse is GenericEnum)
            {
                GenericEnum ge = reuse as GenericEnum;
                if (ge.Schema.Equals(es))
                {
                    ge.Value = symbol;
                    return ge;
                }
            }
            return new GenericEnum(es, symbol);
        }

        protected virtual object ReadArray(object reuse, ArraySchema writerSchema, Schema readerSchema, Decoder d)
        {

            ArraySchema rs = (ArraySchema)readerSchema;
            object result = CreateArray(reuse, rs);
            int i = 0;
            for (int n = (int)d.ReadArrayStart(); n != 0; n = (int)d.ReadArrayNext())
            {
                if (GetArraySize(result) < (i + n)) ResizeArray(ref result, i + n);
                for (int j = 0; j < n; j++, i++)
                {
                    SetArrayElement(result, i, Read(GetArrayElement(result, i), writerSchema.ItemSchema, rs.ItemSchema, d));
                }
            }
            if (GetArraySize(result) != i) ResizeArray(ref result, i);
            return result;
        }

        protected virtual object CreateArray(object reuse, ArraySchema rs)
        {
            return (reuse != null && reuse is object[]) ? (object[])reuse : new object[0];
        }

        protected virtual int GetArraySize(object array)
        {
            return (array as object[]).Length;
        }

        protected virtual void ResizeArray(ref object array, int n)
        {
            object[] o = array as object[];
            Array.Resize(ref o, n);
            array = o;
        }

        protected virtual void SetArrayElement(object array, int index, object value)
        {
            object[] a = array as object[];
            a[index] = value;
        }

        protected virtual object GetArrayElement(object array, int index)
        {
            return (array as object[])[index];
        }

        protected virtual object ReadMap(object reuse, MapSchema writerSchema, Schema readerSchema, Decoder d)
        {
            MapSchema rs = (MapSchema)readerSchema;
            object result = CreateMap(reuse, rs);
            for (int n = (int)d.ReadMapStart(); n != 0; n = (int)d.ReadMapNext())
            {
                for (int j = 0; j < n; j++)
                {
                    string k = d.ReadString();
                    AddMapEntry(result, k, Read(null, writerSchema.ValueSchema, rs.ValueSchema, d));
                }
            }
            return result;
        }

        protected virtual object CreateMap(object reuse, MapSchema ms)
        {
            if (reuse != null && reuse is IDictionary<string, object>)
            {
                IDictionary<string, object> result = reuse as IDictionary<string, object>;
                result.Clear();
                return result;
            }
            return new Dictionary<string, object>();
        }

        protected virtual void AddMapEntry(object map, string key, object value)
        {
            (map as IDictionary<string, object>).Add(key, value);
        }

        protected virtual object ReadUnion(object reuse, UnionSchema writerSchema, Schema readerSchema, Decoder d)
        {
            int index = d.ReadUnionIndex();
            Schema ws = writerSchema[index];

            if (readerSchema is UnionSchema)
                readerSchema = findBranch(readerSchema as UnionSchema, ws);
            else
                if (!readerSchema.CanRead(ws))
                    throw new AvroException("Schema mismatch. Reader: " + ReaderSchema + ", writer: " + WriterSchema);

            return Read(reuse, ws, readerSchema, d);
        }

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

            object ru = CreateFixed(reuse, rs);
            byte[] bb = GetFixedBuffer(ru);
            d.ReadFixed(bb);
            return ru;
        }

        protected virtual object CreateFixed(object reuse, FixedSchema rs)
        {
            return (reuse != null && reuse is GenericFixed && (reuse as GenericFixed).Schema.Equals(rs)) ?
                (GenericFixed)reuse : new GenericFixed(rs);
        }

        protected virtual byte[] GetFixedBuffer(object f)
        {
            return (f as GenericFixed).Value;
        }

        protected virtual void Skip(Schema writerSchema, Decoder d)
        {
            switch (writerSchema.Tag)
            {
                case Schema.Type.Null:
                    d.SkipNull();
                    break;
                case Schema.Type.Boolean:
                    d.SkipBoolean();
                    break;
                case Schema.Type.Int:
                    d.SkipInt();
                    break;
                case Schema.Type.Long:
                    d.SkipLong();
                    break;
                case Schema.Type.Float:
                    d.SkipFloat();
                    break;
                case Schema.Type.Double:
                    d.SkipDouble();
                    break;
                case Schema.Type.String:
                    d.SkipString();
                    break;
                case Schema.Type.Bytes:
                    d.SkipBytes();
                    break;
                case Schema.Type.Record:
                    foreach (Field f in writerSchema as RecordSchema) Skip(f.Schema, d);
                    break;
                case Schema.Type.Enumeration:
                    d.SkipEnum();
                    break;
                case Schema.Type.Fixed:
                    d.SkipFixed((writerSchema as FixedSchema).Size);
                    break;
                case Schema.Type.Array:
                    {
                        Schema s = (writerSchema as ArraySchema).ItemSchema;
                        for (long n = d.ReadArrayStart(); n != 0; n = d.ReadArrayNext())
                        {
                            for (long i = 0; i < n; i++) Skip(s, d);
                        }
                    }
                    break;
                case Schema.Type.Map:
                    {
                        Schema s = (writerSchema as MapSchema).ValueSchema;
                        for (long n = d.ReadMapStart(); n != 0; n = d.ReadMapNext())
                        {
                            for (long i = 0; i < n; i++) { d.SkipString(); Skip(s, d); }
                        }
                    }
                    break;
                case Schema.Type.Union:
                    Skip((writerSchema as UnionSchema)[d.ReadUnionIndex()], d);
                    break;
                default:
                    throw new AvroException("Unknown schema type: " + writerSchema);
            }
        }

        protected static Schema findBranch(UnionSchema us, Schema s)
        {
            int index = us.MatchingBranch(s);
            if (index >= 0) return us[index];
            throw new AvroException("No matching schema for " + s + " in " + us);
        }

    }
}