Refactored the code for building pipelines in Metamorph

In the original implementation of Metamorph `Data` and `Collect` objects
managed the plumbing of their function pipelines internally (via
`AbstractNamedValuePipeHead`). From the perspective of the
`MorphBuilder` `Data` and `Collect` objects were connected directly. The
fact that the connection was actually done through the function pipeline
was hidden. While this approach simplified the `MorphBuilder`, it
increased the complexity of the interfaces for connecting the different
types of objects in Metamorph. Additionally, functions were required to
handle the source parameter in the `receive` method differently compared
to the implementations of the method in `Data` or `Collect` objects.
Finally, it the half-hidden implementation of the function pipelines
makes it very hard to add interceptors (e.g. for logging or debugging)
to the pipe system generated from a Metamorph script.

This commit tries to solve the short-comings of the original
implementation with to changes. First, connections of `Function`, `Data`
and `Collect` objects are now handled in the same way and the function
pipelines are no longer hidden in the other two objects. Second, the
objects are no longer connected to a receiver (via
`setNamedValueReceiver`) but the connection is set the other way round
using `addNamedValueSource`. While this approach appears slightly less
intutive than the original method, it removes the necessity for
calling `addNamedValueSource` in addition to `setNamedValueReceiver`.
Furthermore it unifies setting of an object as name source or
condition source and setting an object as standard data source. This is
now all done from the perspective of the receiver. This simplifies the
implementation of the relevant methods in the collectors.

Author: cboehme

src/main/java/org/culturegraph/mf/morph/AbstractNamedValuePipeHead.java renamed to src/main/java/org/culturegraph/mf/morph/AbstractNamedValuePipe.java

@@ -18,35 +18,35 @@
 /**
  * Base class for {@link NamedValuePipe}s
  * @author Markus Michael Geipel
+ * @author Chistohp Böhme
  *
  */
