PreresolvingDatumWriter.cs

Go to the documentation of this file.

using System;
using System.Collections;
using System.Collections.Generic;
using System.Linq;
using Encoder = Avro.IO.Encoder;

namespace Avro.Generic
{
    public abstract class PreresolvingDatumWriter<T> : DatumWriter<T>
    {
        public Schema Schema { get; private set; }

        protected delegate void WriteItem(Object value, Encoder encoder);

        private readonly WriteItem _writer;
        private readonly ArrayAccess _arrayAccess;
        private readonly MapAccess _mapAccess;

        private readonly Dictionary<RecordSchema,WriteItem> _recordWriters = new Dictionary<RecordSchema,WriteItem>();

        public void Write(T datum, Encoder encoder)
        {
            _writer( datum, encoder );
        }

        protected PreresolvingDatumWriter(Schema schema, ArrayAccess arrayAccess, MapAccess mapAccess)
        {
            Schema = schema;
            _arrayAccess = arrayAccess;
            _mapAccess = mapAccess;
            _writer = ResolveWriter(schema);
        }

        private WriteItem ResolveWriter( Schema schema )
        {
            switch (schema.Tag)
            {
                case Schema.Type.Null:
                    return WriteNull;
                case Schema.Type.Boolean:
                    return (v, e) => Write<bool>( v, schema.Tag, e.WriteBoolean );
                case Schema.Type.Int:
                    return (v, e) => Write<int>( v, schema.Tag, e.WriteInt );
                case Schema.Type.Long:
                    return (v, e) => Write<long>( v, schema.Tag, e.WriteLong );
                case Schema.Type.Float:
                    return (v, e) => Write<float>( v, schema.Tag, e.WriteFloat );
                case Schema.Type.Double:
                    return (v, e) => Write<double>( v, schema.Tag, e.WriteDouble );
                case Schema.Type.String:
                    return (v, e) => Write<string>( v, schema.Tag, e.WriteString );
                case Schema.Type.Bytes:
                    return (v, e) => Write<byte[]>( v, schema.Tag, e.WriteBytes );
                case Schema.Type.Error:
                case Schema.Type.Record:
                    return ResolveRecord((RecordSchema) schema);
                case Schema.Type.Enumeration:
                    return ResolveEnum(schema as EnumSchema);
                case Schema.Type.Fixed:
                    return (v, e) => WriteFixed(schema as FixedSchema, v, e);
                case Schema.Type.Array:
                    return ResolveArray((ArraySchema)schema);
                case Schema.Type.Map:
                    return ResolveMap((MapSchema)schema);
                case Schema.Type.Union:
                    return ResolveUnion((UnionSchema)schema);
                default:
                    return (v, e) => error(schema, v);
            }
        }

        protected void WriteNull(object value, Encoder encoder)
        {
            if (value != null) throw TypeMismatch(value, "null", "null");
        }

        protected void Write<S>(object value, Schema.Type tag, Writer<S> writer)
        {
            if (!(value is S)) throw TypeMismatch(value, tag.ToString(), typeof(S).ToString());
            writer((S)value);
        }


        private WriteItem ResolveRecord(RecordSchema recordSchema)
        {
            WriteItem recordResolver;
            if (_recordWriters.TryGetValue(recordSchema, out recordResolver))
            {
                return recordResolver;
            }
            var writeSteps = new RecordFieldWriter[recordSchema.Fields.Count];
            recordResolver = (v, e) => WriteRecordFields(v, writeSteps, e);
            
            _recordWriters.Add(recordSchema, recordResolver);

            int index = 0;
            foreach (Field field in recordSchema)
            {
                var record = new RecordFieldWriter
                                 {
                                     WriteField = ResolveWriter(field.Schema),
                                     Field = field
                                 };
                writeSteps[index++] = record;
            }

            return recordResolver;
        }

        protected abstract void WriteRecordFields(object record, RecordFieldWriter[] writers, Encoder encoder);


        protected class RecordFieldWriter
        {
            public WriteItem WriteField { get; set; }
            public Field Field { get; set; }
        }

        protected abstract void EnsureRecordObject(RecordSchema recordSchema, object value);

