avro-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cutt...@apache.org
Subject svn commit: r835165 - in /hadoop/avro/trunk: ./ src/java/org/apache/avro/generic/ src/java/org/apache/avro/reflect/ src/java/org/apache/avro/specific/ src/test/java/org/apache/avro/
Date Wed, 11 Nov 2009 23:53:04 GMT
Author: cutting
Date: Wed Nov 11 23:53:00 2009
New Revision: 835165

URL: http://svn.apache.org/viewvc?rev=835165&view=rev
Log:
AVRO-185. Java's specific API no longer depends on reflection.

Added:
    hadoop/avro/trunk/src/java/org/apache/avro/specific/FixedSize.java
      - copied, changed from r834652, hadoop/avro/trunk/src/java/org/apache/avro/reflect/FixedSize.java
Removed:
    hadoop/avro/trunk/src/java/org/apache/avro/reflect/FixedSize.java
Modified:
    hadoop/avro/trunk/CHANGES.txt
    hadoop/avro/trunk/build.xml
    hadoop/avro/trunk/src/java/org/apache/avro/generic/GenericRequestor.java
    hadoop/avro/trunk/src/java/org/apache/avro/generic/GenericResponder.java
    hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectData.java
    hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectDatumReader.java
    hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectDatumWriter.java
    hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectRequestor.java
    hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectResponder.java
    hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificCompiler.java
    hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificData.java
    hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificDatumReader.java
    hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificDatumWriter.java
    hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificFixed.java
    hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificRequestor.java
    hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificResponder.java
    hadoop/avro/trunk/src/test/java/org/apache/avro/TestDataFileReflect.java

Modified: hadoop/avro/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/avro/trunk/CHANGES.txt?rev=835165&r1=835164&r2=835165&view=diff
==============================================================================
--- hadoop/avro/trunk/CHANGES.txt (original)
+++ hadoop/avro/trunk/CHANGES.txt Wed Nov 11 23:53:00 2009
@@ -4,6 +4,10 @@
 
   INCOMPATIBLE CHANGES
 
+    AVRO-185. Java's specific API no longer depends on reflection.
+    This reverses the inheritance of most classes in the specific and
+    reflect packages. (cutting)
+
   NEW FEATURES
 
     AVRO-151. Validating Avro schema parser for C (massie)

Modified: hadoop/avro/trunk/build.xml
URL: http://svn.apache.org/viewvc/hadoop/avro/trunk/build.xml?rev=835165&r1=835164&r2=835165&view=diff
==============================================================================
--- hadoop/avro/trunk/build.xml (original)
+++ hadoop/avro/trunk/build.xml Wed Nov 11 23:53:00 2009
@@ -237,6 +237,13 @@
     <java-compiler src="${test.java.src.dir}"
 		   dest="${test.java.classes}"
 		   classpath="test.java.classpath"/>
+    <taskdef
+       name="paranamer" 
+       classname="com.thoughtworks.paranamer.ant.ParanamerGeneratorTask">
+      <classpath refid="java.classpath" />
+    </taskdef>
+    <paranamer sourceDirectory="${test.java.generated.dir}"
+	       outputDirectory="${test.java.generated.classes}"/>
   </target>
 
   <macrodef name="java-avro-compiler">
@@ -253,11 +260,6 @@
       <taskdef name="schema" classname="org.apache.avro.specific.SchemaTask">
 	<classpath refid="java.classpath" />
       </taskdef>
-      <taskdef
-	 name="paranamer" 
-	 classname="com.thoughtworks.paranamer.ant.ParanamerGeneratorTask">
-	<classpath refid="java.classpath" />
-      </taskdef>
 
       <mkdir dir="@{generated}"/>
       
@@ -275,7 +277,6 @@
 
       <java-compiler src="@{generated}" dest="@{dest}"
 		     classpath="@{classpath}"/>
-      <paranamer sourceDirectory="@{generated}" outputDirectory="@{dest}"/>
     </sequential>
   </macrodef>
 

Modified: hadoop/avro/trunk/src/java/org/apache/avro/generic/GenericRequestor.java
URL: http://svn.apache.org/viewvc/hadoop/avro/trunk/src/java/org/apache/avro/generic/GenericRequestor.java?rev=835165&r1=835164&r2=835165&view=diff
==============================================================================
--- hadoop/avro/trunk/src/java/org/apache/avro/generic/GenericRequestor.java (original)
+++ hadoop/avro/trunk/src/java/org/apache/avro/generic/GenericRequestor.java Wed Nov 11 23:53:00 2009
@@ -35,15 +35,18 @@
     super(protocol, transceiver);
   }
 
+  @Override
   public void writeRequest(Schema schema, Object request, Encoder out)
     throws IOException {
     new GenericDatumWriter<Object>(schema).write(request, out);
   }
 
+  @Override
   public Object readResponse(Schema schema, Decoder in) throws IOException {
     return new GenericDatumReader<Object>(schema).read(null, in);
   }
 
+  @Override
   public AvroRemoteException readError(Schema schema, Decoder in)
     throws IOException {
     return new AvroRemoteException(new GenericDatumReader<Object>(schema).read(null,in));

Modified: hadoop/avro/trunk/src/java/org/apache/avro/generic/GenericResponder.java
URL: http://svn.apache.org/viewvc/hadoop/avro/trunk/src/java/org/apache/avro/generic/GenericResponder.java?rev=835165&r1=835164&r2=835165&view=diff
==============================================================================
--- hadoop/avro/trunk/src/java/org/apache/avro/generic/GenericResponder.java (original)
+++ hadoop/avro/trunk/src/java/org/apache/avro/generic/GenericResponder.java Wed Nov 11 23:53:00 2009
@@ -34,18 +34,18 @@
     super(local);
   }
 
-  /** Reads a request message. */
+  @Override
   public Object readRequest(Schema schema, Decoder in) throws IOException {
     return new GenericDatumReader<Object>(schema).read(null, in);
   }
 
-  /** Writes a response message. */
+  @Override
   public void writeResponse(Schema schema, Object response, Encoder out)
     throws IOException {
     new GenericDatumWriter<Object>(schema).write(response, out);
   }
 
