GenericWriter.cs

Go to the documentation of this file.

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

namespace Avro.Generic
{
    public delegate void Writer<T>(T t);
    public class GenericWriter<T> : DatumWriter<T>
    {
        private readonly DefaultWriter writer;
        public GenericWriter(Schema schema) : this(new DefaultWriter(schema))
        {

        }

        public Schema Schema { get { return writer.Schema; } }

        public GenericWriter(DefaultWriter writer)
        {
            this.writer = writer;
        }

        public void Write(T value, Encoder encoder)
        {
            writer.Write(value, encoder);
        }
    }

    public class DefaultWriter
    {
        public Schema Schema { get; private set; }

        public DefaultWriter(Schema schema)
        {
            this.Schema = schema;
        }

        public void Write<T>(T value, Encoder encoder)
        {
            Write(Schema, value, encoder);
        }
        public virtual void Write(Schema schema, object value, Encoder encoder)
        {
            switch (schema.Tag)
            {
                case Schema.Type.Null:
                    WriteNull(value, encoder);
                    break;
                case Schema.Type.Boolean:
                    Write<bool>(value, schema.Tag, encoder.WriteBoolean);
                    break;
                case Schema.Type.Int:
                    Write<int>(value, schema.Tag, encoder.WriteInt);
                    break;
                case Schema.Type.Long:
                    Write<long>(value, schema.Tag, encoder.WriteLong);
                    break;
                case Schema.Type.Float:
                    Write<float>(value, schema.Tag, encoder.WriteFloat);
                    break;
                case Schema.Type.Double:
                    Write<double>(value, schema.Tag, encoder.WriteDouble);
                    break;
                case Schema.Type.String:
                    Write<string>(value, schema.Tag, encoder.WriteString);
                    break;
                case Schema.Type.Bytes:
                    Write<byte[]>(value, schema.Tag, encoder.WriteBytes);
                    break;
                case Schema.Type.Record:
                case Schema.Type.Error:
                    WriteRecord(schema as RecordSchema, value, encoder);
                    break;
                case Schema.Type.Enumeration:
                    WriteEnum(schema as EnumSchema, value, encoder);
                    break;
                case Schema.Type.Fixed:
                    WriteFixed(schema as FixedSchema, value, encoder);
                    break;
                case Schema.Type.Array:
                    WriteArray(schema as ArraySchema, value, encoder);
                    break;
                case Schema.Type.Map:
                    WriteMap(schema as MapSchema, value, encoder);
                    break;
                case Schema.Type.Union:
                    WriteUnion(schema as UnionSchema, value, encoder);
                    break;
                default:
                    error(schema, value);
                    break;
            }
        }

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

