RecordSchema.cs

Go to the documentation of this file.

using System;
using System.Collections.Generic;
using System.Text;
using Newtonsoft.Json.Linq;
using Newtonsoft.Json;

namespace Avro
{
    internal delegate T Function<T>();

    public class RecordSchema : NamedSchema
    {
        public List<Field> Fields { get; private set; }

        public int Count { get { return Fields.Count; } }

        private readonly IDictionary<string, Field> fieldLookup;

        private readonly IDictionary<string, Field> fieldAliasLookup;
        private bool request;

        internal static RecordSchema NewInstance(Type type, JToken jtok, PropertyMap props, SchemaNames names, string encspace)
        {
            bool request = false;
            JToken jfields = jtok["fields"];    // normal record
            if (null == jfields)
            {
                jfields = jtok["request"];      // anonymous record from messages
                if (null != jfields) request = true;
            }
            if (null == jfields)
                throw new SchemaParseException("'fields' cannot be null for record");
            if (jfields.Type != JTokenType.Array)
                throw new SchemaParseException("'fields' not an array for record");

            var name = GetName(jtok, encspace);
            var aliases = NamedSchema.GetAliases(jtok, name.Space, name.EncSpace);
            var fields = new List<Field>();
            var fieldMap = new Dictionary<string, Field>();
            var fieldAliasMap = new Dictionary<string, Field>();
            var result = new RecordSchema(type, name, aliases, props, fields, request, fieldMap, fieldAliasMap, names);

            int fieldPos = 0;
            foreach (JObject jfield in jfields)
            {
                string fieldName = JsonHelper.GetRequiredString(jfield, "name");
                Field field = createField(jfield, fieldPos++, names, name.Namespace);  // add record namespace for field look up
                fields.Add(field);
                addToFieldMap(fieldMap, fieldName, field);
                addToFieldMap(fieldAliasMap, fieldName, field);

                if (null != field.aliases)    // add aliases to field lookup map so reader function will find it when writer field name appears only as an alias on the reader field
                    foreach (string alias in field.aliases)
                        addToFieldMap(fieldAliasMap, alias, field);
            }
            return result;
        }

        private RecordSchema(Type type, SchemaName name, IList<SchemaName> aliases,  PropertyMap props, 
                                List<Field> fields, bool request, IDictionary<string, Field> fieldMap, 
                                IDictionary<string, Field> fieldAliasMap, SchemaNames names)
                                : base(type, name, aliases, props, names)
        {
            if (!request && null == name.Name) throw new SchemaParseException("name cannot be null for record schema.");
            this.Fields = fields;
            this.request = request;
            this.fieldLookup = fieldMap;
            this.fieldAliasLookup = fieldAliasMap;
        }

        private static Field createField(JToken jfield, int pos, SchemaNames names, string encspace)
        {
            var name = JsonHelper.GetRequiredString(jfield, "name");
            var doc = JsonHelper.GetOptionalString(jfield, "doc");

            var jorder = JsonHelper.GetOptionalString(jfield, "order");
            Field.SortOrder sortorder = Field.SortOrder.ignore;
            if (null != jorder)
                sortorder = (Field.SortOrder) Enum.Parse(typeof(Field.SortOrder), jorder);

            var aliases = Field.GetAliases(jfield);
            var props = Schema.GetProperties(jfield);
            var defaultValue = jfield["default"];

            JToken jtype = jfield["type"];
            if (null == jtype) 
                throw new SchemaParseException("'type' was not found for field: " + name);
            var schema = Schema.ParseJson(jtype, names, encspace);

            return new Field(schema, name, aliases, pos, doc, defaultValue, sortorder, props);
        }

        private static void addToFieldMap(Dictionary<string, Field> map, string name, Field field)
        {
            if (map.ContainsKey(name))
                throw new SchemaParseException("field or alias " + name + " is a duplicate name");
            map.Add(name, field);
        }