-  /** Writes an error message. */
+  @Override
   public void writeError(Schema schema, AvroRemoteException error,
                          Encoder out) throws IOException {
     new GenericDatumWriter<Object>(schema).write(error.getValue(), out);

Modified: hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectData.java
URL: http://svn.apache.org/viewvc/hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectData.java?rev=835165&r1=835164&r2=835165&view=diff
==============================================================================
--- hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectData.java (original)
+++ hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectData.java Wed Nov 11 23:53:00 2009
@@ -20,34 +20,28 @@
 import java.lang.reflect.Field;
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
-import java.lang.reflect.ParameterizedType;
-import java.nio.ByteBuffer;
+import java.util.List;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.LinkedHashMap;
-import java.util.List;
 import java.util.Map;
-import java.util.WeakHashMap;
-import java.util.concurrent.ConcurrentHashMap;
 
 import org.apache.avro.AvroRuntimeException;
-import org.apache.avro.AvroTypeException;
 import org.apache.avro.Protocol;
 import org.apache.avro.Schema;
-import org.apache.avro.Protocol.Message;
 import org.apache.avro.Schema.Type;
-import org.apache.avro.generic.GenericData;
-import org.apache.avro.generic.GenericArray;
+import org.apache.avro.Protocol.Message;
 import org.apache.avro.generic.GenericFixed;
+import org.apache.avro.specific.SpecificData;
+import org.apache.avro.specific.FixedSize;
 import org.apache.avro.ipc.AvroRemoteException;
-import org.apache.avro.util.Utf8;
 
 import com.thoughtworks.paranamer.CachingParanamer;
 import com.thoughtworks.paranamer.Paranamer;
 
 /** Utilities to use existing Java classes and interfaces via reflection. */
-public class ReflectData extends GenericData {
+public class ReflectData extends SpecificData {
   
   /** {@link ReflectData} implementation that permits null field values.  The
    * schema generated for each field is a union of its declared type and
@@ -86,12 +80,6 @@
   }
 
   @Override
-  protected boolean isEnum(Object datum) {
-    return datum instanceof Enum;
-  }
-
-  /** Returns true if an object matches a schema. */
-  @Override
   public boolean validate(Schema schema, Object datum) {
     switch (schema.getType()) {
     case RECORD:
@@ -107,30 +95,8 @@
         }
       }
       return true;
-    case ENUM:
-      return datum instanceof Enum
-        && schema.getEnumSymbols().contains(((Enum)datum).name());
-    case ARRAY:
-      if (!(datum instanceof GenericArray)) return false;
-      for (Object element : (GenericArray)datum)
-        if (!validate(schema.getElementType(), element))
-          return false;
-      return true;
-    case UNION:
-      for (Schema type : schema.getTypes())
-        if (validate(type, datum))
-          return true;
-      return false;
-    case FIXED:   return datum instanceof GenericFixed;
-    case STRING:  return datum instanceof Utf8;
-    case BYTES:   return datum instanceof ByteBuffer;
-    case INT:     return datum instanceof Integer;
-    case LONG:    return datum instanceof Long;
-    case FLOAT:   return datum instanceof Float;
-    case DOUBLE:  return datum instanceof Double;
-    case BOOLEAN: return datum instanceof Boolean;
-    case NULL:    return datum == null;
-    default: return false;
+    default:
+      return super.validate(schema, datum);
     }
   }
 
@@ -144,152 +110,42 @@
     }
   }
 
-  private Map<String,Class> classCache = new ConcurrentHashMap<String,Class>();
-
-  /** Return the class that implements this schema. */
-  public Class getClass(Schema schema) {
-    switch (schema.getType()) {
-    case FIXED:
-    case RECORD:
-    case ENUM:
-      String full = schema.getFullName();
-      Class c = classCache.get(full);
-      if (c == null) {
-        try {
-          c = Class.forName(getClassName(schema));
-          classCache.put(full, c);
-        } catch (ClassNotFoundException e) {
-          throw new AvroRuntimeException(e);
-        }
-      }
-      return c;
-    case ARRAY:   return GenericArray.class;
-    case MAP:     return Map.class;
-    case UNION:   return Object.class;
-    case STRING:  return Utf8.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);
-    }
-
-  }
-
-  /** Returns the Java class name indicated by a schema's name and namespace. */
-  public String getClassName(Schema schema) {
-    String namespace = schema.getNamespace();
-    String name = schema.getName();
-    if (namespace == null)
-      return name;
-    String dot = namespace.endsWith("$") ? "" : ".";
-    return namespace + dot + name;
-  }
-
-  private final WeakHashMap<java.lang.reflect.Type,Schema> schemaCache =
-    new WeakHashMap<java.lang.reflect.Type,Schema>();
-
-  /** Generate a schema for a Java type.
-   * <p>For records, {@link Class#getDeclaredFields() declared fields} (not
-   * inherited) which are not static or transient are used.</p>
-   * <p>Note that unions cannot be automatically generated by this method,
-   * since Java provides no representation for unions.</p>
-   */
-  public Schema getSchema(java.lang.reflect.Type type) {
-    Schema schema = schemaCache.get(type);
-    if (schema == null) {
-      schema = createSchema(type, new LinkedHashMap<String,Schema>());
-      schemaCache.put(type, schema);
-    }
-    return schema;
-  }
-
-  /**
-   * Create a schema for a type and it's fields. Note that by design only fields
-   * of the direct class, not it's super classes, are used for creating the
-   * schema.  Also, fields are not permitted to be null.
-   */
+  /** Create a schema for a Java class.  Note that by design only fields of the
+   * direct class, not it's super classes, are used for creating a record
+   * schema.  Also, fields are not permitted to be null.  {@link
+   * Class#getDeclaredFields() Declared fields} (not inherited) which are not
+   * static or transient are used.*/
+  @Override
   @SuppressWarnings(value="unchecked")
