Linkage { public static void link(Step source, Step stage, String fieldName) throws IncompatibleProcessorException { try { Class sourceClass = source.getClass(); Field processorField = sourceClass.getField(fieldName); Class fieldClass = processorField.getType(); if (fieldClass.isInstance(stage)) { processorField.set(source, stage); return; } Class stageClass = stage.getClass(); for (Class sourceInterfaceClass : sourceClass.getInterfaces()) { String interfaceName = sourceInterfaceClass.getName(); String typeName; Object adapted = null; try { if ((typeName = Utility.strip(interfaceName, "$Source")) != null) { done: for (Class stageInterfaceClass : stageClass.getInterfaces()) { String stageInterfaceName = stageInterfaceClass.getName(); if (stageInterfaceName.startsWith(typeName) && stageInterfaceName.endsWith("$ShreddedProcessor")) { String stageOrderName = Utility.strip(stageInterfaceName, "$ShreddedProcessor"); Constructor[] constructors = Class.forName( stageOrderName + "$TupleShredder").getConstructors(); for (Constructor c : constructors) { Class[] types = c.getParameterTypes(); if (types.length == 1 && types[0].isInstance(stage)) { adapted = c.newInstance(stage); break done; } } } } } else if (ShreddedSource.class.isAssignableFrom(sourceInterfaceClass) || (typeName = Utility.strip(interfaceName, "$ShreddedSource")) != null) { Constructor[] constructors = Class.forName(typeName + "$TupleUnshredder"). getConstructors(); for (Constructor c : constructors) { Class[] types = c.getParameterTypes(); if (types.length == 1 && types[0].isInstance(stage)) { adapted = c.newInstance(stage); break; } } } } catch (Exception e) { continue; } if (adapted != null && fieldClass.isInstance(adapted)) { processorField.set(source, adapted); return; } } throw new IncompatibleProcessorException("Stage of type '" + stage.getClass().getName() + "' cannot process the objects that '" + sourceClass.getName() + "' produces."); } catch (NoSuchFieldException e) { throw new IncompatibleProcessorException( "Stage of type '" + source.getClass().getName() + "' has no field called '" + fieldName + "' that can hold a processor object (or maybe it's just not public)."); } catch (IllegalAccessException e) { throw new IncompatibleProcessorException( "Stage of type '" + source.getClass().getName() + "' has a field called '" + fieldName + "', but " + "it is private or protected and can't be modified by link()."); } }  static void link(Step source, Step stage, String fieldName); static void link(Step source, Step stage); static void close(Step stage);  }
@Test public void testLinkage() { Step start = new Source<TupleflowString>() { Processor<TupleflowString> processor; @Override public void setProcessor(Step processor) throws IncompatibleProcessorException { Linkage.link(this, processor); } }; }