        public Field this[string name]
        {
            get
            {
                if (string.IsNullOrEmpty(name)) throw new ArgumentNullException("name");
                Field field;
                return (fieldLookup.TryGetValue(name, out field)) ? field : null;
            }
        }

        public bool Contains(string fieldName)
        {
            return fieldLookup.ContainsKey(fieldName);
        }

        public bool TryGetField(string fieldName, out Field field)
        {
            return fieldLookup.TryGetValue(fieldName, out field);
        }
        public bool TryGetFieldAlias(string fieldName, out Field field)
        {
            return fieldAliasLookup.TryGetValue(fieldName, out field);
        }

        public IEnumerator<Field> GetEnumerator()
        {
            return Fields.GetEnumerator();
        }

        protected internal override void WriteJsonFields(Newtonsoft.Json.JsonTextWriter writer, SchemaNames names, string encspace)
        {
            base.WriteJsonFields(writer, names, encspace);

            // we allow reading for empty fields, so writing of records with empty fields are allowed as well
            if (request)
                writer.WritePropertyName("request");
            else
                writer.WritePropertyName("fields");
            writer.WriteStartArray();

            if (null != this.Fields && this.Fields.Count > 0)
            {
                foreach (Field field in this)
                    field.writeJson(writer, names, this.Namespace); // use the namespace of the record for the fields
            }
            writer.WriteEndArray();
        }

        public override bool Equals(object obj)
        {
            if (obj == this) return true;
            if (obj != null && obj is RecordSchema)
            {
                RecordSchema that = obj as RecordSchema;
                return protect(() => true, () =>
                {
                    if (this.SchemaName.Equals(that.SchemaName) && this.Count == that.Count)
                    {
                        for (int i = 0; i < Fields.Count; i++) if (!Fields[i].Equals(that.Fields[i])) return false;
                        return areEqual(that.Props, this.Props);
                    }
                    return false;
                }, that);
            }
            return false;
        }

        public override int GetHashCode()
        {
            return protect(() => 0, () =>
            {
                int result = SchemaName.GetHashCode();
                foreach (Field f in Fields) result += 29 * f.GetHashCode();
                result += getHashCode(Props);
                return result;
            }, this);
        }

        public override bool CanRead(Schema writerSchema)
        {
            if ((writerSchema.Tag != Type.Record) && (writerSchema.Tag != Type.Error)) return false;

            RecordSchema that = writerSchema as RecordSchema;
            return protect(() => true, () =>
            {
                if (!that.SchemaName.Equals(SchemaName))
                    if (!InAliases(that.SchemaName)) 
                        return false;

                foreach (Field f in this)
                {
                    Field f2 = that[f.Name];
                    if (null == f2) // reader field not in writer field, check aliases of reader field if any match with a writer field
                        if (null != f.aliases)
                            foreach (string alias in f.aliases)
                            {
                                f2 = that[alias];
                                if (null != f2) break;
                            }

                    if (f2 == null && f.DefaultValue != null)
                        continue;         // Writer field missing, reader has default.

                    if (f2 != null && f.Schema.CanRead(f2.Schema)) continue;    // Both fields exist and are compatible.
                    return false;
                }
                return true;
            }, that);
        }

        private class RecordSchemaPair
        {
            public readonly RecordSchema first;
            public readonly RecordSchema second;

            public RecordSchemaPair(RecordSchema first, RecordSchema second)
            {
                this.first = first;
                this.second = second;
            }
        }

        [ThreadStatic]
        private static List<RecordSchemaPair> seen;

        private T protect<T>(Function<T> bypass, Function<T> main, RecordSchema that)
        {
            if (seen == null) 
                seen = new List<RecordSchemaPair>();

            else if (seen.Find((RecordSchemaPair rs) => rs.first == this && rs.second == that) != null) 
                return bypass();

            RecordSchemaPair p = new RecordSchemaPair(this, that);
            seen.Add(p);
            try { return main(); }
            finally { seen.Remove(p); }
        }

    }
}