-  protected Schema createSchema(java.lang.reflect.Type type,
-                                Map<String,Schema> names) {
-    if (type == Utf8.class)
-      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();
-      for (int i = 0; i < params.length; i++)
-      if (GenericArray.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 == Utf8.class))
-          throw new AvroTypeException("Map key class not Utf8: "+key);
-        return Schema.createMap(createSchema(value, names));
-      }
-    } else if (type instanceof Class) {
-      Class c = (Class)type;
-      String name = c.getSimpleName();
-      String space = c.getPackage().getName();
-      if (c.getEnclosingClass() != null)          // nested class
-        space = c.getEnclosingClass().getName() + "$";
-      String fullName = c.getName();
-      Schema schema = names.get(fullName);
-      if (schema == null) {
-
-        if (c.isEnum()) {                         // enum
-          List<String> symbols = new ArrayList<String>();
-          Enum[] constants = (Enum[])c.getEnumConstants();
-          for (int i = 0; i < constants.length; i++)
-            symbols.add(constants[i].name());
-          schema = Schema.createEnum(name, space, symbols);
-          names.put(fullName, schema);
-          return schema;
-        }
-                                                  // fixed
-        if (GenericFixed.class.isAssignableFrom(c)) {
-          int size = ((FixedSize)c.getAnnotation(FixedSize.class)).value();
-          schema = Schema.createFixed(name, space, size);
-          names.put(fullName, schema);
-          return schema;
+  protected Schema createClassSchema(Class c, Map<String,Schema> names) {
+    String name = c.getSimpleName();
+    String space = c.getPackage().getName();
+    if (c.getEnclosingClass() != null)                   // nested class
+      space = c.getEnclosingClass().getName() + "$";
+    Schema schema;
+    if (c.isEnum()) {                                    // enum
+      List<String> symbols = new ArrayList<String>();
+      Enum[] constants = (Enum[])c.getEnumConstants();
+      for (int i = 0; i < constants.length; i++)
+        symbols.add(constants[i].name());
+      schema = Schema.createEnum(name, space, symbols);
+    } else if (GenericFixed.class.isAssignableFrom(c)) { // fixed
+      int size = ((FixedSize)c.getAnnotation(FixedSize.class)).value();
+      schema = Schema.createFixed(name, space, size);
+    } else {                                             // record
+      LinkedHashMap<String,Schema.Field> fields =
+        new LinkedHashMap<String,Schema.Field>();
+      schema = Schema.createRecord(name, space,
+                                   Throwable.class.isAssignableFrom(c));
+      names.put(c.getName(), schema);
+      for (Field field : c.getDeclaredFields())
+        if ((field.getModifiers()&(Modifier.TRANSIENT|Modifier.STATIC))==0) {
+          Schema fieldSchema = createFieldSchema(field, names);
+          fields.put(field.getName(), new Schema.Field(fieldSchema, null));
         }
-                                                  // record
-        LinkedHashMap<String,Schema.Field> fields =
-          new LinkedHashMap<String,Schema.Field>();
-        schema = Schema.createRecord(name, space,
-                                     Throwable.class.isAssignableFrom(c));
-        if (!names.containsKey(fullName))
-          names.put(fullName, schema);
-        for (Field field : c.getDeclaredFields())
-          if ((field.getModifiers()&(Modifier.TRANSIENT|Modifier.STATIC))==0) {
-            Schema fieldSchema = createFieldSchema(field, names);
-            fields.put(field.getName(), new Schema.Field(fieldSchema, null));
-          }
-        schema.setFields(fields);
-      }
-      return schema;
+      schema.setFields(fields);
     }
-    throw new AvroTypeException("Unknown type: "+type);
+    return schema;
   }
 
   /** Create a schema for a field. */
@@ -297,12 +153,12 @@
     return createSchema(field.getGenericType(), names);
   }
 
-  /** Generate a protocol for a Java interface.
+  /** Return the protocol for a Java interface.
    * <p>Note that this requires that <a
    * href="http://paranamer.codehaus.org/">Paranamer</a> is run over compiled
    * interface declarations, since Java 6 reflection does not provide access to
-   * method parameter names.  See Avro's build.xml for an example. </p>
-   */
+   * method parameter names.  See Avro's build.xml for an example. */
+  @Override
   public Protocol getProtocol(Class iface) {
     Protocol protocol =
       new Protocol(iface.getSimpleName(), iface.getPackage().getName()); 
@@ -346,4 +202,10 @@
     return protocol.createMessage(method.getName(), request, response, errors);
   }
 
+  @Override
+  public int compare(Object o1, Object o2, Schema s) {
+    throw new UnsupportedOperationException();
+  }
+
+
 }

Modified: hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectDatumReader.java
URL: http://svn.apache.org/viewvc/hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectDatumReader.java?rev=835165&r1=835164&r2=835165&view=diff
==============================================================================
--- hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectDatumReader.java (original)
+++ hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectDatumReader.java Wed Nov 11 23:53:00 2009
@@ -17,19 +17,15 @@
  */
 package org.apache.avro.reflect;
 
-import java.lang.reflect.Constructor;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
 import org.apache.avro.AvroRuntimeException;
 import org.apache.avro.Schema;
-import org.apache.avro.generic.GenericDatumReader;
+import org.apache.avro.specific.SpecificDatumReader;
 
 /**
  * {@link org.apache.avro.io.DatumReader DatumReader} for existing classes via
  * Java reflection.
  */
