package org.apache.avro.specific;
import org.apache.avro.Conversion;
import org.apache.avro.Schema;
import org.apache.avro.AvroRuntimeException;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.io.ResolvingDecoder;
import org.apache.avro.util.ClassUtils;
import java.io.IOException;
public class SpecificDatumReader<T> extends GenericDatumReader<T> {
public SpecificDatumReader() {
this(null, null, SpecificData.get());
}
public SpecificDatumReader(Class<T> c) {
this(SpecificData.getForClass(c));
setSchema(getSpecificData().getSchema(c));
}
public SpecificDatumReader(Schema schema) {
this(schema, schema, SpecificData.getForSchema(schema));
}
public SpecificDatumReader(Schema writer, Schema reader) {
this(writer, reader, SpecificData.getForSchema(reader));
}
public SpecificDatumReader(Schema writer, Schema reader, SpecificData data) {
super(writer, reader, data);
}
public SpecificDatumReader(SpecificData data) {
super(data);
}
public SpecificData getSpecificData() {
return (SpecificData) getData();
}
@Override
public void setSchema(Schema actual) {
if (getExpected() == null && actual != null && actual.getType() == Schema.Type.RECORD) {
SpecificData data = getSpecificData();
Class c = data.getClass(actual);
if (c != null && SpecificRecord.class.isAssignableFrom(c))
setExpected(data.getSchema(c));
}
super.setSchema(actual);
}
@Override
protected Class findStringClass(Schema schema) {
Class stringClass = null;
switch (schema.getType()) {
case STRING:
stringClass = getPropAsClass(schema, SpecificData.CLASS_PROP);
break;
case MAP:
stringClass = getPropAsClass(schema, SpecificData.KEY_CLASS_PROP);
break;
}
if (stringClass != null)
return stringClass;
return super.findStringClass(schema);
}
private Class getPropAsClass(Schema schema, String prop) {
String name = schema.getProp(prop);
if (name == null)
return null;
try {
return ClassUtils.forName(getData().getClassLoader(), name);
} catch (ClassNotFoundException e) {
throw new AvroRuntimeException(e);
}
}
@Override
protected Object readRecord(Object old, Schema expected, ResolvingDecoder in) throws IOException {
SpecificData data = getSpecificData();
if (data.useCustomCoders()) {
old = data.newRecord(old, expected);
if (old instanceof SpecificRecordBase) {
SpecificRecordBase d = (SpecificRecordBase) old;
if (d.hasCustomCoders()) {
d.customDecode(in);
return d;
}
}
}
return super.readRecord(old, expected, in);
}
@Override
protected void readField(Object r, Schema.Field f, Object oldDatum, ResolvingDecoder in, Object state)
throws IOException {
if (r instanceof SpecificRecordBase) {
Conversion<?> conversion = ((SpecificRecordBase) r).getConversion(f.pos());
Object datum;
if (conversion != null) {
datum = readWithConversion(oldDatum, f.schema(), f.schema().getLogicalType(), conversion, in);
} else {
datum = readWithoutConversion(oldDatum, f.schema(), in);
}
getData().setField(r, f.name(), f.pos(), datum);
} else {
super.readField(r, f, oldDatum, in, state);
}
}
}