/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
using System;
using System.Collections;
using System.Collections.Generic;
using System.Linq;
using Encoder = Avro.IO.Encoder;
namespace Avro.Generic
{
///
/// A general purpose writer of data from avro streams. This writer analyzes the writer schema
/// when constructed so that writes can be more efficient. Once constructed, a writer can be reused or shared among threads
/// to avoid incurring more resolution costs.
///
public abstract class PreresolvingDatumWriter : DatumWriter
{
///
public Schema Schema { get; private set; }
///
/// Defines the signature for a method that writes a value to an encoder.
///
/// Value to write
/// Encoder to write to
protected delegate void WriteItem(object value, Encoder encoder);
private readonly WriteItem _writer;
private readonly ArrayAccess _arrayAccess;
private readonly MapAccess _mapAccess;
private readonly Dictionary _recordWriters = new Dictionary();
///
public void Write(T datum, Encoder encoder)
{
_writer(datum, encoder);
}
///
/// Initializes a new instance of the class.
///
/// Schema used by this writer
/// Object used to access array properties
/// Object used to access map properties
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( v, schema.Tag, e.WriteBoolean );
case Schema.Type.Int:
return (v, e) => Write( v, schema.Tag, e.WriteInt );
case Schema.Type.Long:
return (v, e) => Write( v, schema.Tag, e.WriteLong );
case Schema.Type.Float:
return (v, e) => Write( v, schema.Tag, e.WriteFloat );
case Schema.Type.Double:
return (v, e) => Write( v, schema.Tag, e.WriteDouble );
case Schema.Type.String:
return (v, e) => Write( v, schema.Tag, e.WriteString );
case Schema.Type.Bytes:
return (v, e) => Write( 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);
case Schema.Type.Logical:
return ResolveLogical((LogicalSchema)schema);
default:
return (v, e) => Error(schema, v);
}
}
///
/// Serializes a "null"
///
/// The object to be serialized using null schema
/// The encoder to use while serialization
protected void WriteNull(object value, Encoder encoder)
{
if (value != null) throw TypeMismatch(value, "null", "null");
}
///
/// A generic method to serialize primitive Avro types.
///
/// Type of the C# type to be serialized
/// The value to be serialized
/// The schema type tag
/// The writer which should be used to write the given type.
protected void Write(object value, Schema.Type tag, Writer writer)
{
if (!(value is TValue)) throw TypeMismatch(value, tag.ToString(), typeof(TValue).ToString());
writer((TValue)value);
}
///
/// Serializes a record using the given RecordSchema. It uses GetField method
/// to extract the field value from the given object.
///
/// The RecordSchema to use for serialization
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;
}
///
/// Writes each field of a record to the encoder.
///
/// Record to write
/// Writers for each field in the record
/// Encoder to write to
protected abstract void WriteRecordFields(object record, RecordFieldWriter[] writers, Encoder encoder);
///
/// Correlates a record field with the writer used to serialize that field.
///
protected class RecordFieldWriter
{
///
/// Delegate used to write the to an encoder.
///
public WriteItem WriteField { get; set; }
///
/// Field this object is associated with.
///
public Field Field { get; set; }
}
///
/// Ensures that the given value is a record and that it corresponds to the given schema.
/// Throws an exception if either of those assertions are false.
///
/// Schema associated with the record
/// Ensure this object is a record
protected abstract void EnsureRecordObject(RecordSchema recordSchema, object value);
///
/// Writes a field from the to the
/// using the .
///
/// The record value from which the field needs to be extracted
/// The name of the field in the record
/// The position of field in the record
/// Used to write the field value to the encoder
/// Encoder to write to
protected abstract void WriteField(object record, string fieldName, int fieldPos, WriteItem writer, Encoder encoder );
///
/// Serializes an enumeration.
///
/// The EnumSchema for serialization
protected abstract WriteItem ResolveEnum(EnumSchema es);
///
/// Creates a delegate that serializes an array.
/// The default implementation calls EnsureArrayObject() to ascertain that the
/// given value is an array. It then calls GetArrayLength() and GetArrayElement()
/// to access the members of the array and then serialize them.
///
/// The ArraySchema for serialization
/// A that serializes an array.
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();
}
///
/// Serializes a logical value object by using the underlying logical type to convert the value
/// to its base value.
///
/// The logical schema.
protected WriteItem ResolveLogical(LogicalSchema schema)
{
var baseWriter = ResolveWriter(schema.BaseSchema);
return (d, e) => baseWriter(schema.LogicalType.ConvertToBaseValue(d, schema), e);
}
private WriteItem ResolveMap(MapSchema mapSchema)
{
var itemWriter = ResolveWriter(mapSchema.ValueSchema);
return (v, e) => WriteMap(itemWriter, v, e);
}
///
/// Serializes a map. The default implementation first ensure that the value is indeed a map and then uses
/// GetMapSize() and GetMapElements() to access the contents of the map.
///
/// Delegate used to write each map item.
/// The value to be serialized
/// The encoder for serialization
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);
}
///
/// Resolves the given value against the given UnionSchema and serializes the object against
/// the resolved schema member.
///
/// The UnionSchema to resolve against
/// Schemas for each type in the union
/// Writers for each type in the union
/// The value to be serialized
/// The encoder for serialization
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);
}
///
/// Finds the branch within the given UnionSchema that matches the given object. The default implementation
/// calls Matches() method in the order of branches within the UnionSchema. If nothing matches, throws
/// an exception.
///
/// The UnionSchema to resolve against
/// Schemas for types within the union
/// The object that should be used in matching
///
/// Index of the schema in that matches the .
///
///
/// No match found for the object in the union schema.
///
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);
}
///
/// Serialized a fixed object. The default implementation requires that the value is
/// a GenericFixed object with an identical schema as es.
///
/// The schema for serialization
/// The value to be serialized
/// The encoder for serialization
protected abstract void WriteFixed(FixedSchema es, object value, Encoder encoder);
///
/// Creates a new and uses the provided parameters to build an
/// exception message indicathing there was a type mismatch.
///
/// Object whose type does not the expected type
/// Schema that we tried to write against
/// Type that we expected
/// A new indicating a type mismatch.
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);
}
///
/// Tests whether the given schema an object are compatible.
///
/// Schema to compare
/// Object to compare
/// True if the two parameters are compatible, false otherwise.
protected abstract bool UnionBranchMatches(Schema sc, object obj);
///
/// Obsolete - This will be removed from the public API in a future version.
///
[Obsolete("This will be removed from the public API in a future version.")]
protected interface EnumAccess
{
///
/// Obsolete - This will be removed from the public API in a future version.
///
[Obsolete("This will be removed from the public API in a future version.")]
void WriteEnum(object value);
}
///
/// Defines the interface for a class that provides access to an array implementation.
///
protected interface ArrayAccess
{
///
/// Checks if the given object is an array. If it is a valid array, this function returns normally. Otherwise,
/// it throws an exception. The default implementation checks if the value is an array.
///
///
void EnsureArrayObject(object value);
///
/// Returns the length of an array. The default implementation requires the object
/// to be an array of objects and returns its length. The defaul implementation
/// gurantees that EnsureArrayObject() has been called on the value before this
/// function is called.
///
/// The object whose array length is required
/// The array length of the given object
long GetArrayLength(object value);
///
/// Writes each value in the given array to the using the
/// . The default implementation of this method requires
/// that the is an object array.
///
/// The array object
/// Value writer to send the array to.
/// Encoder to the write the array values to.
void WriteArrayValues(object array, WriteItem valueWriter, Encoder encoder);
}
///
/// Defines the interface for a class that provides access to a map implementation.
///
///
protected interface MapAccess
{
///
/// Checks if the given object is a map. If it is a valid map, this function returns normally. Otherwise,
/// it throws an exception. The default implementation checks if the value is an IDictionary<string, object>.
///
/// Ensures that this object is a valid map.
void EnsureMapObject(object value);
///
/// Returns the size of the map object. The default implementation gurantees that EnsureMapObject has been
/// successfully called with the given value. The default implementation requires the value
/// to be an IDictionary<string, object> and returns the number of elements in it.
///
/// The map object whose size is desired
/// The size of the given map object
long GetMapSize(object value);
///
/// Writes each value in the given map to the using the
/// . The default implementation of this method requires
/// that the is an IDictionary<string, object>.
///
/// Map object to write the contents of.
/// Value writer to send the map to.
/// Encoder to the write the map values to.
void WriteMapValues(object map, WriteItem valueWriter, Encoder encoder);
}
///
/// Provides access to map properties from an .
///
protected class DictionaryMapAccess : MapAccess
{
///
public void EnsureMapObject(object value)
{
if (value as IDictionary == null)
{
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);
}
}
}
}
}