[13/51] [partial] incubator-reef git commit: [REEF-93] Move java sources to lang/java

[we...@apache.org]

http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/java/InjectorImpl.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/java/InjectorImpl.java b/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/java/InjectorImpl.java
new file mode 100644
index 0000000..16ce791
--- /dev/null
+++ b/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/java/InjectorImpl.java
@@ -0,0 +1,760 @@
+/**
+ * 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
+ *
+ *   http://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.reef.tang.implementation.java;
+
+import org.apache.reef.tang.*;
+import org.apache.reef.tang.annotations.Name;
+import org.apache.reef.tang.exceptions.*;
+import org.apache.reef.tang.implementation.*;
+import org.apache.reef.tang.types.*;
+import org.apache.reef.tang.util.MonotonicHashSet;
+import org.apache.reef.tang.util.MonotonicSet;
+import org.apache.reef.tang.util.ReflectionUtilities;
+import org.apache.reef.tang.util.TracingMonotonicTreeMap;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.*;
+
+public class InjectorImpl implements Injector {
+  static final InjectionPlan<?> BUILDING = new InjectionPlan<Object>(null) {
+    @Override
+    public int getNumAlternatives() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public String toString() {
+      return "BUILDING INJECTION PLAN";
+    }
+
+    @Override
+    public boolean isAmbiguous() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public boolean isInjectable() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    protected String toAmbiguousInjectString() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    protected String toInfeasibleInjectString() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    protected boolean isInfeasibleLeaf() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public String toShallowString() {
+      throw new UnsupportedOperationException();
+    }
+
+  };
+  final Map<ClassNode<?>, Object> instances = new TracingMonotonicTreeMap<>();
+  final Map<NamedParameterNode<?>, Object> namedParameterInstances = new TracingMonotonicTreeMap<>();
+  private final Configuration c;
+  private final ClassHierarchy namespace;
+  private final JavaClassHierarchy javaNamespace;
+  private final Set<InjectionFuture<?>> pendingFutures = new HashSet<>();
+  private boolean concurrentModificationGuard = false;
+  private Aspect aspect;
+
+  public InjectorImpl(Configuration c) throws BindException {
+    this.c = c;
+    this.namespace = c.getClassHierarchy();
+    this.javaNamespace = (ClassHierarchyImpl) this.namespace;
+  }
+
+  private static InjectorImpl copy(InjectorImpl old,
+                                   Configuration... configurations) throws BindException {
+    final InjectorImpl i;
+    try {
+      final ConfigurationBuilder cb = old.c.newBuilder();
+      for (Configuration c : configurations) {
+        cb.addConfiguration(c);
+      }
+      i = new InjectorImpl(cb.build());
+    } catch (BindException e) {
+      throw new IllegalStateException(
+          "Unexpected error copying configuration!", e);
+    }
+    for (ClassNode<?> cn : old.instances.keySet()) {
+      if (cn.getFullName().equals(ReflectionUtilities.getFullName(Injector.class))
+          || cn.getFullName().equals(ReflectionUtilities.getFullName(InjectorImpl.class))) {
+        // This would imply that we're treating injector as a singleton somewhere.  It should be copied fresh each time.
+        throw new IllegalStateException();
+      }
+      try {
+        ClassNode<?> new_cn = (ClassNode<?>) i.namespace.getNode(cn
+            .getFullName());
+        i.instances.put(new_cn, old.instances.get(cn));
+      } catch (BindException e) {
+        throw new IllegalStateException("Could not resolve name "
+            + cn.getFullName() + " when copying injector");
+      }
+    }
+    // Copy references to the remaining (which must have been set with
+    // bindVolatileParameter())
+    for (NamedParameterNode<?> np : old.namedParameterInstances.keySet()) {
+      // if (!builder.namedParameters.containsKey(np)) {
+      Object o = old.namedParameterInstances.get(np);
+      NamedParameterNode<?> new_np = (NamedParameterNode<?>) i.namespace
+          .getNode(np.getFullName());
+      i.namedParameterInstances.put(new_np, o);
+    }
+    // Fork the aspect (if any)
+    if (old.aspect != null) {
+      i.bindAspect(old.aspect.createChildAspect());
+    }
+    return i;
+  }
+
+  private void assertNotConcurrent() {
+    if (concurrentModificationGuard) {
+      throw new ConcurrentModificationException("Detected attempt to use Injector from within an injected constructor!");
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private <T> T getCachedInstance(ClassNode<T> cn) {
+    if (cn.getFullName().equals("org.apache.reef.tang.Injector")) {
+      return (T) this;// TODO: We should be insisting on injection futures here! .forkInjector();
+    } else {
+      T t = (T) instances.get(cn);
+      if (t instanceof InjectionFuture) {
+        throw new IllegalStateException("Found an injection future in getCachedInstance: " + cn);
+      }
+      return t;
+    }
+  }
+
+  /**
+   * Produce a list of "interesting" constructors from a set of ClassNodes.
+   * <p/>
+   * Tang Constructors expose a isMoreSpecificThan function that embeds all
+   * a skyline query over the lattices.  Precisely:
+   * <p/>
+   * Let candidateConstructors be the union of all constructors defined by
+   * ClassNodes in candidateImplementations.
+   * <p/>
+   * This function returns a set called filteredImplementations, defined as
+   * follows:
+   * <p/>
+   * For each member f of filteredConstructors, there does not exist
+   * a g in candidateConstructors s.t. g.isMoreSpecificThan(f).
+   */
+  private <T> List<InjectionPlan<T>> filterCandidateConstructors(
+      final List<ClassNode<T>> candidateImplementations,
+      final Map<Node, InjectionPlan<?>> memo) {
+
+    final List<InjectionPlan<T>> sub_ips = new ArrayList<>();
+    for (final ClassNode<T> thisCN : candidateImplementations) {
+      final List<Constructor<T>> constructors = new ArrayList<>();
+      final List<ConstructorDef<T>> constructorList = new ArrayList<>();
+      if (null != c.getLegacyConstructor(thisCN)) {
+        constructorList.add(c.getLegacyConstructor(thisCN));
+      }
+      constructorList
+          .addAll(Arrays.asList(thisCN.getInjectableConstructors()));
+
+      for (ConstructorDef<T> def : constructorList) {
+        final List<InjectionPlan<?>> args = new ArrayList<InjectionPlan<?>>();
+        final ConstructorArg[] defArgs = def.getArgs();
+
+        for (final ConstructorArg arg : defArgs) {
+          if (!arg.isInjectionFuture()) {
+            try {
+              final Node argNode = namespace.getNode(arg.getName());
+              buildInjectionPlan(argNode, memo);
+              args.add(memo.get(argNode));
+            } catch (NameResolutionException e) {
+              throw new IllegalStateException("Detected unresolvable "
+                  + "constructor arg while building injection plan.  "
+                  + "This should have been caught earlier!", e);
+            }
+          } else {
+            try {
+              args.add(new InjectionFuturePlan<>(namespace.getNode(arg
+                  .getName())));
+            } catch (NameResolutionException e) {
+              throw new IllegalStateException("Detected unresolvable "
+                  + "constructor arg while building injection plan.  "
+                  + "This should have been caught earlier!", e);
+            }
+          }
+        }
+        final Constructor<T> constructor = new Constructor<T>(thisCN, def,
+            args.toArray(new InjectionPlan[0]));
+        constructors.add(constructor);
+      }
+      // The constructors are embedded in a lattice defined by
+      // isMoreSpecificThan().  We want to see if, amongst the injectable
+      // plans, there is a unique dominant plan, and select it.
+
+      // First, compute the set of injectable plans.
+      final List<Integer> liveIndices = new ArrayList<>();
+      for (int i = 0; i < constructors.size(); i++) {
+        if (constructors.get(i).getNumAlternatives() > 0) {
+          liveIndices.add(i);
+        }
+      }
+      // Now, do an all-by-all comparison, removing indices that are dominated
+      // by others.
+      for (int i = 0; i < liveIndices.size(); i++) {
+        for (int j = i + 1; j < liveIndices.size(); j++) {
+          final ConstructorDef<T> ci = constructors.get(liveIndices.get(i))
+              .getConstructorDef();
+          final ConstructorDef<T> cj = constructors.get(liveIndices.get(j))
+              .getConstructorDef();
+
+          if (ci.isMoreSpecificThan(cj)) {
+            liveIndices.remove(j);
+            j--;
+          } else if (cj.isMoreSpecificThan(ci)) {
+            liveIndices.remove(i);
+            // Done with this inner loop invocation. Check the new ci.
+            i--;
+            break;
+          }
+        }
+      }
+      if (constructors.size() > 0) {
+        sub_ips.add(wrapInjectionPlans(thisCN, constructors, false,
+            liveIndices.size() == 1 ? liveIndices.get(0) : -1));
+      }
+    }
+    return sub_ips;
+  }
+
+  @SuppressWarnings("unchecked")
+  private <T> InjectionPlan<T> buildClassNodeInjectionPlan(ClassNode<T> cn,
+                                                           T cachedInstance,
+                                                           ClassNode<ExternalConstructor<T>> externalConstructor,
+                                                           ClassNode<T> boundImpl,
+                                                           ClassNode<T> defaultImpl,
+                                                           Map<Node, InjectionPlan<?>> memo) {
+
+    if (cachedInstance != null) {
+      return new JavaInstance<T>(cn, cachedInstance);
+    } else if (externalConstructor != null) {
+      buildInjectionPlan(externalConstructor, memo);
+      return new Subplan<>(cn, 0, (InjectionPlan<T>) memo.get(externalConstructor));
+    } else if (boundImpl != null && !cn.equals(boundImpl)) {
+      // We need to delegate to boundImpl, so recurse.
+      buildInjectionPlan(boundImpl, memo);
+      return new Subplan<>(cn, 0, (InjectionPlan<T>) memo.get(boundImpl));
+    } else if (defaultImpl != null && !cn.equals(defaultImpl)) {
+      buildInjectionPlan(defaultImpl, memo);
+      return new Subplan<>(cn, 0, (InjectionPlan<T>) memo.get(defaultImpl));
+    } else {
+      // if we're here and there isn't a bound impl or a default impl,
+      // then we're bound / defaulted to ourselves, so don't add
+      // other impls to the list of things to consider.
+      final List<ClassNode<T>> candidateImplementations = new ArrayList<>();
+      candidateImplementations.add(cn);
+      final List<InjectionPlan<T>> sub_ips = filterCandidateConstructors(candidateImplementations, memo);
+      if (sub_ips.size() == 1) {
+        return wrapInjectionPlans(cn, sub_ips, false, -1);
+      } else {
+        return wrapInjectionPlans(cn, sub_ips, true, -1);
+      }
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private <T> InjectionPlan<T> wrapInjectionPlans(ClassNode<T> infeasibleNode,
+                                                  List<? extends InjectionPlan<T>> list, boolean forceAmbiguous, int selectedIndex) {
+    if (list.size() == 0) {
+      return new Subplan<>(infeasibleNode);
+    } else if ((!forceAmbiguous) && list.size() == 1) {
+      return list.get(0);
+    } else {
+      return new Subplan<>(infeasibleNode, selectedIndex, list.toArray(new InjectionPlan[0]));
+    }
+  }
+
+  /**
+   * Parse the bound value of np.  When possible, this returns a cached instance.
+   *
+   * @return null if np has not been bound.
+   * @throws ParseException
+   */
+  @SuppressWarnings("unchecked")
+  private <T> T parseBoundNamedParameter(NamedParameterNode<T> np) {
+    final T ret;
+
+    @SuppressWarnings("rawtypes")
+    final Set<Object> boundSet = c.getBoundSet((NamedParameterNode) np);
+    if (!boundSet.isEmpty()) {
+      Set<T> ret2 = new MonotonicSet<>();
+      for (Object o : boundSet) {
+        if (o instanceof String) {
+          try {
+            ret2.add(javaNamespace.parse(np, (String) o));
+          } catch (ParseException e) {
+            // Parsability is now pre-checked in bindSet, so it should not be reached!
+            throw new IllegalStateException("Could not parse " + o + " which was passed into " + np + " FIXME: Parsability is not currently checked by bindSetEntry(Node,String)");
+          }
+        } else if (o instanceof Node) {
+          ret2.add((T) o);
+        } else {
+          throw new IllegalStateException("Unexpected object " + o + " in bound set.  Should consist of nodes and strings");
+        }
+      }
+      return (T) ret2;
+    }
+    final List<Object> boundList = c.getBoundList((NamedParameterNode) np);
+    if (boundList != null) {
+      List<T> ret2 = new ArrayList<>();
+      for (Object o : boundList) {
+        if (o instanceof String) {
+          try {
+            ret2.add(javaNamespace.parse(np, (String) o));
+          } catch (ParseException e) {
+            // Parsability is now pre-checked in bindList, so it should not be reached!
+            throw new IllegalStateException("Could not parse " + o + " which was passed into " + np + " FIXME: " +
+                "Parsability is not currently checked by bindList(Node,List)");
+          }
+        } else if (o instanceof Node) {
+          ret2.add((T) o);
+        } else {
+          throw new IllegalStateException("Unexpected object " + o + " in bound list.  Should consist of nodes and " +
+              "strings");
+        }
+      }
+      return (T) ret2;
+    } else if (namedParameterInstances.containsKey(np)) {
+      ret = (T) namedParameterInstances.get(np);
+    } else {
+      final String value = c.getNamedParameter(np);
+      if (value == null) {
+        ret = null;
+      } else {
+        try {
+          ret = javaNamespace.parse(np, value);
+          namedParameterInstances.put(np, ret);
+        } catch (BindException e) {
+          throw new IllegalStateException(
+              "Could not parse pre-validated value", e);
+        }
+      }
+    }
+    return ret;
+  }
+
+  @SuppressWarnings("unchecked")
+  private <T> ClassNode<T> parseDefaultImplementation(ClassNode<T> cn) {
+    if (cn.getDefaultImplementation() != null) {
+      try {
+        return (ClassNode<T>) javaNamespace.getNode(cn.getDefaultImplementation());
+      } catch (ClassCastException | NameResolutionException e) {
+        throw new IllegalStateException("After validation, " + cn + " had a bad default implementation named " + cn.getDefaultImplementation(), e);
+      }
+    } else {
+      return null;
+    }
+  }
+
+  @SuppressWarnings({"unchecked"})
+  private <T> void buildInjectionPlan(final Node n,
+                                      Map<Node, InjectionPlan<?>> memo) {
+    if (memo.containsKey(n)) {
+      if (BUILDING == memo.get(n)) {
+        StringBuilder loopyList = new StringBuilder("[");
+        for (Node node : memo.keySet()) {
+          if (memo.get(node) == BUILDING) {
+            loopyList.append(" " + node.getFullName());
+          }
+        }
+        loopyList.append(" ]");
+        throw new ClassHierarchyException("Detected loopy constructor involving "
+            + loopyList.toString());
+      } else {
+        return;
+      }
+    }
+    memo.put(n, BUILDING);
+    final InjectionPlan<T> ip;
+    if (n instanceof NamedParameterNode) {
+      final NamedParameterNode<T> np = (NamedParameterNode<T>) n;
+
+      final T boundInstance = parseBoundNamedParameter(np);
+      final T defaultInstance = javaNamespace.parseDefaultValue(np);
+      final T instance = boundInstance != null ? boundInstance : defaultInstance;
+
+      if (instance instanceof Node) {
+        buildInjectionPlan((Node) instance, memo);
+        ip = new Subplan<T>(n, 0, (InjectionPlan<T>) memo.get(instance));
+      } else if (instance instanceof Set) {
+        Set<T> entries = (Set<T>) instance;
+        Set<InjectionPlan<T>> plans = new MonotonicHashSet<>();
+        for (T entry : entries) {
+          if (entry instanceof ClassNode) {
+            buildInjectionPlan((ClassNode<?>) entry, memo);
+            plans.add((InjectionPlan<T>) memo.get(entry));
+          } else {
+            plans.add(new JavaInstance<T>(n, entry));
+          }
+
+        }
+        ip = new SetInjectionPlan<T>(n, plans);
+      } else if (instance instanceof List) {
+        List<T> entries = (List<T>) instance;
+        List<InjectionPlan<T>> plans = new ArrayList<>();
+        for (T entry : entries) {
+          if (entry instanceof ClassNode) {
+            buildInjectionPlan((ClassNode<?>) entry, memo);
+            plans.add((InjectionPlan<T>) memo.get(entry));
+          } else {
+            plans.add(new JavaInstance<T>(n, entry));
+          }
+        }
+        ip = new ListInjectionPlan<T>(n, plans);
+      } else {
+        ip = new JavaInstance<T>(np, instance);
+      }
+    } else if (n instanceof ClassNode) {
+      final ClassNode<T> cn = (ClassNode<T>) n;
+
+      // Any (or all) of the next four values might be null; that's fine.
+      final T cached = getCachedInstance(cn);
+      final ClassNode<T> boundImpl = c.getBoundImplementation(cn);
+      final ClassNode<T> defaultImpl = parseDefaultImplementation(cn);
+      final ClassNode<ExternalConstructor<T>> ec = c.getBoundConstructor(cn);
+
+      ip = buildClassNodeInjectionPlan(cn, cached, ec, boundImpl, defaultImpl, memo);
+    } else if (n instanceof PackageNode) {
+      throw new IllegalArgumentException(
+          "Request to instantiate Java package as object");
+    } else {
+      throw new IllegalStateException(
+          "Type hierarchy contained unknown node type!:" + n);
+    }
+    memo.put(n, ip);
+  }
+
+  /**
+   * Return an injection plan for the given class / parameter name.
+   *
+   * @param n The name of an injectable class or interface, or a NamedParameter.
+   * @return
+   * @throws NameResolutionException
+   */
+  public InjectionPlan<?> getInjectionPlan(final Node n) {
+    Map<Node, InjectionPlan<?>> memo = new HashMap<>();
+    buildInjectionPlan(n, memo);
+    return memo.get(n);
+  }
+
+  @Override
+  public InjectionPlan<?> getInjectionPlan(String name) throws NameResolutionException {
+    return getInjectionPlan(namespace.getNode(name));
+  }
+
+  @SuppressWarnings("unchecked")
+  public <T> InjectionPlan<T> getInjectionPlan(Class<T> name) {
+    return (InjectionPlan<T>) getInjectionPlan(javaNamespace.getNode(name));
+  }
+
+  @Override
+  public boolean isInjectable(String name) throws NameResolutionException {
+    return getInjectionPlan(namespace.getNode(name)).isInjectable();
+  }
+
+  @Override
+  public boolean isInjectable(Class<?> clazz) {
+    try {
+      return isInjectable(ReflectionUtilities.getFullName(clazz));
+    } catch (NameResolutionException e) {
+      throw new IllegalStateException("Could not round trip " + clazz + " through ClassHierarchy", e);
+    }
+  }
+
+  @Override
+  public boolean isParameterSet(String name) throws NameResolutionException {
+    InjectionPlan<?> p = getInjectionPlan(namespace.getNode(name));
+    return p.isInjectable();
+  }
+
+  @Override
+  public boolean isParameterSet(Class<? extends Name<?>> name)
+      throws BindException {
+    return isParameterSet(name.getName());
+  }
+
+  private <U> U getInstance(Node n) throws InjectionException {
+    assertNotConcurrent();
+    @SuppressWarnings("unchecked")
+    InjectionPlan<U> plan = (InjectionPlan<U>) getInjectionPlan(n);
+    U u = (U) injectFromPlan(plan);
+
+    while (!pendingFutures.isEmpty()) {
+      Iterator<InjectionFuture<?>> i = pendingFutures.iterator();
+      InjectionFuture<?> f = i.next();
+      pendingFutures.remove(f);
+      f.get();
+    }
+    return u;
+  }
+
+  @Override
+  public <U> U getInstance(Class<U> clazz) throws InjectionException {
+    if (Name.class.isAssignableFrom(clazz)) {
+      throw new InjectionException("getInstance() called on Name "
+          + ReflectionUtilities.getFullName(clazz)
+          + " Did you mean to call getNamedInstance() instead?");
+    }
+    return getInstance(javaNamespace.getNode(clazz));
+  }
+
+  @SuppressWarnings("unchecked")
+  @Override
+  public <U> U getInstance(String clazz) throws InjectionException, NameResolutionException {
+    return (U) getInstance(namespace.getNode(clazz));
+  }
+
+  @Override
+  @SuppressWarnings("unchecked")
+  public <T> T getNamedInstance(Class<? extends Name<T>> clazz)
+      throws InjectionException {
+    return (T) getInstance(javaNamespace.getNode(clazz));
+  }
+
+  public <T> T getNamedParameter(Class<? extends Name<T>> clazz)
+      throws InjectionException {
+    return getNamedInstance(clazz);
+  }
+
+  private <T> java.lang.reflect.Constructor<T> getConstructor(
+      ConstructorDef<T> constructor) throws ClassNotFoundException,
+      NoSuchMethodException, SecurityException {
+    @SuppressWarnings("unchecked")
+    Class<T> clazz = (Class<T>) javaNamespace.classForName(constructor
+        .getClassName());
+    ConstructorArg[] args = constructor.getArgs();
+    Class<?> parameterTypes[] = new Class[args.length];
+    for (int i = 0; i < args.length; i++) {
+      if (args[i].isInjectionFuture()) {
+        parameterTypes[i] = InjectionFuture.class;
+      } else {
+        parameterTypes[i] = javaNamespace.classForName(args[i].getType());
+      }
+    }
+    java.lang.reflect.Constructor<T> cons = clazz
+        .getDeclaredConstructor(parameterTypes);
+    cons.setAccessible(true);
+    return cons;
+  }
+
+  /**
+   * This gets really nasty now that constructors can invoke operations on us.
+   * The upshot is that we should check to see if instances have been
+   * registered by callees after each recursive invocation of injectFromPlan or
+   * constructor invocations. The error handling currently bails if the thing we
+   * just instantiated should be discarded.
+   * <p/>
+   * This could happen if (for instance), a constructor did a
+   * bindVolatileInstance of its own class to an instance, or somehow triggered
+   * an injection of itself with a different plan (an injection of itself with
+   * the same plan would lead to an infinite recursion, so it's not really our
+   * problem).
+   *
+   * @param plan
+   * @return
+   * @throws InjectionException
+   */
+  @SuppressWarnings("unchecked")
+  private <T> T injectFromPlan(InjectionPlan<T> plan) throws InjectionException {
+
+    if (!plan.isFeasible()) {
+      throw new InjectionException("Cannot inject " + plan.getNode().getFullName() + ": "
+          + plan.toCantInjectString());
+    }
+    if (plan.isAmbiguous()) {
+      throw new InjectionException("Cannot inject " + plan.getNode().getFullName() + " "
+          + plan.toCantInjectString());
+    }
+    if (plan instanceof InjectionFuturePlan) {
+      InjectionFuturePlan<T> fut = (InjectionFuturePlan<T>) plan;
+      final String key = fut.getNode().getFullName();
+      try {
+        InjectionFuture<?> ret = new InjectionFuture<>(this, javaNamespace.classForName(fut.getNode().getFullName()));
+        pendingFutures.add(ret);
+        return (T) ret;
+      } catch (ClassNotFoundException e) {
+        throw new InjectionException("Could not get class for " + key);
+      }
+    } else if (plan.getNode() instanceof ClassNode && null != getCachedInstance((ClassNode<T>) plan.getNode())) {
+      return getCachedInstance((ClassNode<T>) plan.getNode());
+    } else if (plan instanceof JavaInstance) {
+      // TODO: Must be named parameter node.  Check.
+//      throw new IllegalStateException("Instance from plan not in Injector's set of instances?!?");
+      return ((JavaInstance<T>) plan).instance;
+    } else if (plan instanceof Constructor) {
+      final Constructor<T> constructor = (Constructor<T>) plan;
+      final Object[] args = new Object[constructor.getArgs().length];
+      final InjectionPlan<?>[] argPlans = constructor.getArgs();
+
+      for (int i = 0; i < argPlans.length; i++) {
+        args[i] = injectFromPlan(argPlans[i]);
+      }
+      try {
+        concurrentModificationGuard = true;
+        T ret;
+        try {
+          ConstructorDef<T> def = (ConstructorDef<T>) constructor.getConstructorDef();
+          java.lang.reflect.Constructor<T> c = getConstructor(def);
+
+          if (aspect != null) {
+            ret = aspect.inject(def, c, args);
+          } else {
+            ret = c.newInstance(args);
+          }
+        } catch (IllegalArgumentException e) {
+          StringBuilder sb = new StringBuilder("Internal Tang error?  Could not call constructor " + constructor.getConstructorDef() + " with arguments [");
+          for (Object o : args) {
+            sb.append("\n\t" + o);
+          }
+          sb.append("]");
+          throw new IllegalStateException(sb.toString(), e);
+        }
+        if (ret instanceof ExternalConstructor) {
+          ret = ((ExternalConstructor<T>) ret).newInstance();
+        }
+        instances.put(constructor.getNode(), ret);
+        return ret;
+      } catch (ReflectiveOperationException e) {
+        throw new InjectionException("Could not invoke constructor: " + plan, e instanceof InvocationTargetException ? e.getCause() : e);
+      } finally {
+        concurrentModificationGuard = false;
+      }
+    } else if (plan instanceof Subplan) {
+      Subplan<T> ambiguous = (Subplan<T>) plan;
+      return injectFromPlan(ambiguous.getDelegatedPlan());
+    } else if (plan instanceof SetInjectionPlan) {
+      SetInjectionPlan<T> setPlan = (SetInjectionPlan<T>) plan;
+      Set<T> ret = new MonotonicHashSet<>();
+      for (InjectionPlan<T> subplan : setPlan.getEntryPlans()) {
+        ret.add(injectFromPlan(subplan));
+      }
+      return (T) ret;
+    } else if (plan instanceof ListInjectionPlan) {
+      ListInjectionPlan<T> listPlan = (ListInjectionPlan<T>) plan;
+      List<T> ret = new ArrayList<>();
+      for (InjectionPlan<T> subplan : listPlan.getEntryPlans()) {
+        ret.add(injectFromPlan(subplan));
+      }
+      return (T) ret;
+    } else {
+      throw new IllegalStateException("Unknown plan type: " + plan);
+    }
+  }
+
+  @Override
+  public <T> void bindVolatileInstance(Class<T> c, T o) throws BindException {
+    bindVolatileInstanceNoCopy(c, o);
+  }
+
+  @Override
+  public <T> void bindVolatileParameter(Class<? extends Name<T>> c, T o)
+      throws BindException {
+    bindVolatileParameterNoCopy(c, o);
+  }
+
+  <T> void bindVolatileInstanceNoCopy(Class<T> c, T o) throws BindException {
+    assertNotConcurrent();
+    Node n = javaNamespace.getNode(c);
+    if (n instanceof ClassNode) {
+      ClassNode<?> cn = (ClassNode<?>) n;
+      Object old = getCachedInstance(cn);
+      if (old != null) {
+        throw new BindException("Attempt to re-bind instance.  Old value was "
+            + old + " new value is " + o);
+      }
+      instances.put(cn, o);
+    } else {
+      throw new IllegalArgumentException("Expected Class but got " + c
+          + " (probably a named parameter).");
+    }
+  }
+
+  <T> void bindVolatileParameterNoCopy(Class<? extends Name<T>> c, T o)
+      throws BindException {
+    Node n = javaNamespace.getNode(c);
+    if (n instanceof NamedParameterNode) {
+      NamedParameterNode<?> np = (NamedParameterNode<?>) n;
+      Object old = this.c.getNamedParameter(np);
+      if (old != null) {
+        // XXX need to get the binding site here!
+        throw new BindException(
+            "Attempt to re-bind named parameter " + ReflectionUtilities.getFullName(c) + ".  Old value was [" + old
+                + "] new value is [" + o + "]");
+      }
+      try {
+        namedParameterInstances.put(np, o);
+      } catch (IllegalArgumentException e) {
+        throw new BindException(
+            "Attempt to re-bind named parameter " + ReflectionUtilities.getFullName(c) + ".  Old value was [" + old
+                + "] new value is [" + o + "]");
+
+      }
+    } else {
+      throw new IllegalArgumentException("Expected Name, got " + c
+          + " (probably a class)");
+    }
+  }
+
+  @Override
+  public Injector forkInjector() {
+    try {
+      return forkInjector(new Configuration[0]);
+    } catch (BindException e) {
+      throw new IllegalStateException(e);
+    }
+  }
+
+  @Override
+  public Injector forkInjector(Configuration... configurations)
+      throws BindException {
+    InjectorImpl ret;
+    ret = copy(this, configurations);
+    return ret;
+  }
+
+  @Override
+  public <T> void bindAspect(Aspect a) throws BindException {
+    if (aspect != null) {
+      throw new BindException("Attempt to re-bind aspect! old=" + aspect + " new=" + a);
+    }
+    aspect = a;
+  }
+
+  @Override
+  public Aspect getAspect() {
+    return aspect;
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/java/JavaConfigurationBuilderImpl.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/java/JavaConfigurationBuilderImpl.java b/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/java/JavaConfigurationBuilderImpl.java
new file mode 100644
index 0000000..d13a014
--- /dev/null
+++ b/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/java/JavaConfigurationBuilderImpl.java
@@ -0,0 +1,238 @@
+/**
+ * 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
+ *
+ *   http://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.reef.tang.implementation.java;
+
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.ExternalConstructor;
+import org.apache.reef.tang.JavaClassHierarchy;
+import org.apache.reef.tang.JavaConfigurationBuilder;
+import org.apache.reef.tang.annotations.Name;
+import org.apache.reef.tang.exceptions.BindException;
+import org.apache.reef.tang.implementation.ConfigurationBuilderImpl;
+import org.apache.reef.tang.implementation.ConfigurationImpl;
+import org.apache.reef.tang.types.ClassNode;
+import org.apache.reef.tang.types.NamedParameterNode;
+import org.apache.reef.tang.types.Node;
+import org.apache.reef.tang.util.ReflectionUtilities;
+
+import java.lang.reflect.Type;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+public class JavaConfigurationBuilderImpl extends ConfigurationBuilderImpl
+    implements JavaConfigurationBuilder {
+
+  public JavaConfigurationBuilderImpl(URL[] jars, Configuration[] confs, Class<? extends ExternalConstructor<?>>[] parsers)
+      throws BindException {
+    super(jars, confs, parsers);
+  }
+
+  JavaConfigurationBuilderImpl() {
+    super();
+  }
+
+  public JavaConfigurationBuilderImpl(URL[] jars) throws BindException {
+    super(jars);
+  }
+
+  JavaConfigurationBuilderImpl(JavaConfigurationBuilderImpl impl) {
+    super(impl);
+  }
+
+  public JavaConfigurationBuilderImpl(Configuration[] confs)
+      throws BindException {
+    super(confs);
+  }
+
+  @Override
+  public ConfigurationImpl build() {
+    return new JavaConfigurationImpl(new JavaConfigurationBuilderImpl(this));
+  }
+
+  private Node getNode(Class<?> c) {
+    return ((JavaClassHierarchy) namespace).getNode(c);
+  }
+
+  @Override
+  public <T> JavaConfigurationBuilder bind(Class<T> c, Class<?> val) throws BindException {
+    super.bind(getNode(c), getNode(val));
+    return this;
+  }
+
+  @Override
+  @SuppressWarnings("unchecked")
+  public <T> JavaConfigurationBuilder bindImplementation(Class<T> c, Class<? extends T> d)
+      throws BindException {
+    final Node cn = getNode(c);
+    final Node dn = getNode(d);
+    if (!(cn instanceof ClassNode)) {
+      throw new BindException(
+          "bindImplementation passed interface that resolved to " + cn
+              + " expected a ClassNode<?>");
+    }
+    if (!(dn instanceof ClassNode)) {
+      throw new BindException(
+          "bindImplementation passed implementation that resolved to " + dn
+              + " expected a ClassNode<?>");
+    }
+    super.bindImplementation((ClassNode<T>) cn, (ClassNode<? extends T>) dn);
+    return this;
+  }
+
+  @Override
+  public JavaConfigurationBuilder bindNamedParameter(final Class<? extends Name<?>> name, final String value)
+      throws BindException {
+    if (value == null) {
+      throw new IllegalStateException("The value null set to the named parameter is illegal: " + name);
+    }
+    final Node np = getNode(name);
+    if (np instanceof NamedParameterNode) {
+      super.bindParameter((NamedParameterNode<?>) np, value);
+      return this;
+    } else {
+      throw new BindException(
+          "Detected type mismatch when setting named parameter " + name
+              + "  Expected NamedParameterNode, but namespace contains a " + np);
+    }
+  }
+
+  @Override
+  public <T> JavaConfigurationBuilder bindNamedParameter(Class<? extends Name<T>> iface,
+                                                         Class<? extends T> impl) throws BindException {
+    Node ifaceN = getNode(iface);
+    Node implN = getNode(impl);
+    if (!(ifaceN instanceof NamedParameterNode)) {
+      throw new BindException("Type mismatch when setting named parameter " + ifaceN
+          + " Expected NamedParameterNode");
+    }
+    bind(ifaceN, implN);
+    return this;
+  }
+
+  @SuppressWarnings({"unchecked"})
+  public <T> JavaConfigurationBuilder bindConstructor(Class<T> c,
+                                                      Class<? extends ExternalConstructor<? extends T>> v) throws BindException {
+    final Node n = getNode(c);
+    final Node m = getNode(v);
+    if (!(n instanceof ClassNode)) {
+      throw new BindException("BindConstructor got class that resolved to " + n + "; expected ClassNode");
+    }
+    if (!(m instanceof ClassNode)) {
+      throw new BindException("BindConstructor got class that resolved to " + m + "; expected ClassNode");
+    }
+    super.bindConstructor((ClassNode<T>) n, (ClassNode<? extends ExternalConstructor<? extends T>>) m);
+    return this;
+  }
+
+  @SuppressWarnings("unchecked")
+  @Override
+  public <T> JavaConfigurationBuilder bindSetEntry(Class<? extends Name<Set<T>>> iface, String value) throws BindException {
+    final Node n = getNode(iface);
+
+    if (!(n instanceof NamedParameterNode)) {
+      throw new BindException("BindSetEntry got an interface that resolved to " + n + "; expected a NamedParameter");
+    }
+    final Type setType = ReflectionUtilities.getInterfaceTarget(Name.class, iface);
+    if (!ReflectionUtilities.getRawClass(setType).equals(Set.class)) {
+      throw new BindException("BindSetEntry got a NamedParameter that takes a " + setType + "; expected Set<...>");
+    }
+//    Type valType = ReflectionUtilities.getInterfaceTarget(Set.class, setType);
+    super.bindSetEntry((NamedParameterNode<Set<T>>) n, value);
+    return this;
+  }
+
+  @SuppressWarnings("unchecked")
+  @Override
+  public <T> JavaConfigurationBuilder bindSetEntry(Class<? extends Name<Set<T>>> iface, Class<? extends T> impl) throws BindException {
+    final Node n = getNode(iface);
+    final Node m = getNode(impl);
+
+    if (!(n instanceof NamedParameterNode)) {
+      throw new BindException("BindSetEntry got an interface that resolved to " + n + "; expected a NamedParameter");
+    }
+    final Type setType = ReflectionUtilities.getInterfaceTarget(Name.class, iface);
+    if (!ReflectionUtilities.getRawClass(setType).equals(Set.class)) {
+      throw new BindException("BindSetEntry got a NamedParameter that takes a " + setType + "; expected Set<...>");
+    }
+    final Type valType = ReflectionUtilities.getInterfaceTarget(Set.class, setType);
+    if (!ReflectionUtilities.getRawClass(valType).isAssignableFrom(impl)) {
+      throw new BindException("BindSetEntry got implementation " + impl + " that is incompatible with expected type " + valType);
+    }
+
+    super.bindSetEntry((NamedParameterNode<Set<T>>) n, m);
+    return this;
+  }
+
+  /**
+   * Binding list method for JavaConfigurationBuilder. It checks the type of a given named parameter,
+   * and whether all the list's elements can be applied to the named parameter. The elements' type
+   * should be either java Class or String.
+   * <p/>
+   * It does not check whether the list's String values can be parsed to T, like bindSetEntry.
+   *
+   * @param iface    target named parameter to be instantiated
+   * @param implList implementation list used to instantiate the named parameter
+   * @param <T>      type of the list
+   * @return bound JavaConfigurationBuilder object
+   * @throws BindException
+   */
+  @SuppressWarnings("unchecked")
+  @Override
+  public <T> JavaConfigurationBuilder bindList(Class<? extends Name<List<T>>> iface, List implList)
+      throws BindException {
+    final Node n = getNode(iface);
+    List<Object> result = new ArrayList<>();
+
+    if (!(n instanceof NamedParameterNode)) {
+      throw new BindException("BindList got an interface that resolved to " + n + "; expected a NamedParameter");
+    }
+    final Type listType = ReflectionUtilities.getInterfaceTarget(Name.class, iface);
+    if (!ReflectionUtilities.getRawClass(listType).equals(List.class)) {
+      throw new BindException("BindList got a NamedParameter that takes a " + listType + "; expected List<...>");
+    }
+    if (!implList.isEmpty()) {
+      final Type valType = ReflectionUtilities.getInterfaceTarget(List.class, listType);
+      for (Object item : implList) {
+        if (item instanceof Class) {
+          if (!ReflectionUtilities.getRawClass(valType).isAssignableFrom((Class) item)) {
+            throw new BindException("BindList got a list element which is not assignable to the given Type; " +
+                "expected: " + valType);
+          }
+          result.add(getNode((Class) item));
+        } else if (item instanceof String) {
+          result.add(item);
+        } else {
+          throw new BindException("BindList got an list element with unsupported type; expected Class or String " +
+              "object");
+        }
+      }
+    }
+
+    super.bindList((NamedParameterNode<List<T>>) n, result);
+    return this;
+  }
+
+  private class JavaConfigurationImpl extends ConfigurationImpl {
+    JavaConfigurationImpl(JavaConfigurationBuilderImpl builder) {
+      super(builder);
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/java/JavaInstance.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/java/JavaInstance.java b/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/java/JavaInstance.java
new file mode 100644
index 0000000..2d1a21c
--- /dev/null
+++ b/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/java/JavaInstance.java
@@ -0,0 +1,75 @@
+/**
+ * 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
+ *
+ *   http://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.reef.tang.implementation.java;
+
+import org.apache.reef.tang.implementation.InjectionPlan;
+import org.apache.reef.tang.types.Node;
+
+final public class JavaInstance<T> extends InjectionPlan<T> {
+  final T instance;
+
+  public JavaInstance(Node name, T instance) {
+    super(name);
+    this.instance = instance;
+  }
+
+  @Override
+  public int getNumAlternatives() {
+    return instance == null ? 0 : 1;
+  }
+
+  @Override
+  public String toString() {
+    return getNode() + " = " + instance;
+  }
+
+  @Override
+  public boolean isAmbiguous() {
+    return false;
+  }
+
+  @Override
+  public boolean isInjectable() {
+    return instance != null;
+  }
+
+  public String getInstanceAsString() {
+    return instance.toString();
+  }
+
+  @Override
+  protected String toAmbiguousInjectString() {
+    throw new IllegalArgumentException("toAmbiguousInjectString called on JavaInstance!" + this.toString());
+  }
+
+  @Override
+  protected String toInfeasibleInjectString() {
+    return getNode() + " is not bound.";
+  }
+
+  @Override
+  protected boolean isInfeasibleLeaf() {
+    return true;
+  }
+
+  @Override
+  public String toShallowString() {
+    return toString();
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/java/JavaNodeFactory.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/java/JavaNodeFactory.java b/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/java/JavaNodeFactory.java
new file mode 100644
index 0000000..f727c89
--- /dev/null
+++ b/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/java/JavaNodeFactory.java
@@ -0,0 +1,333 @@
+/**
+ * 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
+ *
+ *   http://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.reef.tang.implementation.java;
+
+import org.apache.reef.tang.ExternalConstructor;
+import org.apache.reef.tang.InjectionFuture;
+import org.apache.reef.tang.annotations.*;
+import org.apache.reef.tang.exceptions.ClassHierarchyException;
+import org.apache.reef.tang.implementation.types.*;
+import org.apache.reef.tang.types.*;
+import org.apache.reef.tang.util.MonotonicSet;
+import org.apache.reef.tang.util.ReflectionUtilities;
+
+import javax.inject.Inject;
+import java.lang.annotation.Annotation;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Modifier;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+
+public class JavaNodeFactory {
+
+  @SuppressWarnings("unchecked")
+  static <T> ClassNodeImpl<T> createClassNode(Node parent, Class<T> clazz) throws ClassHierarchyException {
+    final boolean injectable;
+    final boolean unit = clazz.isAnnotationPresent(Unit.class);
+    final String simpleName = ReflectionUtilities.getSimpleName(clazz);
+    final String fullName = ReflectionUtilities.getFullName(clazz);
+    final boolean isStatic = Modifier.isStatic(clazz.getModifiers());
+    final boolean parentIsUnit = ((parent instanceof ClassNode) && !isStatic) ?
+        ((ClassNode<?>) parent).isUnit() : false;
+
+    if (clazz.isLocalClass() || clazz.isMemberClass()) {
+      if (!isStatic) {
+        if (parent instanceof ClassNode) {
+          injectable = ((ClassNode<?>) parent).isUnit();
+        } else {
+          injectable = false;
+        }
+      } else {
+        injectable = true;
+      }
+    } else {
+      injectable = true;
+    }
+
+    boolean foundNonStaticInnerClass = false;
+    for (Class<?> c : clazz.getDeclaredClasses()) {
+      if (!Modifier.isStatic(c.getModifiers())) {
+        foundNonStaticInnerClass = true;
+      }
+    }
+
+    if (unit && !foundNonStaticInnerClass) {
+      throw new ClassHierarchyException("Class " + ReflectionUtilities.getFullName(clazz) + " has an @Unit annotation, but no non-static inner classes.  Such @Unit annotations would have no effect, and are therefore disallowed.");
+    }
+
+    Constructor<T>[] constructors = (Constructor<T>[]) clazz
+        .getDeclaredConstructors();
+    MonotonicSet<ConstructorDef<T>> injectableConstructors = new MonotonicSet<>();
+    ArrayList<ConstructorDef<T>> allConstructors = new ArrayList<>();
+    for (int k = 0; k < constructors.length; k++) {
+      boolean constructorAnnotatedInjectable = (constructors[k]
+          .getAnnotation(Inject.class) != null);
+      if (constructorAnnotatedInjectable && constructors[k].isSynthetic()) {
+        // Not sure if we *can* unit test this one.
+        throw new ClassHierarchyException(
+            "Synthetic constructor was annotated with @Inject!");
+      }
+      if (parentIsUnit && (constructorAnnotatedInjectable || constructors[k].getParameterTypes().length != 1)) {
+        throw new ClassHierarchyException(
+            "Detected explicit constructor in class enclosed in @Unit " + fullName + "  Such constructors are disallowed.");
+      }
+      boolean constructorInjectable = constructorAnnotatedInjectable || parentIsUnit;
+      // ConstructorDef's constructor checks for duplicate
+      // parameters
+      // The injectableConstructors set checks for ambiguous
+      // boundConstructors.
+      ConstructorDef<T> def = JavaNodeFactory.createConstructorDef(injectable,
+          constructors[k], constructorAnnotatedInjectable);
+      if (constructorInjectable) {
+        if (injectableConstructors.contains(def)) {
+          throw new ClassHierarchyException(
+              "Ambiguous boundConstructors detected in class " + clazz + ": "
+                  + def + " differs from some other" + " constructor only "
+                  + "by parameter order.");
+        } else {
+          injectableConstructors.add(def);
+        }
+      }
+      allConstructors.add(def);
+    }
+    final String defaultImplementation;
+    if (clazz.isAnnotationPresent(DefaultImplementation.class)) {
+      DefaultImplementation defaultImpl
+          = clazz.getAnnotation(DefaultImplementation.class);
+      Class<?> defaultImplementationClazz = defaultImpl.value();
+      if (defaultImplementationClazz.equals(Void.class)) {
+        defaultImplementation = defaultImpl.name();
+        // XXX check isAssignableFrom, other type problems here.
+      } else {
+        if (!clazz.isAssignableFrom(defaultImplementationClazz)) {
+          throw new ClassHierarchyException(clazz
+              + " declares its default implementation to be non-subclass "
+              + defaultImplementationClazz);
+        }
+        defaultImplementation = ReflectionUtilities.getFullName(defaultImplementationClazz);
+      }
+    } else {
+      defaultImplementation = null;
+    }
+
+    return new ClassNodeImpl<T>(parent, simpleName, fullName, unit, injectable,
+        ExternalConstructor.class.isAssignableFrom(clazz),
+        injectableConstructors.toArray(new ConstructorDefImpl[0]),
+        allConstructors.toArray(new ConstructorDefImpl[0]), defaultImplementation);
+  }
+
+  /**
+   * XXX: This method assumes that all generic types have exactly one type parameter.
+   */
+  public static <T> NamedParameterNode<T> createNamedParameterNode(Node parent,
+                                                                   Class<? extends Name<T>> clazz, Type argClass) throws ClassHierarchyException {
+
+    Class<?> argRawClass = ReflectionUtilities.getRawClass(argClass);
+
+    final boolean isSet = argRawClass.equals(Set.class);
+    final boolean isList = argRawClass.equals(List.class);
+
+
+    if (isSet || isList) {
+      argClass = ReflectionUtilities.getInterfaceTarget(Collection.class, argClass);
+      argRawClass = ReflectionUtilities.getRawClass(argClass);
+    }
+
+    final String simpleName = ReflectionUtilities.getSimpleName(clazz);
+    final String fullName = ReflectionUtilities.getFullName(clazz);
+    final String fullArgName = ReflectionUtilities.getFullName(argClass);
+    final String simpleArgName = ReflectionUtilities.getSimpleName(argClass);
+
+
+    final NamedParameter namedParameter = clazz.getAnnotation(NamedParameter.class);
+
+    if (namedParameter == null) {
+      throw new IllegalStateException("Got name without named parameter post-validation!");
+    }
+    final boolean hasStringDefault, hasClassDefault, hasStringSetDefault, hasClassSetDefault;
+
+    int default_count = 0;
+    if (!namedParameter.default_value().isEmpty()) {
+      hasStringDefault = true;
+      default_count++;
+    } else {
+      hasStringDefault = false;
+    }
+    if (namedParameter.default_class() != Void.class) {
+      hasClassDefault = true;
+      default_count++;
+    } else {
+      hasClassDefault = false;
+    }
+    if (namedParameter.default_values() != null && namedParameter.default_values().length > 0) {
+      hasStringSetDefault = true;
+      default_count++;
+    } else {
+      hasStringSetDefault = false;
+    }
+    if (namedParameter.default_classes() != null && namedParameter.default_classes().length > 0) {
+      hasClassSetDefault = true;
+      default_count++;
+    } else {
+      hasClassSetDefault = false;
+    }
+    if (default_count > 1) {
+      throw new ClassHierarchyException("Named parameter " + fullName + " defines more than one of default_value, default_class, default_values and default_classes");
+    }
+
+    final String[] defaultInstanceAsStrings;
+
+    if (default_count == 0) {
+      defaultInstanceAsStrings = new String[]{};
+    } else if (hasClassDefault) {
+      final Class<?> default_class = namedParameter.default_class();
+      assertIsSubclassOf(clazz, default_class, argClass);
+      defaultInstanceAsStrings = new String[]{ReflectionUtilities.getFullName(default_class)};
+    } else if (hasStringDefault) {
+      // Don't know if the string is a class or literal here, so don't bother validating.
+      defaultInstanceAsStrings = new String[]{namedParameter.default_value()};
+    } else if (hasClassSetDefault) {
+      Class<?>[] clzs = namedParameter.default_classes();
+      defaultInstanceAsStrings = new String[clzs.length];
+      for (int i = 0; i < clzs.length; i++) {
+        assertIsSubclassOf(clazz, clzs[i], argClass);
+        defaultInstanceAsStrings[i] = ReflectionUtilities.getFullName(clzs[i]);
+      }
+    } else if (hasStringSetDefault) {
+      defaultInstanceAsStrings = namedParameter.default_values();
+    } else {
+      throw new IllegalStateException();
+    }
+
+    final String documentation = namedParameter.doc();
+
+    final String shortName = namedParameter.short_name().isEmpty()
+        ? null : namedParameter.short_name();
+
+    return new NamedParameterNodeImpl<>(parent, simpleName, fullName,
+        fullArgName, simpleArgName, isSet, isList, documentation, shortName, defaultInstanceAsStrings);
+  }
+
+  private static void assertIsSubclassOf(Class<?> named_parameter, Class<?> default_class,
+                                         Type argClass) {
+    boolean isSubclass = false;
+    boolean isGenericSubclass = false;
+    Class<?> argRawClass = ReflectionUtilities.getRawClass(argClass);
+
+    // Note: We intentionally strip the raw type information here.  The reason is to handle
+    // EventHandler-style patterns and collections.
+
+    /// If we have a Name that takes EventHandler<A>, we want to be able to pass in an EventHandler<Object>.
+
+    for (final Type c : ReflectionUtilities.classAndAncestors(default_class)) {
+      if (ReflectionUtilities.getRawClass(c).equals(argRawClass)) {
+        isSubclass = true;
+        if (argClass instanceof ParameterizedType &&
+            c instanceof ParameterizedType) {
+          ParameterizedType argPt = (ParameterizedType) argClass;
+          ParameterizedType defaultPt = (ParameterizedType) c;
+
+          Class<?> rawDefaultParameter = ReflectionUtilities.getRawClass(defaultPt.getActualTypeArguments()[0]);
+          Class<?> rawArgParameter = ReflectionUtilities.getRawClass(argPt.getActualTypeArguments()[0]);
+
+          for (final Type d : ReflectionUtilities.classAndAncestors(argPt.getActualTypeArguments()[0])) {
+            if (ReflectionUtilities.getRawClass(d).equals(rawDefaultParameter)) {
+              isGenericSubclass = true;
+            }
+          }
+          for (final Type d : ReflectionUtilities.classAndAncestors(defaultPt.getActualTypeArguments()[0])) {
+            if (ReflectionUtilities.getRawClass(d).equals(rawArgParameter)) {
+              isGenericSubclass = true;
+            }
+          }
+        } else {
+          isGenericSubclass = true;
+        }
+      }
+    }
+
+    if (!(isSubclass)) {
+      throw new ClassHierarchyException(named_parameter + " defines a default class "
+          + ReflectionUtilities.getFullName(default_class) + " with a raw type that does not extend of its target's raw type " + argRawClass);
+    }
+    if (!(isGenericSubclass)) {
+      throw new ClassHierarchyException(named_parameter + " defines a default class "
+          + ReflectionUtilities.getFullName(default_class) + " with a type that does not extend its target's type " + argClass);
+    }
+  }
+
+  public static PackageNode createRootPackageNode() {
+    return new PackageNodeImpl();
+  }
+
+  private static <T> ConstructorDef<T> createConstructorDef(
+      boolean isClassInjectionCandidate, Constructor<T> constructor,
+      boolean injectable) throws ClassHierarchyException {
+    // We don't support injection of non-static member classes with @Inject
+    // annotations.
+    if (injectable && !isClassInjectionCandidate) {
+      throw new ClassHierarchyException("Cannot @Inject non-static member class unless the enclosing class an @Unit.  Nested class is:"
+          + ReflectionUtilities.getFullName(constructor.getDeclaringClass()));
+    }
+    // TODO: When we use paramTypes below, we strip generic parameters.  Is that OK?
+    Class<?>[] paramTypes = constructor.getParameterTypes();
+    Type[] genericParamTypes = constructor.getGenericParameterTypes();
+    Annotation[][] paramAnnotations = constructor.getParameterAnnotations();
+    if (paramTypes.length != paramAnnotations.length) {
+      throw new IllegalStateException();
+    }
+    ConstructorArg[] args = new ConstructorArg[genericParamTypes.length];
+    for (int i = 0; i < genericParamTypes.length; i++) {
+      // If this parameter is an injection future, unwrap the target class,
+      // and remember by setting isFuture to true.
+      final Type type;
+      final boolean isFuture;
+      if (InjectionFuture.class.isAssignableFrom(paramTypes[i])) {
+        type = ReflectionUtilities.getInterfaceTarget(InjectionFuture.class, genericParamTypes[i]);
+        isFuture = true;
+      } else {
+        type = paramTypes[i];
+        isFuture = false;
+      }
+      // Make node of the named parameter annotation (if any).
+      Parameter named = null;
+      for (int j = 0; j < paramAnnotations[i].length; j++) {
+        Annotation annotation = paramAnnotations[i][j];
+        if (annotation instanceof Parameter) {
+          if ((!isClassInjectionCandidate) || !injectable) {
+            throw new ClassHierarchyException(constructor + " is not injectable, but it has an @Parameter annotation.");
+          }
+          named = (Parameter) annotation;
+        }
+      }
+      args[i] = new ConstructorArgImpl(
+          ReflectionUtilities.getFullName(type), named == null ? null
+          : ReflectionUtilities.getFullName(named.value()),
+          isFuture);
+    }
+    return new ConstructorDefImpl<T>(
+        ReflectionUtilities.getFullName(constructor.getDeclaringClass()),
+        args, injectable);
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/java/package-info.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/java/package-info.java b/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/java/package-info.java
new file mode 100644
index 0000000..cc8cbc0
--- /dev/null
+++ b/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/java/package-info.java
@@ -0,0 +1,29 @@
+/**
+ * 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
+ *
+ *   http://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.
+ */
+/**
+ * Private implementation classes that configure and inject code written
+ * in Java.  Tang supports cross-language object injections.  The classes
+ * in org.apache.reef.tang.implementation are agnostic to the application
+ * language.  In contrast, the classes in this package provide convenience
+ * APIs that move type-checking to Java's generic type system, and are also
+ * responsible for actually injecting objects (since, by definition, Java
+ * JVMs inject Java objects).  
+ */
+package org.apache.reef.tang.implementation.java;
+

http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/package-info.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/package-info.java b/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/package-info.java
new file mode 100644
index 0000000..7c68e79
--- /dev/null
+++ b/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/package-info.java
@@ -0,0 +1,27 @@
+/**
+ * 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
+ *
+ *   http://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.
+ */
+/**
+ * Tang's implementation.  Ideally, applications should never need to be
+ * aware of anything in this package or its sub-packages.  As of this
+ * writing, some layering violations still exist, and portions of this
+ * package (such as the InjectionPlan APIs) are needed in rare
+ * circumstances. 
+ */
+package org.apache.reef.tang.implementation;
+

http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/protobuf/ProtocolBufferClassHierarchy.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/protobuf/ProtocolBufferClassHierarchy.java b/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/protobuf/ProtocolBufferClassHierarchy.java
new file mode 100644
index 0000000..8053078
--- /dev/null
+++ b/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/protobuf/ProtocolBufferClassHierarchy.java
@@ -0,0 +1,396 @@
+/**
+ * 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
+ *
+ *   http://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.reef.tang.implementation.protobuf;
+
+import org.apache.reef.tang.ClassHierarchy;
+import org.apache.reef.tang.exceptions.NameResolutionException;
+import org.apache.reef.tang.implementation.types.*;
+import org.apache.reef.tang.proto.ClassHierarchyProto;
+import org.apache.reef.tang.types.*;
+
+import java.io.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+
+public class ProtocolBufferClassHierarchy implements ClassHierarchy {
+
+  private static final String regex = "[\\.\\$\\+]";
+  private final PackageNode namespace;
+  private HashMap<String, Node> lookupTable = new HashMap<>();
+
+  // ############## Serialize implementation ############## 
+
+  // protoc doesn't believe in auto-generating constructors, so here are
+  // hand-generated ones. *sigh*
+
+  /**
+   * Deserialize a class hierarchy from a protocol buffer object.  The resulting
+   * object is immutable, and does not make use of reflection to fill in any
+   * missing values.  This allows it to represent non-native classes as well
+   * as snapshots of Java class hierarchies.
+   */
+  public ProtocolBufferClassHierarchy(ClassHierarchyProto.Node root) {
+    namespace = new PackageNodeImpl();
+    if (!root.hasPackageNode()) {
+      throw new IllegalArgumentException("Expected a package node.  Got: "
+          + root);
+    }
+    // Register all the classes.
+    for (ClassHierarchyProto.Node child : root.getChildrenList()) {
+      parseSubHierarchy(namespace, child);
+    }
+    buildLookupTable(namespace);
+    // Now, register the implementations
+    for (ClassHierarchyProto.Node child : root.getChildrenList()) {
+      wireUpInheritanceRelationships(child);
+    }
+  }
+
+  private static ClassHierarchyProto.Node newClassNode(String name,
+                                                       String fullName, boolean isInjectionCandidate,
+                                                       boolean isExternalConstructor, boolean isUnit,
+                                                       List<ClassHierarchyProto.ConstructorDef> injectableConstructors,
+                                                       List<ClassHierarchyProto.ConstructorDef> otherConstructors,
+                                                       List<String> implFullNames, Iterable<ClassHierarchyProto.Node> children) {
+    return ClassHierarchyProto.Node
+        .newBuilder()
+        .setName(name)
+        .setFullName(fullName)
+        .setClassNode(
+            ClassHierarchyProto.ClassNode.newBuilder()
+                .setIsInjectionCandidate(isInjectionCandidate)
+                .setIsExternalConstructor(isExternalConstructor)
+                .setIsUnit(isUnit)
+                .addAllInjectableConstructors(injectableConstructors)
+                .addAllOtherConstructors(otherConstructors)
+                .addAllImplFullNames(implFullNames).build())
+        .addAllChildren(children).build();
+  }
+
+  private static ClassHierarchyProto.Node newNamedParameterNode(String name,
+                                                                String fullName, String simpleArgClassName, String fullArgClassName,
+                                                                boolean isSet,
+                                                                boolean isList,
+                                                                String documentation, // can be null
+                                                                String shortName, // can be null
+                                                                String[] instanceDefault, // can be null
+                                                                Iterable<ClassHierarchyProto.Node> children) {
+    ClassHierarchyProto.NamedParameterNode.Builder namedParameterNodeBuilder
+        = ClassHierarchyProto.NamedParameterNode.newBuilder()
+        .setSimpleArgClassName(simpleArgClassName)
+        .setFullArgClassName(fullArgClassName)
+        .setIsSet(isSet)
+        .setIsList(isList);
+    if (documentation != null) {
+      namedParameterNodeBuilder.setDocumentation(documentation);
+    }
+    if (shortName != null) {
+      namedParameterNodeBuilder.setShortName(shortName);
+    }
+    if (instanceDefault != null) {
+      namedParameterNodeBuilder.addAllInstanceDefault(Arrays.asList(instanceDefault));
+    }
+
+    return ClassHierarchyProto.Node.newBuilder().setName(name)
+        .setFullName(fullName)
+        .setNamedParameterNode(namedParameterNodeBuilder.build())
+        .addAllChildren(children).build();
+  }
+
+  private static ClassHierarchyProto.Node newPackageNode(String name,
+                                                         String fullName, Iterable<ClassHierarchyProto.Node> children) {
+    return ClassHierarchyProto.Node.newBuilder()
+        .setPackageNode(ClassHierarchyProto.PackageNode.newBuilder().build())
+        .setName(name).setFullName(fullName).addAllChildren(children).build();
+  }
+
+  private static ClassHierarchyProto.ConstructorDef newConstructorDef(
+      String fullClassName, List<ClassHierarchyProto.ConstructorArg> args) {
+    return ClassHierarchyProto.ConstructorDef.newBuilder()
+        .setFullClassName(fullClassName).addAllArgs(args).build();
+  }
+
+  // these serialize...() methods copy a pieces of the class hierarchy into
+  // protobufs 
+
+  private static ClassHierarchyProto.ConstructorArg newConstructorArg(
+      String fullArgClassName, String namedParameterName, boolean isFuture) {
+    ClassHierarchyProto.ConstructorArg.Builder builder =
+        ClassHierarchyProto.ConstructorArg.newBuilder()
+            .setFullArgClassName(fullArgClassName)
+            .setIsInjectionFuture(isFuture);
+    if (namedParameterName != null) {
+      builder.setNamedParameterName(namedParameterName).build();
+    }
+    return builder.build();
+  }
+
+  private static ClassHierarchyProto.ConstructorDef serializeConstructorDef(
+      ConstructorDef<?> def) {
+    List<ClassHierarchyProto.ConstructorArg> args = new ArrayList<>();
+    for (ConstructorArg arg : def.getArgs()) {
+      args.add(newConstructorArg(arg.getType(), arg.getNamedParameterName(), arg.isInjectionFuture()));
+    }
+    return newConstructorDef(def.getClassName(), args);
+  }
+
+  private static ClassHierarchyProto.Node serializeNode(Node n) {
+    List<ClassHierarchyProto.Node> children = new ArrayList<>();
+    for (Node child : n.getChildren()) {
+      children.add(serializeNode(child));
+    }
+    if (n instanceof ClassNode) {
+      ClassNode<?> cn = (ClassNode<?>) n;
+      ConstructorDef<?>[] injectable = cn.getInjectableConstructors();
+      ConstructorDef<?>[] all = cn.getAllConstructors();
+      List<ConstructorDef<?>> others = new ArrayList<>(Arrays.asList(all));
+      others.removeAll(Arrays.asList(injectable));
+
+      List<ClassHierarchyProto.ConstructorDef> injectableConstructors = new ArrayList<>();
+      for (ConstructorDef<?> inj : injectable) {
+        injectableConstructors.add(serializeConstructorDef(inj));
+      }
+      List<ClassHierarchyProto.ConstructorDef> otherConstructors = new ArrayList<>();
+      for (ConstructorDef<?> other : others) {
+        otherConstructors.add(serializeConstructorDef(other));
+      }
+      List<String> implFullNames = new ArrayList<>();
+      for (ClassNode<?> impl : cn.getKnownImplementations()) {
+        implFullNames.add(impl.getFullName());
+      }
+      return newClassNode(cn.getName(), cn.getFullName(),
+          cn.isInjectionCandidate(), cn.isExternalConstructor(), cn.isUnit(),
+          injectableConstructors, otherConstructors, implFullNames, children);
+    } else if (n instanceof NamedParameterNode) {
+      NamedParameterNode<?> np = (NamedParameterNode<?>) n;
+      return newNamedParameterNode(np.getName(), np.getFullName(),
+          np.getSimpleArgName(), np.getFullArgName(), np.isSet(), np.isList(), np.getDocumentation(),
+          np.getShortName(), np.getDefaultInstanceAsStrings(), children);
+    } else if (n instanceof PackageNode) {
+      return newPackageNode(n.getName(), n.getFullName(), children);
+    } else {
+      throw new IllegalStateException("Encountered unknown type of Node: " + n);
+    }
+  }
+
+  /**
+   * Serialize a class hierarchy into a protocol buffer object.
+   *
+   * @param classHierarchy
+   * @return
+   */
+  public static ClassHierarchyProto.Node serialize(ClassHierarchy classHierarchy) {
+    return serializeNode(classHierarchy.getNamespace());
+  }
+
+  /**
+   * serialize a class hierarchy into a file
+   *
+   * @param file
+   * @param classHierarchy
+   * @throws IOException
+   */
+  public static void serialize(final File file, final ClassHierarchy classHierarchy) throws IOException {
+    final ClassHierarchyProto.Node node = serializeNode(classHierarchy.getNamespace());
+    try (final FileOutputStream output = new FileOutputStream(file)) {
+      try (final DataOutputStream dos = new DataOutputStream(output)) {
+        node.writeTo(dos);
+      }
+    }
+  }
+
+  /**
+   * Deserialize a class hierarchy from a file. The file can be generated from either Java or C#
+   *
+   * @param file
+   * @return
+   * @throws IOException
+   */
+  public static ClassHierarchy deserialize(final File file) throws IOException {
+    try (final InputStream stream = new FileInputStream(file)) {
+      final ClassHierarchyProto.Node root = ClassHierarchyProto.Node.parseFrom(stream);
+      return new ProtocolBufferClassHierarchy(root);
+    }
+  }
+
+  private static void parseSubHierarchy(Node parent, ClassHierarchyProto.Node n) {
+    final Node parsed;
+    if (n.hasPackageNode()) {
+      parsed = new PackageNodeImpl(parent, n.getName(), n.getFullName());
+    } else if (n.hasNamedParameterNode()) {
+      ClassHierarchyProto.NamedParameterNode np = n.getNamedParameterNode();
+      parsed = new NamedParameterNodeImpl<Object>(parent, n.getName(),
+          n.getFullName(), np.getFullArgClassName(), np.getSimpleArgClassName(),
+          np.getIsSet(), np.getIsList(), np.getDocumentation(), np.getShortName(),
+          np.getInstanceDefaultList().toArray(new String[0]));
+    } else if (n.hasClassNode()) {
+      ClassHierarchyProto.ClassNode cn = n.getClassNode();
+      List<ConstructorDef<?>> injectableConstructors = new ArrayList<>();
+      List<ConstructorDef<?>> allConstructors = new ArrayList<>();
+
+      for (ClassHierarchyProto.ConstructorDef injectable : cn
+          .getInjectableConstructorsList()) {
+        ConstructorDef<?> def = parseConstructorDef(injectable, true);
+        injectableConstructors.add(def);
+        allConstructors.add(def);
+      }
+      for (ClassHierarchyProto.ConstructorDef other : cn
+          .getOtherConstructorsList()) {
+        ConstructorDef<?> def = parseConstructorDef(other, false);
+        allConstructors.add(def);
+
+      }
+      @SuppressWarnings("unchecked")
+      ConstructorDef<Object>[] dummy = new ConstructorDef[0];
+      parsed = new ClassNodeImpl<>(parent, n.getName(), n.getFullName(),
+          cn.getIsUnit(), cn.getIsInjectionCandidate(),
+          cn.getIsExternalConstructor(), injectableConstructors.toArray(dummy),
+          allConstructors.toArray(dummy), cn.getDefaultImplementation());
+    } else {
+      throw new IllegalStateException("Bad protocol buffer: got abstract node"
+          + n);
+    }
+    for (ClassHierarchyProto.Node child : n.getChildrenList()) {
+      parseSubHierarchy(parsed, child);
+    }
+  }
+
+  private static ConstructorDef<?> parseConstructorDef(
+      org.apache.reef.tang.proto.ClassHierarchyProto.ConstructorDef def,
+      boolean isInjectable) {
+    List<ConstructorArg> args = new ArrayList<>();
+    for (ClassHierarchyProto.ConstructorArg arg : def.getArgsList()) {
+      args.add(new ConstructorArgImpl(arg.getFullArgClassName(), arg
+          .getNamedParameterName(), arg.getIsInjectionFuture()));
+    }
+    return new ConstructorDefImpl<>(def.getFullClassName(),
+        args.toArray(new ConstructorArg[0]), isInjectable);
+  }
+
+  private static String getNthPrefix(String str, int n) {
+    n++; // want this function to be zero indexed...
+    for (int i = 0; i < str.length(); i++) {
+      char c = str.charAt(i);
+      if (c == '.' || c == '$' || c == '+') {
+        n--;
+      }
+      if (n == 0) {
+        return str.substring(0, i);
+      }
+    }
+    if (n == 1) {
+      return str;
+    } else {
+      throw new ArrayIndexOutOfBoundsException();
+    }
+  }
+
+  private void buildLookupTable(Node n) {
+    for (Node child : n.getChildren()) {
+      lookupTable.put(child.getFullName(), child);
+      buildLookupTable(child);
+    }
+  }
+
+  @SuppressWarnings({"rawtypes", "unchecked"})
+  private void wireUpInheritanceRelationships(final ClassHierarchyProto.Node n) {
+    if (n.hasClassNode()) {
+      final ClassHierarchyProto.ClassNode cn = n.getClassNode();
+      final ClassNode iface;
+      try {
+        iface = (ClassNode) getNode(n.getFullName());
+      } catch (NameResolutionException e) {
+        throw new IllegalStateException("When reading protocol buffer node "
+            + n.getFullName() + " does not exist.  Full record is " + n, e);
+      }
+      for (String impl : cn.getImplFullNamesList()) {
+        try {
+          iface.putImpl((ClassNode) getNode(impl));
+        } catch (NameResolutionException e) {
+          throw new IllegalStateException("When reading protocol buffer node "
+              + n + " refers to non-existent implementation:" + impl);
+        } catch (ClassCastException e) {
+          try {
+            throw new IllegalStateException(
+                "When reading protocol buffer node " + n
+                    + " found implementation" + getNode(impl)
+                    + " which is not a ClassNode!");
+          } catch (NameResolutionException e2) {
+            throw new IllegalStateException(
+                "Got 'cant happen' exception when producing error message for "
+                    + e);
+          }
+        }
+      }
+    }
+
+    for (ClassHierarchyProto.Node child : n.getChildrenList()) {
+      wireUpInheritanceRelationships(child);
+    }
+  }
+
+  @Override
+  public Node getNode(String fullName) throws NameResolutionException {
+
+    Node ret = lookupTable.get(fullName);
+/*    String[] tok = fullName.split(regex);
+
+    Node ret = namespace.get(fullName);
+    for (int i = 0; i < tok.length; i++) {
+      Node n = namespace.get(getNthPrefix(fullName, i));
+      if (n != null) {
+        for (i++; i < tok.length; i++) {
+          n = n.get(tok[i]);
+          if (n == null) {
+            throw new NameResolutionException(fullName, getNthPrefix(fullName,
+                i - 1));
+          }
+        }
+        return n;
+      }
+    } */
+    if (ret != null) {
+      return ret;
+    } else {
+      throw new NameResolutionException(fullName, "");
+    }
+  }
+
+  @Override
+  public boolean isImplementation(ClassNode<?> inter, ClassNode<?> impl) {
+    return impl.isImplementationOf(inter);
+  }
+
+  @Override
+  public ClassHierarchy merge(ClassHierarchy ch) {
+    if (this == ch) {
+      return this;
+    }
+    throw new UnsupportedOperationException(
+        "Cannot merge ExternalClassHierarchies yet!");
+  }
+
+  @Override
+  public Node getNamespace() {
+    return namespace;
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/protobuf/ProtocolBufferInjectionPlan.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/protobuf/ProtocolBufferInjectionPlan.java b/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/protobuf/ProtocolBufferInjectionPlan.java
new file mode 100644
index 0000000..9615d52
--- /dev/null
+++ b/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/protobuf/ProtocolBufferInjectionPlan.java
@@ -0,0 +1,148 @@
+/**
+ * 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
+ *
+ *   http://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.reef.tang.implementation.protobuf;
+
+import org.apache.reef.tang.ClassHierarchy;
+import org.apache.reef.tang.exceptions.BindException;
+import org.apache.reef.tang.exceptions.NameResolutionException;
+import org.apache.reef.tang.implementation.Constructor;
+import org.apache.reef.tang.implementation.InjectionPlan;
+import org.apache.reef.tang.implementation.Subplan;
+import org.apache.reef.tang.implementation.java.JavaInstance;
+import org.apache.reef.tang.proto.InjectionPlanProto;
+import org.apache.reef.tang.types.ClassNode;
+import org.apache.reef.tang.types.ConstructorDef;
+import org.apache.reef.tang.types.Node;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class ProtocolBufferInjectionPlan {
+
+  <T> InjectionPlanProto.InjectionPlan newConstructor(final String fullName,
+                                                      List<InjectionPlanProto.InjectionPlan> plans) {
+    return InjectionPlanProto.InjectionPlan
+        .newBuilder()
+        .setName(fullName)
+        .setConstructor(
+            InjectionPlanProto.Constructor.newBuilder().addAllArgs(plans)
+                .build()).build();
+  }
+
+  <T> InjectionPlanProto.InjectionPlan newSubplan(final String fullName,
+                                                  int selectedPlan, List<InjectionPlanProto.InjectionPlan> plans) {
+    return InjectionPlanProto.InjectionPlan
+        .newBuilder()
+        .setName(fullName)
+        .setSubplan(
+            InjectionPlanProto.Subplan.newBuilder()
+                .setSelectedPlan(selectedPlan).addAllPlans(plans).build())
+        .build();
+  }
+
+  <T> InjectionPlanProto.InjectionPlan newInstance(final String fullName,
+                                                   final String value) {
+    return InjectionPlanProto.InjectionPlan.newBuilder().setName(fullName)
+        .setInstance(InjectionPlanProto.Instance.newBuilder().setValue(value))
+        .build();
+  }
+
+  public <T> InjectionPlanProto.InjectionPlan serialize(InjectionPlan<T> ip) {
+    if (ip instanceof Constructor) {
+      Constructor<T> cons = (Constructor<T>) ip;
+      InjectionPlan<?>[] args = cons.getArgs();
+      InjectionPlanProto.InjectionPlan[] protoArgs = new InjectionPlanProto.InjectionPlan[args.length];
+      for (int i = 0; i < args.length; i++) {
+        protoArgs[i] = serialize(args[i]);
+      }
+      return newConstructor(ip.getNode().getFullName(),
+          Arrays.asList(protoArgs));
+    } else if (ip instanceof Subplan) {
+      Subplan<T> sp = (Subplan<T>) ip;
+      InjectionPlan<?>[] args = sp.getPlans();
+      InjectionPlanProto.InjectionPlan[] subPlans = new InjectionPlanProto.InjectionPlan[args.length];
+      for (int i = 0; i < args.length; i++) {
+        subPlans[i] = serialize(args[i]);
+      }
+      return newSubplan(ip.getNode().getFullName(), sp.getSelectedIndex(),
+          Arrays.asList(subPlans));
+    } else if (ip instanceof JavaInstance) {
+      JavaInstance<T> ji = (JavaInstance<T>) ip;
+      return newInstance(ip.getNode().getFullName(), ji.getInstanceAsString());
+    } else {
+      throw new IllegalStateException(
+          "Encountered unknown type of InjectionPlan: " + ip);
+    }
+  }
+
+  private Object parse(String type, String value) {
+    // XXX this is a placeholder for now.  We need a parser API that will
+    // either produce a live java object or (partially) validate stuff to
+    // see if it looks like the target language will be able to handle this
+    // type + value.
+    return value;
+  }
+
+  @SuppressWarnings("unchecked")
+  public <T> InjectionPlan<T> deserialize(ClassHierarchy ch,
+                                          InjectionPlanProto.InjectionPlan ip) throws NameResolutionException,
+      BindException {
+    final String fullName = ip.getName();
+    if (ip.hasConstructor()) {
+      final InjectionPlanProto.Constructor cons = ip.getConstructor();
+
+      final ClassNode<T> cn = (ClassNode<T>) ch.getNode(fullName);
+
+      final InjectionPlanProto.InjectionPlan protoBufArgs[] = cons
+          .getArgsList().toArray(new InjectionPlanProto.InjectionPlan[0]);
+      final ClassNode<?>[] cnArgs = new ClassNode[protoBufArgs.length];
+
+      for (int i = 0; i < protoBufArgs.length; i++) {
+        cnArgs[i] = (ClassNode<?>) ch.getNode(protoBufArgs[i].getName());
+      }
+
+      final InjectionPlan<?> ipArgs[] = new InjectionPlan[protoBufArgs.length];
+
+      for (int i = 0; i < protoBufArgs.length; i++) {
+        ipArgs[i] = (InjectionPlan<?>) deserialize(ch, protoBufArgs[i]);
+      }
+
+      final ConstructorDef<T> constructor = cn.getConstructorDef(cnArgs);
+      return new Constructor<T>(cn, constructor, ipArgs);
+    } else if (ip.hasInstance()) {
+      InjectionPlanProto.Instance ins = ip.getInstance();
+      T instance = (T) parse(ip.getName(), ins.getValue());
+      return new JavaInstance<T>(ch.getNode(ip.getName()), instance);
+    } else if (ip.hasSubplan()) {
+      final InjectionPlanProto.Subplan subplan = ip.getSubplan();
+      final InjectionPlanProto.InjectionPlan protoBufPlans[] = subplan
+          .getPlansList().toArray(new InjectionPlanProto.InjectionPlan[0]);
+
+      final InjectionPlan<T> subPlans[] = new InjectionPlan[protoBufPlans.length];
+      for (int i = 0; i < protoBufPlans.length; i++) {
+        subPlans[i] = (InjectionPlan<T>) deserialize(ch, protoBufPlans[i]);
+      }
+      Node n = ch.getNode(fullName);
+      return new Subplan<T>(n, subPlans);
+    } else {
+      throw new IllegalStateException(
+          "Encountered unknown type of injection plan: " + ip);
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/protobuf/package-info.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/protobuf/package-info.java b/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/protobuf/package-info.java
new file mode 100644
index 0000000..da901a2
--- /dev/null
+++ b/lang/java/reef-tang/tang/src/main/java/org/apache/reef/tang/implementation/protobuf/package-info.java
@@ -0,0 +1,25 @@
+/**
+ * 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
+ *
+ *   http://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.
+ */
+/**
+ * Implementation classes that translate between Tang's core API and protocol
+ * buffers.  This enables cross-language injection sessions.
+ */
+
+package org.apache.reef.tang.implementation.protobuf;
+