-public class ReflectDatumReader extends GenericDatumReader<Object> {
+public class ReflectDatumReader extends SpecificDatumReader {
   public ReflectDatumReader() {}
 
   public ReflectDatumReader(Class c) {
@@ -37,14 +33,10 @@
   }
 
   public ReflectDatumReader(Schema root) {
-    setSchema(root);
-  }
-
-  protected Object newRecord(Object old, Schema schema) {
-    Class c = ReflectData.get().getClass(schema);
-    return (c.isInstance(old) ? old : newInstance(c));
+    super(root);
   }
 
+  @Override
   protected void addField(Object record, String name, int position, Object o) {
     try {
       ReflectData.getField(record.getClass(), name).set(record, o);
@@ -53,6 +45,7 @@
     }
   }
 
+  @Override
   protected Object getField(Object record, String name, int position) {
     try {
       return ReflectData.getField(record.getClass(), name).get(record);
@@ -61,41 +54,10 @@
     }
   }
 
+  @Override
   protected void removeField(Object record, String name, int position) {
     addField(record, name, position, null);
   }
 
-  @SuppressWarnings("unchecked")
-  protected Object createEnum(String symbol, Schema schema) {
-    return Enum.valueOf(ReflectData.get().getClass(schema), symbol);
-  }
-
-  protected Object createFixed(Object old, Schema schema) {
-    Class c = ReflectData.get().getClass(schema);
-    return c.isInstance(old) ? old : newInstance(c);
-  }
-
-  private static final Class<?>[] EMPTY_ARRAY = new Class[]{};
-  private static final Map<Class,Constructor> CTOR_CACHE =
-    new ConcurrentHashMap<Class,Constructor>();
-
-  /** Create a new instance of the named class. */
-  @SuppressWarnings("unchecked")
-  protected static Object newInstance(Class c) {
-    Object result;
-    try {
-      Constructor meth = (Constructor)CTOR_CACHE.get(c);
-      if (meth == null) {
-        meth = c.getDeclaredConstructor(EMPTY_ARRAY);
-        meth.setAccessible(true);
-        CTOR_CACHE.put(c, meth);
-      }
-      result = meth.newInstance();
-    } catch (Exception e) {
-      throw new RuntimeException(e);
-    }
-    return result;
-  }
-
 }
 

Modified: hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectDatumWriter.java
URL: http://svn.apache.org/viewvc/hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectDatumWriter.java?rev=835165&r1=835164&r2=835165&view=diff
==============================================================================
--- hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectDatumWriter.java (original)
+++ hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectDatumWriter.java Wed Nov 11 23:53:00 2009
@@ -17,18 +17,15 @@
  */
 package org.apache.avro.reflect;
 
-import java.io.IOException;
-
 import org.apache.avro.AvroRuntimeException;
 import org.apache.avro.Schema;
-import org.apache.avro.generic.GenericDatumWriter;
-import org.apache.avro.io.Encoder;
+import org.apache.avro.specific.SpecificDatumWriter;
 
 /**
  * {@link org.apache.avro.io.DatumWriter DatumWriter} for existing classes
  * via Java reflection.
  */
-public class ReflectDatumWriter extends GenericDatumWriter<Object> {
+public class ReflectDatumWriter extends SpecificDatumWriter {
   public ReflectDatumWriter() {
     this(ReflectData.get());
   }
@@ -45,14 +42,15 @@
     this(root, ReflectData.get());
   }
 
-  public ReflectDatumWriter(Schema root, ReflectData reflectData) {
+  protected ReflectDatumWriter(Schema root, ReflectData reflectData) {
     super(root, reflectData);
   }
   
-  public ReflectDatumWriter(ReflectData reflectData) {
+  protected ReflectDatumWriter(ReflectData reflectData) {
     super(reflectData);
   }
   
+  @Override
   protected Object getField(Object record, String name, int position) {
     try {
       return ReflectData.getField(record.getClass(), name).get(record);
@@ -61,14 +59,5 @@
     }
   }
   
-  protected void writeEnum(Schema schema, Object datum, Encoder out)
-    throws IOException {
-    out.writeEnum(((Enum)datum).ordinal());
-  }
-
-  protected boolean isEnum(Object datum) {
-    return datum instanceof Enum;
-  }
-
 }
 

Modified: hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectRequestor.java
URL: http://svn.apache.org/viewvc/hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectRequestor.java?rev=835165&r1=835164&r2=835165&view=diff
==============================================================================
--- hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectRequestor.java (original)
+++ hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectRequestor.java Wed Nov 11 23:53:00 2009
@@ -19,74 +19,38 @@
 package org.apache.avro.reflect;
 
 import java.io.IOException;
-import java.lang.reflect.InvocationHandler;
-import java.lang.reflect.Method;
 import java.lang.reflect.Proxy;
-import java.util.Map;
 
 import org.apache.avro.Protocol;
 import org.apache.avro.Schema;
 import org.apache.avro.io.DatumReader;
 import org.apache.avro.io.DatumWriter;
-import org.apache.avro.io.Decoder;
-import org.apache.avro.io.Encoder;
-import org.apache.avro.ipc.AvroRemoteException;
-import org.apache.avro.ipc.Requestor;
+import org.apache.avro.specific.SpecificRequestor;
 import org.apache.avro.ipc.Transceiver;
 
-/** A {@link Requestor} for existing interfaces via Java reflection. */
-public class ReflectRequestor extends Requestor implements InvocationHandler {
+/** A {@link org.apache.avro.ipc.Requestor} for existing interfaces. */
+public class ReflectRequestor extends SpecificRequestor {
   
   public ReflectRequestor(Class<?> iface, Transceiver transceiver)
     throws IOException {
-    this(iface, transceiver, ReflectData.get());
+    this(ReflectData.get().getProtocol(iface), transceiver);
   }
 
   protected ReflectRequestor(Protocol protocol, Transceiver transceiver)
     throws IOException {
-    this(protocol, transceiver, ReflectData.get());
-  }
-
-  public ReflectRequestor(Class<?> iface, Transceiver transceiver, ReflectData reflectData)
-    throws IOException {
-    this(reflectData.getProtocol(iface), transceiver, reflectData);
-  }
-
-  protected ReflectRequestor(Protocol protocol, Transceiver transceiver, ReflectData reflectData)
-    throws IOException {
     super(protocol, transceiver);
   }
-
-  public Object invoke(Object proxy, Method method, Object[] args)
-    throws Throwable {
-    return request(method.getName(), args);
-  }
     
+  @Override
   protected DatumWriter<Object> getDatumWriter(Schema schema) {
     return new ReflectDatumWriter(schema);
   }
 
+  @Override
   protected DatumReader<Object> getDatumReader(Schema schema) {
     return new ReflectDatumReader(schema);
   }
 
-  public void writeRequest(Schema schema, Object request, Encoder out)
-    throws IOException {
-    Object[] args = (Object[])request;
-    int i = 0;
-    for (Map.Entry<String, Schema> param : schema.getFieldSchemas())
-      getDatumWriter(param.getValue()).write(args[i++], out);
-  }
-    
-  public Object readResponse(Schema schema, Decoder in) throws IOException {
-    return getDatumReader(schema).read(null, in);
-  }
-
-  public AvroRemoteException readError(Schema schema, Decoder in)
-    throws IOException {
-    return (AvroRemoteException)getDatumReader(schema).read(null, in);
-  }
-
   /** Create a proxy instance whose methods invoke RPCs. */
   public static Object getClient(Class<?> iface, Transceiver transciever)
     throws IOException {

Modified: hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectResponder.java
URL: http://svn.apache.org/viewvc/hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectResponder.java?rev=835165&r1=835164&r2=835165&view=diff
==============================================================================
--- hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectResponder.java (original)
+++ hadoop/avro/trunk/src/java/org/apache/avro/reflect/ReflectResponder.java Wed Nov 11 23:53:00 2009
@@ -18,83 +18,31 @@
 
 package org.apache.avro.reflect;
 
-import java.io.IOException;
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
-import java.util.Map;
-
-import org.apache.avro.AvroRuntimeException;
 import org.apache.avro.Schema;
-import org.apache.avro.Protocol.Message;
+import org.apache.avro.Protocol;
 import org.apache.avro.io.DatumReader;
 import org.apache.avro.io.DatumWriter;
-import org.apache.avro.io.Decoder;
-import org.apache.avro.io.Encoder;
-import org.apache.avro.ipc.AvroRemoteException;
-import org.apache.avro.ipc.Responder;
-
-/** {@link Responder} for existing interfaces via Java reflection.*/
-public class ReflectResponder extends Responder {
-  private Object impl;
+import org.apache.avro.specific.SpecificResponder;
 
+/** {@link org.apache.avro.ipc.Responder} for existing interfaces.*/
+public class ReflectResponder extends SpecificResponder {
   public ReflectResponder(Class iface, Object impl) {
-    this(iface, impl, ReflectData.get());
+    this(ReflectData.get().getProtocol(iface), impl);
   }
   
-  public ReflectResponder(Class iface, Object impl, ReflectData reflectData) {
-    super(reflectData.getProtocol(iface));
-    this.impl = impl;
+  public ReflectResponder(Protocol protocol, Object impl) {
+    super(protocol, impl);
   }
 
+  @Override
   protected DatumWriter<Object> getDatumWriter(Schema schema) {
     return new ReflectDatumWriter(schema);
   }
 
+  @Override
   protected DatumReader<Object> getDatumReader(Schema schema) {
     return new ReflectDatumReader(schema);
   }
 
-  /** Reads a request message. */
-  public Object readRequest(Schema schema, Decoder in) throws IOException {
-    Object[] args = new Object[schema.getFields().size()];
-    int i = 0;
-    for (Map.Entry<String, Schema> param : schema.getFieldSchemas())
-      args[i++] = getDatumReader(param.getValue()).read(null, in);
-    return args;
-  }
-
-  /** Writes a response message. */
-  public void writeResponse(Schema schema, Object response, Encoder out)
-    throws IOException {
-    getDatumWriter(schema).write(response, out);
-  }
-
-  /** Writes an error message. */
-  public void writeError(Schema schema, AvroRemoteException error,
-                         Encoder out) throws IOException {
-    getDatumWriter(schema).write(error, out);
-  }
-
-  public Object respond(Message message, Object request)
-    throws AvroRemoteException {
-    Class[] paramTypes = new Class[message.getRequest().getFields().size()];
-    int i = 0;
-    try {
-      for (Map.Entry<String,Schema> param: message.getRequest().getFieldSchemas())
-        paramTypes[i++] = ReflectData.get().getClass(param.getValue());
-      Method method = impl.getClass().getMethod(message.getName(), paramTypes);
-      return method.invoke(impl, (Object[])request);
-    } catch (InvocationTargetException e) {
-      Throwable target = e.getTargetException();
-      if (target instanceof AvroRemoteException)
-        throw (AvroRemoteException)target;
-      else throw new AvroRuntimeException(e);
-    } catch (NoSuchMethodException e) {
-      throw new AvroRuntimeException(e);
-    } catch (IllegalAccessException e) {
-      throw new AvroRuntimeException(e);
-    }
-  }
-
 }
 

Copied: hadoop/avro/trunk/src/java/org/apache/avro/specific/FixedSize.java (from r834652, hadoop/avro/trunk/src/java/org/apache/avro/reflect/FixedSize.java)
URL: http://svn.apache.org/viewvc/hadoop/avro/trunk/src/java/org/apache/avro/specific/FixedSize.java?p2=hadoop/avro/trunk/src/java/org/apache/avro/specific/FixedSize.java&p1=hadoop/avro/trunk/src/java/org/apache/avro/reflect/FixedSize.java&r1=834652&r2=835165&rev=835165&view=diff
==============================================================================
--- hadoop/avro/trunk/src/java/org/apache/avro/reflect/FixedSize.java (original)
+++ hadoop/avro/trunk/src/java/org/apache/avro/specific/FixedSize.java Wed Nov 11 23:53:00 2009
@@ -15,7 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.avro.reflect;
+package org.apache.avro.specific;
 
 import java.lang.annotation.Documented;
 import java.lang.annotation.ElementType;

Modified: hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificCompiler.java
URL: http://svn.apache.org/viewvc/hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificCompiler.java?rev=835165&r1=835164&r2=835165&view=diff
==============================================================================
--- hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificCompiler.java (original)
+++ hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificCompiler.java Wed Nov 11 23:53:00 2009
@@ -151,8 +151,8 @@
     StringBuilder out = new StringBuilder();
     header(out, protocol.getNamespace());
     line(out, 0, "public interface "+protocol.getName()+" {");
-
-    out.append("\n");
+    line(out, 1, "public static final Protocol _PROTOCOL = Protocol.parse(\""
+           +esc(protocol)+"\");");
     for (Map.Entry<String,Message> e : protocol.getMessages().entrySet()) {
       String name = e.getKey();
       Message message = e.getValue();
@@ -193,7 +193,7 @@
     line(out, 0, "import org.apache.avro.specific.SpecificRecordBase;");
     line(out, 0, "import org.apache.avro.specific.SpecificRecord;");
     line(out, 0, "import org.apache.avro.specific.SpecificFixed;");
-    line(out, 0, "import org.apache.avro.reflect.FixedSize;");
+    line(out, 0, "import org.apache.avro.specific.FixedSize;");
     for (Schema s : queue)
       if (namespace == null
           ? (s.getNamespace() != null)

Modified: hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificData.java
URL: http://svn.apache.org/viewvc/hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificData.java?rev=835165&r1=835164&r2=835165&view=diff
==============================================================================
--- hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificData.java (original)
+++ hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificData.java Wed Nov 11 23:53:00 2009
@@ -19,14 +19,24 @@
 
 import java.util.Iterator;
 import java.util.Map;
+import java.util.WeakHashMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.LinkedHashMap;
+import java.nio.ByteBuffer;
+import java.lang.reflect.ParameterizedType;
 
 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.Field;
-import org.apache.avro.reflect.ReflectData;
+import org.apache.avro.Schema.Type;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericArray;
+import org.apache.avro.util.Utf8;
 
 /** Utilities for generated Java classes and interfaces. */
-public class SpecificData extends ReflectData {
+public class SpecificData extends GenericData {
 
   private static final SpecificData INSTANCE = new SpecificData();
 
@@ -36,36 +46,165 @@
   public static SpecificData get() { return INSTANCE; }
 
   @Override
-  protected Schema createSchema(java.lang.reflect.Type type,
-                             Map<String,Schema> names) {
-    if (type instanceof Class) {
-      Class c = (Class)type;
-      String name = c.getSimpleName();
-      Schema schema = names.get(name);
-      if (schema != null) return schema;
-      if (SpecificRecord.class.isAssignableFrom(c)) {
+  protected boolean isRecord(Object datum) {
+    return datum instanceof SpecificRecord;
+  }
+
+  @Override
+  protected Schema getRecordSchema(Object record) {
+    return ((SpecificRecord)record).getSchema();
+  }
+
+  @Override
+  protected boolean isEnum(Object datum) {
+    return datum instanceof Enum;
+  }
+
+  @Override
+  public boolean validate(Schema schema, Object datum) {
+    switch (schema.getType()) {
+    case RECORD:
+      Class c = datum.getClass(); 
+      if (!(datum instanceof SpecificRecord)) return false;
+      SpecificRecord record = (SpecificRecord)datum;
+      Iterator<Field> fields = schema.getFields().values().iterator();
+      for (int i = 0; fields.hasNext(); i++)
+        if (!validate(fields.next().schema(), record.get(i)))
+          return false;
+      return true;
+    case ENUM:
+      return datum instanceof Enum
+        && schema.getEnumSymbols().contains(((Enum)datum).name());
+    default:
+      return super.validate(schema, datum);
+    }
+  }
+
+  private Map<String,Class> classCache = new ConcurrentHashMap<String,Class>();
+
+  /** Return the class that implements a schema. */
+  public Class getClass(Schema schema) {
+    switch (schema.getType()) {
+    case FIXED:
+    case RECORD:
+    case ENUM:
+      String name = schema.getFullName();
+      Class c = classCache.get(name);
+      if (c == null) {
         try {
-          schema = (Schema)((Class)type).getDeclaredField("_SCHEMA").get(null);
-        } catch (NoSuchFieldException e) {
-          throw new AvroRuntimeException(e);
-        } catch (IllegalAccessException e) {
+          c = Class.forName(getClassName(schema));
+          classCache.put(name, c);
+        } catch (ClassNotFoundException e) {
           throw new AvroRuntimeException(e);
         }
-        names.put(name, schema);
-        return schema;
       }
+      return c;
+    case ARRAY:   return GenericArray.class;
+    case MAP:     return Map.class;
+    case UNION:   return Object.class;
+    case STRING:  return Utf8.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);
     }
-    return super.createSchema(type, names);
   }
 
-  @Override
-  protected boolean isRecord(Object datum) {
-    return datum instanceof SpecificRecord;
+  /** Returns the Java class name indicated by a schema's name and namespace. */
+  public String getClassName(Schema schema) {
+    String namespace = schema.getNamespace();
+    String name = schema.getName();
+    if (namespace == null)
+      return name;
+    String dot = namespace.endsWith("$") ? "" : ".";
+    return namespace + dot + name;
   }
 
-  @Override
-  protected Schema getRecordSchema(Object record) {
-    return ((SpecificRecord)record).getSchema();
+  private final WeakHashMap<java.lang.reflect.Type,Schema> schemaCache =
+    new WeakHashMap<java.lang.reflect.Type,Schema>();
+
+  /** Find the schema for a Java type. */
+  public Schema getSchema(java.lang.reflect.Type type) {
+    Schema schema = schemaCache.get(type);
+    if (schema == null) {
+      schema = createSchema(type, new LinkedHashMap<String,Schema>());
+      schemaCache.put(type, schema);
+    }
+    return schema;
+  }
+
+  /** Create the schema for a Java type. */
+  @SuppressWarnings(value="unchecked")
+  protected Schema createSchema(java.lang.reflect.Type type,
+                                Map<String,Schema> names) {
+    if (type == Utf8.class)
+      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();
+      for (int i = 0; i < params.length; i++)
+      if (GenericArray.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 == Utf8.class))
+          throw new AvroTypeException("Map key class not Utf8: "+key);
+        return Schema.createMap(createSchema(value, names));
+      }
+    } else if (type instanceof Class) {               // class
+      Class c = (Class)type;
+      String fullName = c.getName();
+      Schema schema = names.get(fullName);
+      if (schema == null)
+        schema = createClassSchema(c, names);
+      names.put(fullName, schema);
+      return schema;
+    }
+    throw new AvroTypeException("Unknown type: "+type);
+  }
+
+  /** Create a schema for a Java class. */
+  protected Schema createClassSchema(Class c, Map<String,Schema> names) {
+    try {
+      return (Schema)(c.getDeclaredField("_SCHEMA").get(null));
+    } catch (NoSuchFieldException e) {
+      throw new AvroRuntimeException(e);
+    } catch (IllegalAccessException e) {
+      throw new AvroRuntimeException(e);
+    }
+  }
+
+  /** Return the protocol for a Java interface. */
+  public Protocol getProtocol(Class iface) {
+    try {
+      return (Protocol)(iface.getDeclaredField("_PROTOCOL").get(null));
+    } catch (NoSuchFieldException e) {
+      throw new AvroRuntimeException(e);
+    } catch (IllegalAccessException e) {
+      throw new AvroRuntimeException(e);
+    }
   }
 
   @Override

Modified: hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificDatumReader.java
URL: http://svn.apache.org/viewvc/hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificDatumReader.java?rev=835165&r1=835164&r2=835165&view=diff
==============================================================================
--- hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificDatumReader.java (original)
+++ hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificDatumReader.java Wed Nov 11 23:53:00 2009
@@ -17,11 +17,15 @@
  */
 package org.apache.avro.specific;
 
+import java.lang.reflect.Constructor;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
 import org.apache.avro.Schema;
-import org.apache.avro.reflect.ReflectDatumReader;
+import org.apache.avro.generic.GenericDatumReader;
 
 /** {@link org.apache.avro.io.DatumReader DatumReader} for generated Java classes. */
-public class SpecificDatumReader extends ReflectDatumReader {
+public class SpecificDatumReader extends GenericDatumReader<Object> {
   public SpecificDatumReader() {}
 
   public SpecificDatumReader(Class c) {
@@ -32,15 +36,57 @@
     super(schema);
   }
 
+  @Override
+  protected Object newRecord(Object old, Schema schema) {
+    Class c = SpecificData.get().getClass(schema);
+    return (c.isInstance(old) ? old : newInstance(c));
+  }
+
+  @Override
   protected void addField(Object record, String name, int position, Object o) {
     ((SpecificRecord)record).set(position, o);
   }
+  @Override
   protected Object getField(Object record, String name, int position) {
     return ((SpecificRecord)record).get(position);
   }
+  @Override
   protected void removeField(Object record, String field, int position) {
     ((SpecificRecord)record).set(position, null);
   }
 
+  @Override
+  @SuppressWarnings("unchecked")
+  protected Object createEnum(String symbol, Schema schema) {
+    return Enum.valueOf(SpecificData.get().getClass(schema), symbol);
+  }
+
+  @Override
+  protected Object createFixed(Object old, Schema schema) {
+    Class c = SpecificData.get().getClass(schema);
+    return c.isInstance(old) ? old : newInstance(c);
+  }
+
+  private static final Class<?>[] EMPTY_ARRAY = new Class[]{};
+  private static final Map<Class,Constructor> CTOR_CACHE =
+    new ConcurrentHashMap<Class,Constructor>();
+
+  @SuppressWarnings("unchecked")
+  private static Object newInstance(Class c) {
+    Object result;
+    try {
+      Constructor meth = (Constructor)CTOR_CACHE.get(c);
+      if (meth == null) {
+        meth = c.getDeclaredConstructor(EMPTY_ARRAY);
+        meth.setAccessible(true);
+        CTOR_CACHE.put(c, meth);
+      }
+      result = meth.newInstance();
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+    return result;
+  }
+
 }
 

Modified: hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificDatumWriter.java
URL: http://svn.apache.org/viewvc/hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificDatumWriter.java?rev=835165&r1=835164&r2=835165&view=diff
==============================================================================
--- hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificDatumWriter.java (original)
+++ hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificDatumWriter.java Wed Nov 11 23:53:00 2009
@@ -17,11 +17,14 @@
  */
 package org.apache.avro.specific;
 
+import java.io.IOException;
+
 import org.apache.avro.Schema;
-import org.apache.avro.reflect.ReflectDatumWriter;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.io.Encoder;
 
 /** {@link org.apache.avro.io.DatumWriter DatumWriter} for generated Java classes. */
-public class SpecificDatumWriter extends ReflectDatumWriter {
+public class SpecificDatumWriter extends GenericDatumWriter<Object> {
   public SpecificDatumWriter() {}
 
   public SpecificDatumWriter(Class c) {
@@ -32,9 +35,24 @@
     super(schema, SpecificData.get());
   }
   
+  protected SpecificDatumWriter(Schema root, SpecificData specificData) {
+    super(root, specificData);
+  }
+  
+  protected SpecificDatumWriter(SpecificData specificData) {
+    super(specificData);
+  }
+  
+  @Override
   protected Object getField(Object record, String name, int position) {
     return ((SpecificRecord)record).get(position);
   }
 
+  @Override
+  protected void writeEnum(Schema schema, Object datum, Encoder out)
+    throws IOException {
+    out.writeEnum(((Enum)datum).ordinal());
+  }
+
 }
 

Modified: hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificFixed.java
URL: http://svn.apache.org/viewvc/hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificFixed.java?rev=835165&r1=835164&r2=835165&view=diff
==============================================================================
--- hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificFixed.java (original)
+++ hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificFixed.java Wed Nov 11 23:53:00 2009
@@ -18,7 +18,6 @@
 package org.apache.avro.specific;
 
 import org.apache.avro.generic.GenericData;
-import org.apache.avro.reflect.FixedSize;
 
 /** Base class for generated fixed-sized data classes. */
 public abstract class SpecificFixed extends GenericData.Fixed {

Modified: hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificRequestor.java
URL: http://svn.apache.org/viewvc/hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificRequestor.java?rev=835165&r1=835164&r2=835165&view=diff
==============================================================================
--- hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificRequestor.java (original)
+++ hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificRequestor.java Wed Nov 11 23:53:00 2009
@@ -19,34 +19,40 @@
 package org.apache.avro.specific;
 
 import java.io.IOException;
+import java.lang.reflect.Method;
 import java.lang.reflect.Proxy;
+import java.lang.reflect.InvocationHandler;
+import java.util.Map;
 
 import org.apache.avro.Protocol;
 import org.apache.avro.Schema;
 import org.apache.avro.io.DatumReader;
 import org.apache.avro.io.DatumWriter;
+import org.apache.avro.io.Decoder;
+import org.apache.avro.io.Encoder;
+import org.apache.avro.ipc.AvroRemoteException;
 import org.apache.avro.ipc.Transceiver;
-import org.apache.avro.reflect.ReflectRequestor;
+import org.apache.avro.ipc.Requestor;
 
 /** {@link org.apache.avro.ipc.Requestor Requestor} for generated interfaces. */
-public class SpecificRequestor extends ReflectRequestor {
+public class SpecificRequestor extends Requestor implements InvocationHandler {
   
   public SpecificRequestor(Class<?> iface, Transceiver transceiver)
     throws IOException {
-    this(iface, transceiver, SpecificData.get());
+    this(SpecificData.get().getProtocol(iface), transceiver);
   }
 
-  public SpecificRequestor(Class<?> iface, Transceiver transceiver,
-                           SpecificData specificData)
-    throws IOException {
-    this(specificData.getProtocol(iface), transceiver);
-  }
-  
-  private SpecificRequestor(Protocol protocol, Transceiver transceiver)
+  protected SpecificRequestor(Protocol protocol, Transceiver transceiver)
     throws IOException {
     super(protocol, transceiver);
   }
-    
+
+  @Override
+  public Object invoke(Object proxy, Method method, Object[] args)
+    throws Throwable {
+    return request(method.getName(), args);
+  }
+
   protected DatumWriter<Object> getDatumWriter(Schema schema) {
     return new SpecificDatumWriter(schema);
   }
@@ -55,6 +61,26 @@
     return new SpecificDatumReader(schema);
   }
 
+  @Override
+  public void writeRequest(Schema schema, Object request, Encoder out)
+    throws IOException {
+    Object[] args = (Object[])request;
+    int i = 0;
+    for (Map.Entry<String, Schema> param : schema.getFieldSchemas())
+      getDatumWriter(param.getValue()).write(args[i++], out);
+  }
+    
+  @Override
+  public Object readResponse(Schema schema, Decoder in) throws IOException {
+    return getDatumReader(schema).read(null, in);
+  }
+
+  @Override
+  public AvroRemoteException readError(Schema schema, Decoder in)
+    throws IOException {
+    return (AvroRemoteException)getDatumReader(schema).read(null, in);
+  }
+
   /** Create a proxy instance whose methods invoke RPCs. */
   public static Object getClient(Class<?> iface, Transceiver transciever)
     throws IOException {

Modified: hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificResponder.java
URL: http://svn.apache.org/viewvc/hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificResponder.java?rev=835165&r1=835164&r2=835165&view=diff
==============================================================================
--- hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificResponder.java (original)
+++ hadoop/avro/trunk/src/java/org/apache/avro/specific/SpecificResponder.java Wed Nov 11 23:53:00 2009
@@ -18,17 +18,35 @@
 
 package org.apache.avro.specific;
 
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.Map;
+
 import org.apache.avro.Schema;
+import org.apache.avro.Protocol;
+import org.apache.avro.Protocol.Message;
+import org.apache.avro.AvroRuntimeException;
 import org.apache.avro.io.DatumReader;
 import org.apache.avro.io.DatumWriter;
-import org.apache.avro.reflect.ReflectResponder;
+import org.apache.avro.io.Decoder;
+import org.apache.avro.io.Encoder;
+import org.apache.avro.ipc.AvroRemoteException;
+import org.apache.avro.ipc.Responder;
 
 /** {@link org.apache.avro.ipc.Responder Responder} for generated interfaces.*/
-public class SpecificResponder extends ReflectResponder {
+public class SpecificResponder extends Responder {
+  private Object impl;
+
   public SpecificResponder(Class iface, Object impl) {
-    super(iface, impl);
+    this(SpecificData.get().getProtocol(iface), impl);
   }
     
+  public SpecificResponder(Protocol protocol, Object impl) {
+    super(protocol);
+    this.impl = impl;
+  }
+
   protected DatumWriter<Object> getDatumWriter(Schema schema) {
     return new SpecificDatumWriter(schema);
   }
@@ -37,5 +55,48 @@
     return new SpecificDatumReader(schema);
   }
 
+  @Override
+  public Object readRequest(Schema schema, Decoder in) throws IOException {
+    Object[] args = new Object[schema.getFields().size()];
+    int i = 0;
+    for (Map.Entry<String, Schema> param : schema.getFieldSchemas())
+      args[i++] = getDatumReader(param.getValue()).read(null, in);
+    return args;
+  }
+
+  @Override
+  public void writeResponse(Schema schema, Object response, Encoder out)
+    throws IOException {
+    getDatumWriter(schema).write(response, out);
+  }
+
+  @Override
+  public void writeError(Schema schema, AvroRemoteException error,
+                         Encoder out) throws IOException {
+    getDatumWriter(schema).write(error, out);
+  }
+
+  @Override
+  public Object respond(Message message, Object request)
+    throws AvroRemoteException {
+    Class[] paramTypes = new Class[message.getRequest().getFields().size()];
+    int i = 0;
+    try {
+      for (Map.Entry<String,Schema> param: message.getRequest().getFieldSchemas())
+        paramTypes[i++] = SpecificData.get().getClass(param.getValue());
+      Method method = impl.getClass().getMethod(message.getName(), paramTypes);
+      return method.invoke(impl, (Object[])request);
+    } catch (InvocationTargetException e) {
+      Throwable target = e.getTargetException();
+      if (target instanceof AvroRemoteException)
+        throw (AvroRemoteException)target;
+      else throw new AvroRuntimeException(e);
+    } catch (NoSuchMethodException e) {
+      throw new AvroRuntimeException(e);
+    } catch (IllegalAccessException e) {
+      throw new AvroRuntimeException(e);
+    }
+  }
+
 }
 

Modified: hadoop/avro/trunk/src/test/java/org/apache/avro/TestDataFileReflect.java
URL: http://svn.apache.org/viewvc/hadoop/avro/trunk/src/test/java/org/apache/avro/TestDataFileReflect.java?rev=835165&r1=835164&r2=835165&view=diff
==============================================================================
--- hadoop/avro/trunk/src/test/java/org/apache/avro/TestDataFileReflect.java (original)
+++ hadoop/avro/trunk/src/test/java/org/apache/avro/TestDataFileReflect.java Wed Nov 11 23:53:00 2009
@@ -132,7 +132,7 @@
     ReflectData reflectData = ReflectData.AllowNull.get();
     Schema schema = reflectData.getSchema(BarRecord.class);
     DataFileWriter<Object> writer = new DataFileWriter<Object>(schema, fos,
-        new ReflectDatumWriter(schema, reflectData));
+        new ReflectDatumWriter(BarRecord.class, reflectData));
 
     // test writing to a file
     CheckList check = new CheckList();



Mime
View raw message