/*
 * 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.
 */
package org.apache.avro.specific;

import java.lang.reflect.Field;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Map;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.WeakHashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.LinkedHashMap;
import java.nio.ByteBuffer;
import java.lang.reflect.Constructor;
import java.lang.reflect.ParameterizedType;
import java.io.ObjectInput;
import java.io.ObjectOutput;

import org.apache.avro.Schema;
import org.apache.avro.Protocol;
import org.apache.avro.AvroRuntimeException;
import org.apache.avro.AvroTypeException;
import org.apache.avro.Schema.Type;
import org.apache.avro.util.ClassUtils;
import org.apache.avro.generic.GenericData;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.BinaryDecoder;

Utilities for generated Java classes and interfaces. /** Utilities for generated Java classes and interfaces. */
public class SpecificData extends GenericData {

  private static final SpecificData INSTANCE = new SpecificData();

  private static final Class<?>[] NO_ARG = new Class[] {};
  private static final Class<?>[] SCHEMA_ARG = new Class[] { Schema.class };
  private static final ClassValue<Constructor> CTOR_CACHE = new ClassValue<Constructor>() {
    @Override
    protected Constructor computeValue(Class<?> c) {
      boolean useSchema = SchemaConstructable.class.isAssignableFrom(c);
      try {
        Constructor meth = c.getDeclaredConstructor(useSchema ? SCHEMA_ARG : NO_ARG);
        meth.setAccessible(true);
        return meth;
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }

  };

  public static final String CLASS_PROP = "java-class";
  public static final String KEY_CLASS_PROP = "java-key-class";
  public static final String ELEMENT_PROP = "java-element-class";

  
List of Java reserved words from
https://docs.oracle.com/javase/specs/jls/se8/html/jls-3.html#jls-3.9 combined
with the boolean and null literals. combined with the classnames used
internally in the generated avro code.
/**
   * List of Java reserved words from
   * https://docs.oracle.com/javase/specs/jls/se8/html/jls-3.html#jls-3.9 combined
   * with the boolean and null literals. combined with the classnames used
   * internally in the generated avro code.
   */
  public static final Set<String> RESERVED_WORDS = new HashSet<>(
      Arrays.asList("abstract", "assert", "boolean", "break", "byte", "case", "catch", "char", "class", "const",
          "continue", "default", "do", "double", "else", "enum", "extends", "false", "final", "finally", "float", "for",
          "goto", "if", "implements", "import", "instanceof", "int", "interface", "long", "native", "new", "null",
          "package", "private", "protected", "public", "return", "short", "static", "strictfp", "super", "switch",
          "synchronized", "this", "throw", "throws", "transient", "true", "try", "void", "volatile", "while",
          /* classnames use internally by the avro code generator */
          "Builder"));

  
Read/write some common builtin classes as strings. Representing these as
strings isn't always best, as they aren't always ordered ideally, but at
least they're stored. Also note that, for compatibility, only classes that
wouldn't be otherwise correctly readable or writable should be added here,
e.g., those without a no-arg constructor or those whose fields are all
transient.
/**
   * Read/write some common builtin classes as strings. Representing these as
   * strings isn't always best, as they aren't always ordered ideally, but at
   * least they're stored. Also note that, for compatibility, only classes that
   * wouldn't be otherwise correctly readable or writable should be added here,
   * e.g., those without a no-arg constructor or those whose fields are all
   * transient.
   */
  protected Set<Class> stringableClasses = new HashSet<>();
  {
    stringableClasses.add(java.math.BigDecimal.class);
    stringableClasses.add(java.math.BigInteger.class);
    stringableClasses.add(java.net.URI.class);
    stringableClasses.add(java.net.URL.class);
    stringableClasses.add(java.io.File.class);
  }

  
For subclasses. Applications normally use get(). /** For subclasses. Applications normally use {@link SpecificData#get()}. */
  public SpecificData() {
  }

  
Construct with a specific classloader. /** Construct with a specific classloader. */
  public SpecificData(ClassLoader classLoader) {
    super(classLoader);
  }

  @Override
  public DatumReader createDatumReader(Schema schema) {
    return new SpecificDatumReader(schema, schema, this);
  }

  @Override
  public DatumReader createDatumReader(Schema writer, Schema reader) {
    return new SpecificDatumReader(writer, reader, this);
  }

  @Override
  public DatumWriter createDatumWriter(Schema schema) {
    return new SpecificDatumWriter(schema, this);
  }

  
Return the singleton instance. /** Return the singleton instance. */
  public static SpecificData get() {
    return INSTANCE;
  }

  
For RECORD type schemas, this method returns the SpecificData instance of the
class associated with the schema, in order to get the right conversions for
any logical types used.
Params:
reader – the reader schema
Returns:
the SpecificData associated with the schema's class, or the default
        instance./**
   * For RECORD type schemas, this method returns the SpecificData instance of the
   * class associated with the schema, in order to get the right conversions for
   * any logical types used.
   *
   * @param reader the reader schema
   * @return the SpecificData associated with the schema's class, or the default
   *         instance.
   */
  public static SpecificData getForSchema(Schema reader) {
    if (reader.getType() == Type.RECORD) {
      final String className = getClassName(reader);
      if (className != null) {
        final Class<?> clazz;
        try {
          clazz = Class.forName(className);
          return getForClass(clazz);
        } catch (ClassNotFoundException e) {
          return SpecificData.get();
        }
      }
    }
    return SpecificData.get();
  }

  
If the given class is assignable to SpecificRecordBase, this method returns the SpecificData instance from the field MODEL$, in order to get the correct Conversion instances for the class. Falls back to the default instance get() for other classes or if the field is not found. 
Params:
c – A class
Type parameters:
<T> – .
Returns:
The SpecificData from the SpecificRecordBase instance, or the default
        SpecificData instance./**
   * If the given class is assignable to {@link SpecificRecordBase}, this method
   * returns the SpecificData instance from the field {@code MODEL$}, in order to
   * get the correct {@link org.apache.avro.Conversion} instances for the class.
   * Falls back to the default instance {@link SpecificData#get()} for other
   * classes or if the field is not found.
   *
   * @param c A class
   * @param   <T> .
   * @return The SpecificData from the SpecificRecordBase instance, or the default
   *         SpecificData instance.
   */
  public static <T> SpecificData getForClass(Class<T> c) {
    if (SpecificRecordBase.class.isAssignableFrom(c)) {
      final Field specificDataField;
      try {
        specificDataField = c.getDeclaredField("MODEL$");
        specificDataField.setAccessible(true);
        return (SpecificData) specificDataField.get(null);
      } catch (NoSuchFieldException e) {
        // Return default instance
        return SpecificData.get();
      } catch (IllegalAccessException e) {
        throw new AvroRuntimeException(e);
      }
    }
    return SpecificData.get();
  }

  private boolean useCustomCoderFlag = Boolean
      .parseBoolean(System.getProperty("org.apache.avro.specific.use_custom_coders", "false"));

  
Retrieve the current value of the custom-coders feature flag. Defaults to
true, but this default can be overriden using the system
property org.apache.avro.specific.use_custom_coders, and can be set dynamically by useCustomCoders(). See  for more about this feature flag.
/**
   * Retrieve the current value of the custom-coders feature flag. Defaults to
   * <code>true</code>, but this default can be overriden using the system
   * property <code>org.apache.avro.specific.use_custom_coders</code>, and can be
   * set dynamically by {@link SpecificData#useCustomCoders()}. See <a
   * href="https://avro.apache.org/docs/current/gettingstartedjava.html#Beta+feature:+Generating+faster+code"Getting
   * started with Java</a> for more about this feature flag.
   */
  public boolean useCustomCoders() {
    return useCustomCoderFlag;
  }

  
Dynamically set the value of the custom-coder feature flag. See useCustomCoders(). /**
   * Dynamically set the value of the custom-coder feature flag. See
   * {@link SpecificData#useCustomCoders()}.
   */
  public void setCustomCoders(boolean flag) {
    useCustomCoderFlag = flag;
  }

  @Override
  protected boolean isEnum(Object datum) {
    return datum instanceof Enum || super.isEnum(datum);
  }

  @Override
  public Object createEnum(String symbol, Schema schema) {
    Class c = getClass(schema);
    if (c == null)
      return super.createEnum(symbol, schema); // punt to generic
    if (RESERVED_WORDS.contains(symbol))
      symbol += "$";
    return Enum.valueOf(c, symbol);
  }

  @Override
  protected Schema getEnumSchema(Object datum) {
    return (datum instanceof Enum) ? getSchema(datum.getClass()) : super.getEnumSchema(datum);
  }

  private Map<String, Class> classCache = new ConcurrentHashMap<>();

  private static final Class NO_CLASS = new Object() {
  }.getClass();
  private static final Schema NULL_SCHEMA = Schema.create(Schema.Type.NULL);

  
Return the class that implements a schema, or null if none exists. /** Return the class that implements a schema, or null if none exists. */
  public Class getClass(Schema schema) {
    switch (schema.getType()) {
    case FIXED:
    case RECORD:
    case ENUM:
      String name = schema.getFullName();
      if (name == null)
        return null;
      Class c = classCache.get(name);
      if (c == null) {
        try {
          c = ClassUtils.forName(getClassLoader(), getClassName(schema));
        } catch (ClassNotFoundException e) {
          try { // nested class?
            c = ClassUtils.forName(getClassLoader(), getNestedClassName(schema));
          } catch (ClassNotFoundException ex) {
            c = NO_CLASS;
          }
        }
        classCache.put(name, c);
      }
      return c == NO_CLASS ? null : c;
    case ARRAY:
      return List.class;
    case MAP:
      return Map.class;
    case UNION:
      List<Schema> types = schema.getTypes(); // elide unions with null
      if ((types.size() == 2) && types.contains(NULL_SCHEMA))
        return getWrapper(types.get(types.get(0).equals(NULL_SCHEMA) ? 1 : 0));
      return Object.class;
    case STRING:
      if (STRING_TYPE_STRING.equals(schema.getProp(STRING_PROP)))
        return String.class;
      return CharSequence.class;
    case BYTES:
      return ByteBuffer.class;
    case INT:
      return Integer.TYPE;
    case LONG:
      return Long.TYPE;
    case FLOAT:
      return Float.TYPE;
    case DOUBLE:
      return Double.TYPE;
    case BOOLEAN:
      return Boolean.TYPE;
    case NULL:
      return Void.TYPE;
    default:
      throw new AvroRuntimeException("Unknown type: " + schema);
    }
  }

  private Class getWrapper(Schema schema) {
    switch (schema.getType()) {
    case INT:
      return Integer.class;
    case LONG:
      return Long.class;
    case FLOAT:
      return Float.class;
    case DOUBLE:
      return Double.class;
    case BOOLEAN:
      return Boolean.class;
    }
    return getClass(schema);
  }

  
Returns the Java class name indicated by a schema's name and namespace. /** Returns the Java class name indicated by a schema's name and namespace. */
  public static String getClassName(Schema schema) {
    String namespace = schema.getNamespace();
    String name = schema.getName();
    if (namespace == null || "".equals(namespace))
      return name;
    String dot = namespace.endsWith("$") ? "" : "."; // back-compatibly handle $
    return namespace + dot + name;
  }

  private String getNestedClassName(Schema schema) {
    String namespace = schema.getNamespace();
    String name = schema.getName();
    if (namespace == null || "".equals(namespace))
      return name;
    return namespace + "$" + name;
  }

  // cache for schemas created from Class objects. Use ClassValue to avoid
  // locking classloaders and is GC and thread safe.
  private final ClassValue<Schema> schemaClassCache = new ClassValue<Schema>() {
    @Override
    protected Schema computeValue(Class<?> type) {
      return createSchema(type, new LinkedHashMap<>());
    }
  };
  // for non-class objects, use a WeakHashMap, but this needs a sync block around
  // it
  private final Map<java.lang.reflect.Type, Schema> schemaTypeCache = Collections.synchronizedMap(new WeakHashMap<>());

  
Find the schema for a Java type. /** Find the schema for a Java type. */
  public Schema getSchema(java.lang.reflect.Type type) {
    try {
      if (type instanceof Class) {
        return schemaClassCache.get((Class<?>) type);
      }
      return schemaTypeCache.computeIfAbsent(type, t -> createSchema(t, new LinkedHashMap<>()));
    } catch (Exception e) {
      throw (e instanceof AvroRuntimeException) ? (AvroRuntimeException) e : new AvroRuntimeException(e);
    }
  }

  
Create the schema for a Java type. /** Create the schema for a Java type. */
  @SuppressWarnings(value = "unchecked")
  protected Schema createSchema(java.lang.reflect.Type type, Map<String, Schema> names) {
    if (type instanceof Class && CharSequence.class.isAssignableFrom((Class) type))
      return Schema.create(Type.STRING);
    else if (type == ByteBuffer.class)
      return Schema.create(Type.BYTES);
    else if ((type == Integer.class) || (type == Integer.TYPE))
      return Schema.create(Type.INT);
    else if ((type == Long.class) || (type == Long.TYPE))
      return Schema.create(Type.LONG);
    else if ((type == Float.class) || (type == Float.TYPE))
      return Schema.create(Type.FLOAT);
    else if ((type == Double.class) || (type == Double.TYPE))
      return Schema.create(Type.DOUBLE);
    else if ((type == Boolean.class) || (type == Boolean.TYPE))
      return Schema.create(Type.BOOLEAN);
    else if ((type == Void.class) || (type == Void.TYPE))
      return Schema.create(Type.NULL);
    else if (type instanceof ParameterizedType) {
      ParameterizedType ptype = (ParameterizedType) type;
      Class raw = (Class) ptype.getRawType();
      java.lang.reflect.Type[] params = ptype.getActualTypeArguments();
      if (Collection.class.isAssignableFrom(raw)) { // array
        if (params.length != 1)
          throw new AvroTypeException("No array type specified.");
        return Schema.createArray(createSchema(params[0], names));
      } else if (Map.class.isAssignableFrom(raw)) { // map
        java.lang.reflect.Type key = params[0];
        java.lang.reflect.Type value = params[1];
        if (!(key instanceof Class && CharSequence.class.isAssignableFrom((Class) key)))
          throw new AvroTypeException("Map key class not CharSequence: " + key);
        return Schema.createMap(createSchema(value, names));
      } else {
        return createSchema(raw, names);
      }
    } else if (type instanceof Class) { // class
      Class c = (Class) type;
      String fullName = c.getName();
      Schema schema = names.get(fullName);
      if (schema == null)
        try {
          schema = (Schema) (c.getDeclaredField("SCHEMA$").get(null));

          if (!fullName.equals(getClassName(schema)))
            // HACK: schema mismatches class. maven shade plugin? try replacing.
            schema = new Schema.Parser()
                .parse(schema.toString().replace(schema.getNamespace(), c.getPackage().getName()));
        } catch (NoSuchFieldException e) {
          throw new AvroRuntimeException("Not a Specific class: " + c);
        } catch (IllegalAccessException e) {
          throw new AvroRuntimeException(e);
        }
      names.put(fullName, schema);
      return schema;
    }
    throw new AvroTypeException("Unknown type: " + type);
  }

  @Override
  protected String getSchemaName(Object datum) {
    if (datum != null) {
      Class c = datum.getClass();
      if (isStringable(c))
        return Schema.Type.STRING.getName();
    }
    return super.getSchemaName(datum);
  }

  
True if a class should be serialized with toString(). /** True if a class should be serialized with toString(). */
  protected boolean isStringable(Class<?> c) {
    return stringableClasses.contains(c);
  }

  
True if a class IS a string type /** True if a class IS a string type */
  protected boolean isStringType(Class<?> c) {
    // this will return true for String, Utf8, CharSequence
    return CharSequence.class.isAssignableFrom(c);
  }

  
Return the protocol for a Java interface. /** Return the protocol for a Java interface. */
  public Protocol getProtocol(Class iface) {
    try {
      Protocol p = (Protocol) (iface.getDeclaredField("PROTOCOL").get(null));
      if (!p.getNamespace().equals(iface.getPackage().getName()))
        // HACK: protocol mismatches iface. maven shade plugin? try replacing.
        p = Protocol.parse(p.toString().replace(p.getNamespace(), iface.getPackage().getName()));
      return p;
    } catch (NoSuchFieldException e) {
      throw new AvroRuntimeException("Not a Specific protocol: " + iface);
    } catch (IllegalAccessException e) {
      throw new AvroRuntimeException(e);
    }
  }

  @Override
  protected int compare(Object o1, Object o2, Schema s, boolean eq) {
    switch (s.getType()) {
    case ENUM:
      if (o1 instanceof Enum)
        return ((Enum) o1).ordinal() - ((Enum) o2).ordinal();
    default:
      return super.compare(o1, o2, s, eq);
    }
  }

  
Create an instance of a class. If the class implements SchemaConstructable, call a constructor with a Schema parameter, otherwise use a no-arg constructor. /**
   * Create an instance of a class. If the class implements
   * {@link SchemaConstructable}, call a constructor with a
   * {@link org.apache.avro.Schema} parameter, otherwise use a no-arg constructor.
   */
  @SuppressWarnings("unchecked")
  public static Object newInstance(Class c, Schema s) {
    boolean useSchema = SchemaConstructable.class.isAssignableFrom(c);
    Object result;
    try {
      Constructor meth = CTOR_CACHE.get(c);
      result = meth.newInstance(useSchema ? new Object[] { s } : null);
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
    return result;
  }

  @Override
  public Object createFixed(Object old, Schema schema) {
    Class c = getClass(schema);
    if (c == null)
      return super.createFixed(old, schema); // punt to generic
    return c.isInstance(old) ? old : newInstance(c, schema);
  }

  @Override
  public Object newRecord(Object old, Schema schema) {
    Class c = getClass(schema);
    if (c == null)
      return super.newRecord(old, schema); // punt to generic
    return (c.isInstance(old) ? old : newInstance(c, schema));
  }

  
Tag interface that indicates that a class has a one-argument constructor that
accepts a Schema.
See Also:
newInstance/**
   * Tag interface that indicates that a class has a one-argument constructor that
   * accepts a Schema.
   * 
   * @see #newInstance
   */
  public interface SchemaConstructable {
  }

  
Runtime utility used by generated classes. /** Runtime utility used by generated classes. */
  public static BinaryDecoder getDecoder(ObjectInput in) {
    return DecoderFactory.get().directBinaryDecoder(new ExternalizableInput(in), null);
  }

  
Runtime utility used by generated classes. /** Runtime utility used by generated classes. */
  public static BinaryEncoder getEncoder(ObjectOutput out) {
    return EncoderFactory.get().directBinaryEncoder(new ExternalizableOutput(out), null);
  }

}