-public abstract class AbstractNamedValuePipeHead implements NamedValuePipeHead {
+public abstract class AbstractNamedValuePipe implements NamedValuePipe {
 
 	private NamedValueReceiver namedValueReceiver;
-	private NamedValuePipe last = this;
-
-	protected final NamedValueReceiver getNamedValueReceiver() {
-		return namedValueReceiver;
-	}
 
 	@Override
-	public final <R extends NamedValueReceiver> R setNamedValueReceiver(final R receiver) {
-		this.namedValueReceiver = receiver;
+	public final <R extends NamedValueReceiver> R setNamedValueReceiver(
+			final R receiver) {
+
+		namedValueReceiver = receiver;
+
 		return receiver;
 	}
 
 	@Override
-	public final void appendPipe(final NamedValuePipe namedValuePipe) {
-		if(last==null){
-			throw new IllegalStateException("NamedValuePipe already finalyzed.");
-		}
-		last = last.setNamedValueReceiver(namedValuePipe);
+	public final void addNamedValueSource(final NamedValueSource namedValueSource) {
+		namedValueSource.setNamedValueReceiver(this);
+
+		onNamedValueSourceAdded(namedValueSource);
 	}
 
-	@Override
-	public final void endPipe(final NamedValueReceiver lastReceiver) {
-		last.setNamedValueReceiver(lastReceiver);
-		last = null;
+	protected void onNamedValueSourceAdded(final NamedValueSource namedValueSource) {
+		// Default implementation does nothing
+	}
+
+	protected final NamedValueReceiver getNamedValueReceiver() {
+		return namedValueReceiver;
 	}
 
 }

src/main/java/org/culturegraph/mf/morph/Data.java

@@ -22,7 +22,7 @@
  *
  * @author Markus Michael Geipel
  */
-final class Data extends AbstractNamedValuePipeHead {
+final class Data extends AbstractNamedValuePipe {
 
 	private String name;
 

src/main/java/org/culturegraph/mf/morph/Flush.java

@@ -36,4 +36,9 @@ public void receive(final String name, final String value, final NamedValueSourc
 		listener.flush(recordCount, entityCount);
 	}
 
+	@Override
+	public void addNamedValueSource(final NamedValueSource namedValueSource) {
+		// Nothing to do
+	}
+
 }

src/main/java/org/culturegraph/mf/morph/Metamorph.java

@@ -44,7 +44,7 @@
 		+ "definition is given in brackets.")
 @In(StreamReceiver.class)
 @Out(StreamReceiver.class)
-public final class Metamorph implements StreamPipe<StreamReceiver>, NamedValueReceiver, MultiMap {
+public final class Metamorph implements StreamPipe<StreamReceiver>, NamedValuePipe, MultiMap {
 
 	public static final String ELSE_KEYWORD = "_else";
 	public static final char FEEDBACK_CHAR = '@';
@@ -322,4 +322,14 @@ public void registerRecordEndFlush(final FlushListener flushListener) {
 		recordEndListener.add(flushListener);
 	}
 
+	@Override
+	public void addNamedValueSource(final NamedValueSource namedValueSource) {
+		namedValueSource.setNamedValueReceiver(this);
+	}
+
+	@Override
+	public <R extends NamedValueReceiver> R setNamedValueReceiver(final R receiver) {
+		throw new UnsupportedOperationException("The Metamorph object cannot act as a NamedValueSender");
+	}
+
 }

src/main/java/org/culturegraph/mf/morph/MorphBuilder.java

@@ -32,6 +32,8 @@
  * Builds a {@link Metamorph} from an xml description
  *
  * @author Markus Michael Geipel
+ * @author Christoph Böhme
+ *
  */
 public final class MorphBuilder extends AbstractMetamorphDomWalker {
 
@@ -43,15 +45,49 @@ public final class MorphBuilder extends AbstractMetamorphDomWalker {
 	private static final Pattern OR_PATTERN = Pattern.compile(OR_STRING, Pattern.LITERAL);
 
 	private final Metamorph metamorph;
-	private final Deque<Collect> collectStack;
-	private Data data;
-	private boolean setEntityName;
-	private boolean setCondition;
+	private final Deque<StackFrame> stack = new LinkedList<StackFrame>();
+
+	private static final class StackFrame {
+
+		private NamedValuePipe pipe;
+		private boolean inEntityName;
+		private boolean inCondition;
+
+		public StackFrame(final NamedValuePipe pipe) {
+			this.pipe = pipe;
+		}
+
+		public void setPipe(final NamedValuePipe pipe) {
+			this.pipe = pipe;
+		}
+
+		public NamedValuePipe getPipe() {
+			return pipe;
+		}
+
+		public void setInEntityName(final boolean inEntityName) {
+			this.inEntityName = inEntityName;
+		}
+
+		public boolean isInEntityName() {
+			return inEntityName;
+		}
+
+		public void setInCondition(final boolean inCondition) {
+			this.inCondition = inCondition;
+		}
+
+		public boolean isInCondition() {
+			return inCondition;
+		}
+
+	}
 
 	protected MorphBuilder(final Metamorph metamorph) {
 		super();
-		this.collectStack = new LinkedList<Collect>();
+
 		this.metamorph = metamorph;
+		stack.push(new StackFrame(metamorph));
 	}
 
 	@Override
@@ -135,52 +171,55 @@ protected void finish() {
 
 	@Override
 	protected void enterData(final Node dataNode) {
-		final String source = resolvedAttribute(dataNode, ATTRITBUTE.SOURCE);
-		data = new Data();
+		final Data data = new Data();
 		data.setName(resolvedAttribute(dataNode, ATTRITBUTE.NAME));
+
+		final String source = resolvedAttribute(dataNode, ATTRITBUTE.SOURCE);
 		metamorph.registerNamedValueReceiver(source, data);
 
-		if (setEntityName) {
-			((Entity) collectStack.peek()).setNameSource(data);
-			setEntityName = false;
-		}
-	
-		if (setCondition) {
-			collectStack.peek().setConditionSource(data);
-			setCondition = false;
-		}
+		stack.push(new StackFrame(data));
 	}
 
 	@Override
 	protected void exitData(final Node node) {
-		if (collectStack.isEmpty()) {
-			data.endPipe(metamorph);
+		final NamedValuePipe dataPipe = stack.pop().getPipe();
+
+		final StackFrame parent = stack.peek();
+		if (parent.isInEntityName()) {
+			// Protected xsd schema and by assertion in enterName:
+			((Entity) parent.getPipe()).setNameSource(dataPipe);
+		} else if (parent.isInCondition()) {
+			// Protected xsd schema and by assertion in enterIf:
+			((ConditionAware) parent.getPipe()).setConditionSource(dataPipe);
 		} else {
-			final Collect parent = collectStack.peek();
-			data.endPipe(parent);
-			parent.addNamedValueSource(data);
+			parent.getPipe().addNamedValueSource(dataPipe);
 		}
-		data = null;
 	}
 
 	@Override
 	protected void enterName(final Node nameNode) {
-		setEntityName = true;
+		assert stack.peek().getPipe() instanceof Entity :
+				"statement `name` is only allowed in `entity` statements";
+
+		stack.peek().setInEntityName(true);
 	}
 
 	@Override
 	protected void exitName(final Node nameNode) {
-		setEntityName = false;
+		stack.peek().setInEntityName(false);
 	}
 
 	@Override
 	protected void enterIf(final Node nameNode) {
-		setCondition = true;
+		assert stack.peek().getPipe() instanceof ConditionAware :
+				"statement `if` is not allowed in the current element";
+
+		stack.peek().setInCondition(true);
 	}
 
 	@Override
 	protected void exitIf(final Node nameNode) {
-		setCondition = false;
+		stack.peek().setInCondition(false);
 	}
 
 	@Override
@@ -194,34 +233,33 @@ protected void enterCollect(final Node node) {
 		}
 		final Collect collect = getCollectFactory().newInstance(node.getLocalName(), attributes, metamorph);
 
-		if (setEntityName) {
-			((Entity) collectStack.peek()).setNameSource(collect);
-			setEntityName = false;
-		}
-		
-		if (setCondition) {
-			collectStack.peek().setConditionSource(collect);
-			setCondition = false;
-		}
-
-		collectStack.push(collect);
+		stack.push(new StackFrame(collect));
 	}
 
 	@Override
 	protected void exitCollect(final Node node) {
-		final Collect collect = collectStack.pop();
-		if (collectStack.isEmpty()) {
-			collect.endPipe(metamorph);
+		final NamedValuePipe collectPipe = stack.pop().getPipe();
+
+		final StackFrame parent = stack.peek();
+
+		if (parent.isInEntityName()) {
+			// Protected xsd schema and by assertion in enterName:
+			((Entity) parent.getPipe()).setNameSource(collectPipe);
+		} else if (parent.isInCondition()) {
+			// Protected xsd schema and by assertion in enterIf:
+			((ConditionAware) parent.getPipe()).setConditionSource(collectPipe);
 		} else {
-			final Collect parent = collectStack.peek();
-			parent.addNamedValueSource(collect);
-			collect.endPipe(parent);
+			parent.getPipe().addNamedValueSource(collectPipe);
 		}
+
 		// must be set after recursive calls to flush descendants before parent
 		final String flushWith = resolvedAttribute(node, ATTRITBUTE.FLUSH_WITH);
 		if (null != flushWith) {
-			collect.setWaitForFlush(true);
-			registerFlush(flushWith, collect);
+			assert collectPipe instanceof Collect :
+					"Invokations of enterXXX and exitXXX are not properly balanced";
+
+			((Collect) collectPipe).setWaitForFlush(true);
+			registerFlush(flushWith, ((Collect) collectPipe));
 		}
 	}
 
@@ -265,11 +303,10 @@ protected void handleFunction(final Node functionNode) {
 			final String entryValue = resolvedAttribute(mapEntryNode, ATTRITBUTE.VALUE);
 			function.putValue(entryName, entryValue);
 		}
-		if (data == null) {
-			collectStack.peek().appendPipe(function);
-		} else {
-			data.appendPipe(function);
-		}
+
+		final StackFrame head = stack.peek();
+		function.addNamedValueSource(head.getPipe());
+		head.setPipe(function);
 	}
 
 }

src/main/java/org/culturegraph/mf/morph/NamedValuePipeHead.java

@@ -19,9 +19,13 @@
  * A receiver of named values.
  *
  * @author Markus Michael Geipel
+ * @author Christoph Böhme
+ *
  */
 public interface NamedValueReceiver {
 
 	void receive(String name, String value, NamedValueSource source, int recordCount, int entityCount);
 
+	void addNamedValueSource(final NamedValueSource namedValueSource);
+
 }

src/main/java/org/culturegraph/mf/morph/NamedValueReceiver.java

@@ -16,14 +16,25 @@
 package org.culturegraph.mf.morph;
 
 /**
- * Base interface for all classes in {@link Metamorph} which
- * emit name-value-pairs.
+ * Base interface for all classes in {@link Metamorph} which emit
+ * name-value-pairs.
  *
  * @author Markus Michael Geipel
+ * @author Christoph Böhme
  *
  */
 public interface NamedValueSource {
 
+	/**
+	 * Connects a source of named values to a receiver of named values.
+	 *
+	 * Users should not call this method to connect sources and
+	 * receivers but rather call
+	 * {@link NamedValueReceiver.addNamedValueSource}.
+	 *
+	 * @param receiver
+	 * @return reference to receiver
+	 */
 	<R extends NamedValueReceiver> R setNamedValueReceiver(R receiver);
 
 }