        protected abstract void WriteField(object record, string fieldName, int fieldPos, WriteItem writer, Encoder encoder );

        protected abstract WriteItem ResolveEnum(EnumSchema es);

        protected WriteItem ResolveArray(ArraySchema schema)
        {
            var itemWriter = ResolveWriter(schema.ItemSchema);
            return (d,e) => WriteArray(itemWriter, d, e);
        }

        private void WriteArray(WriteItem itemWriter, object array, Encoder encoder)
        {
            _arrayAccess.EnsureArrayObject(array);
            long l = _arrayAccess.GetArrayLength(array);
            encoder.WriteArrayStart();
            encoder.SetItemCount(l);
            _arrayAccess.WriteArrayValues(array, itemWriter, encoder);
            encoder.WriteArrayEnd();
        }

        private WriteItem ResolveMap(MapSchema mapSchema)
        {
            var itemWriter = ResolveWriter(mapSchema.ValueSchema);
            return (v, e) => WriteMap(itemWriter, v, e);
        }

        protected void WriteMap(WriteItem itemWriter, object value, Encoder encoder)
        {
            _mapAccess.EnsureMapObject(value);
            encoder.WriteMapStart();
            encoder.SetItemCount(_mapAccess.GetMapSize(value));
            _mapAccess.WriteMapValues(value, itemWriter, encoder);
            encoder.WriteMapEnd();
        }


        private WriteItem ResolveUnion(UnionSchema unionSchema)
        {
            var branchSchemas = unionSchema.Schemas.ToArray();
            var branchWriters = new WriteItem[branchSchemas.Length];
            int branchIndex = 0;
            foreach (var branch in branchSchemas)
            {
                branchWriters[branchIndex++] = ResolveWriter(branch);
            }


            return (v, e) => WriteUnion(unionSchema, branchSchemas, branchWriters, v, e);
        }

        private void WriteUnion(UnionSchema unionSchema, Schema[] branchSchemas, WriteItem[] branchWriters, object value, Encoder encoder)
        {
            int index = ResolveUnion(unionSchema, branchSchemas, value);
            encoder.WriteUnionIndex(index);
            branchWriters[index](value, encoder);
        }

        protected int ResolveUnion(UnionSchema us, Schema[] branchSchemas, object obj)
        {
            for (int i = 0; i < branchSchemas.Length; i++)
            {
                if (UnionBranchMatches(branchSchemas[i], obj)) return i;
            }
            throw new AvroException("Cannot find a match for " + obj.GetType() + " in " + us);
        }

        protected abstract void WriteFixed(FixedSchema es, object value, Encoder encoder);

        protected static AvroException TypeMismatch(object obj, string schemaType, string type)
        {
            return new AvroException(type + " required to write against " + schemaType + " schema but found " + (null == obj ? "null" : obj.GetType().ToString()) );
        }

        private void error(Schema schema, Object value)
        {
            throw new AvroTypeException("Not a " + schema + ": " + value);
        }

        protected abstract bool UnionBranchMatches(Schema sc, object obj);

        protected interface EnumAccess
        {
            void WriteEnum(object value);
        }

        protected interface ArrayAccess
        {
            void EnsureArrayObject(object value);

            long GetArrayLength(object value);

            void WriteArrayValues(object array, WriteItem valueWriter, Encoder encoder);
        }

        protected interface MapAccess
        {
            void EnsureMapObject(object value);

            long GetMapSize(object value);

            void WriteMapValues(object map, WriteItem valueWriter, Encoder encoder);
        }

        protected class DictionaryMapAccess : MapAccess
        {
            public void EnsureMapObject( object value )
            {
                if( value == null || !( value is IDictionary ) ) throw TypeMismatch( value, "map", "IDictionary" );
            }

            public long GetMapSize( object value )
            {
                return ( (IDictionary) value ).Count;
            }

            public void WriteMapValues(object map, WriteItem valueWriter, Encoder encoder)
            {
                foreach (DictionaryEntry entry in ((IDictionary)map))
                {
                    encoder.StartItem();
                    encoder.WriteString(entry.Key.ToString());
                    valueWriter(entry.Value, encoder);
                }
            }
        }
    }
}