        protected virtual 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);
        }


        protected virtual void WriteRecord(RecordSchema schema, object value, Encoder encoder)
        {
            EnsureRecordObject(schema, value);
            foreach (Field field in schema)
            {
                try
                {
                    object obj = GetField(value, field.Name, field.Pos);
                    Write(field.Schema, obj, encoder);
                }
                catch (Exception ex)
                {
                    throw new AvroException(ex.Message + " in field " + field.Name);
                }
            }
        }

        protected virtual void EnsureRecordObject(RecordSchema s, object value)
        {
            if (value == null || !(value is GenericRecord) || !((value as GenericRecord).Schema.Equals(s)))
            {
                throw TypeMismatch(value, "record", "GenericRecord");
            }
        }

        protected virtual object GetField(object value, string fieldName, int fieldPos)
        {
            GenericRecord d = value as GenericRecord;
            return d[fieldName];
        }

        protected virtual void WriteEnum(EnumSchema es, object value, Encoder encoder)
        {
            if (value == null || !(value is GenericEnum) || !((value as GenericEnum).Schema.Equals(es)))
                throw TypeMismatch(value, "enum", "GenericEnum");
            encoder.WriteEnum(es.Ordinal((value as GenericEnum).Value));
        }

        protected virtual void WriteArray(ArraySchema schema, object value, Encoder encoder)
        {
            EnsureArrayObject(value);
            long l = GetArrayLength(value);
            encoder.WriteArrayStart();
            encoder.SetItemCount(l);
            for (long i = 0; i < l; i++)
            {
                encoder.StartItem();
                Write(schema.ItemSchema, GetArrayElement(value, i), encoder);
            }
            encoder.WriteArrayEnd();
        }

        protected virtual void EnsureArrayObject(object value)
        {
            if (value == null || !(value is Array)) throw TypeMismatch(value, "array", "Array");
        }

        protected virtual long GetArrayLength(object value)
        {
            return (value as Array).Length;
        }

        protected virtual object GetArrayElement(object value, long index)
        {
            return (value as Array).GetValue(index);
        }

        protected virtual void WriteMap(MapSchema schema, object value, Encoder encoder)
        {
            EnsureMapObject(value);
            IDictionary<string, object> vv = (IDictionary<string, object>)value;
            encoder.WriteMapStart();
            encoder.SetItemCount(GetMapSize(value));
            foreach (KeyValuePair<string, object> obj in GetMapValues(vv))
            {
                encoder.StartItem();
                encoder.WriteString(obj.Key);
                Write(schema.ValueSchema, obj.Value, encoder);
            }
            encoder.WriteMapEnd();
        }

        protected virtual void EnsureMapObject(object value)
        {
            if (value == null || !(value is IDictionary<string, object>)) throw TypeMismatch(value, "map", "IDictionary<string, object>");
        }

        protected virtual long GetMapSize(object value)
        {
            return (value as IDictionary<string, object>).Count;
        }

        protected virtual IEnumerable<KeyValuePair<string, object>> GetMapValues(object value)
        {
            return value as IDictionary<string, object>;
        }

        protected virtual void WriteUnion(UnionSchema us, object value, Encoder encoder)
        {
            int index = ResolveUnion(us, value);
            encoder.WriteUnionIndex(index);
            Write(us[index], value, encoder);
        }

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

        protected virtual void WriteFixed(FixedSchema es, object value, Encoder encoder)
        {
            if (value == null || !(value is GenericFixed) || !(value as GenericFixed).Schema.Equals(es))
            {
                throw TypeMismatch(value, "fixed", "GenericFixed");
            }
            GenericFixed ba = (GenericFixed)value;
            encoder.WriteFixed(ba.Value);
        }

        protected 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);
        }

        /*
         * FIXME: This method of determining the Union branch has problems. If the data is IDictionary<string, object>
         * if there are two branches one with record schema and the other with map, it choose the first one. Similarly if
         * the data is byte[] and there are fixed and bytes schemas as branches, it choose the first one that matches.
         * Also it does not recognize the arrays of primitive types.
         */
        protected virtual bool Matches(Schema sc, object obj)
        {
            if (obj == null && sc.Tag != Avro.Schema.Type.Null) return false;
            switch (sc.Tag)
            {
                case Schema.Type.Null:
                    return obj == null;
                case Schema.Type.Boolean:
                    return obj is bool;
                case Schema.Type.Int:
                    return obj is int;
                case Schema.Type.Long:
                    return obj is long;
                case Schema.Type.Float:
                    return obj is float;
                case Schema.Type.Double:
                    return obj is double;
                case Schema.Type.Bytes:
                    return obj is byte[];
                case Schema.Type.String:
                    return obj is string;
                case Schema.Type.Record:
                    //return obj is GenericRecord && (obj as GenericRecord).Schema.Equals(s);
                    return obj is GenericRecord && (obj as GenericRecord).Schema.SchemaName.Equals((sc as RecordSchema).SchemaName);
                case Schema.Type.Enumeration:
                    //return obj is GenericEnum && (obj as GenericEnum).Schema.Equals(s);
                    return obj is GenericEnum && (obj as GenericEnum).Schema.SchemaName.Equals((sc as EnumSchema).SchemaName);
                case Schema.Type.Array:
                    return obj is Array && !(obj is byte[]);
                case Schema.Type.Map:
                    return obj is IDictionary<string, object>;
                case Schema.Type.Union:
                    return false;   // Union directly within another union not allowed!
                case Schema.Type.Fixed:
                    //return obj is GenericFixed && (obj as GenericFixed).Schema.Equals(s);
                    return obj is GenericFixed && (obj as GenericFixed).Schema.SchemaName.Equals((sc as FixedSchema).SchemaName);
                default:
                    throw new AvroException("Unknown schema type: " + sc.Tag);
            }
        }
    }
}