Merge pull request #170 from schaeferd/triples-to-stream

adds triple-to-stream in order to process viaf-dump data for entityfacts

Author: cboehme

src/main/java/org/culturegraph/mf/stream/converter/TriplesToStream.java

@@ -0,0 +1,51 @@
+/**
+ * 
+ */
+package org.culturegraph.mf.stream.converter;
+
+import org.culturegraph.mf.formeta.parser.FormetaParser;
+import org.culturegraph.mf.formeta.parser.PartialRecordEmitter;
+import org.culturegraph.mf.framework.DefaultObjectPipe;
+import org.culturegraph.mf.framework.StreamReceiver;
+import org.culturegraph.mf.framework.annotations.Description;
+import org.culturegraph.mf.framework.annotations.In;
+import org.culturegraph.mf.framework.annotations.Out;
+import org.culturegraph.mf.types.Triple;
+import org.culturegraph.mf.types.Triple.ObjectType;
+
+/**
+ * @author schaeferd
+ *
+ */
+@Description("Converts a triple into a record stream")
+@In(Triple.class)
+@Out(StreamReceiver.class)
+public final class TriplesToStream extends
+		DefaultObjectPipe<Triple, StreamReceiver> {
+	
+	private final FormetaParser parser = new FormetaParser();
+	private final PartialRecordEmitter emitter = new PartialRecordEmitter();
+
+	public TriplesToStream() {
+		parser.setEmitter(emitter);
+	}
+	
+	public void process(final Triple triple) {
+		getReceiver().startRecord(triple.getSubject());
+		if(triple.getObjectType() == ObjectType.STRING){
+			getReceiver().literal(triple.getPredicate(), triple.getObject());
+		}else if (triple.getObjectType() == ObjectType.ENTITY){
+			emitter.setDefaultName(triple.getPredicate());
+			parser.parse(triple.getObject());
+		}else{
+			throw new UnsupportedOperationException(triple.getObjectType() + " can not yet be decoded");
+		}
+		getReceiver().endRecord();
+	}
+	
+	@Override
+	protected void onSetReceiver() {
+		emitter.setReceiver(getReceiver());
+	}
+
+}

src/main/resources/flux-commands.properties

@@ -15,7 +15,7 @@ count-triples org.culturegraph.mf.stream.pipe.sort.TripleCount
 collect-triples org.culturegraph.mf.stream.pipe.sort.TripleCollect
 stream-to-triples org.culturegraph.mf.stream.converter.StreamToTriples
 filter-triples org.culturegraph.mf.stream.pipe.TripleFilter
-
+triples-to-stream org.culturegraph.mf.stream.converter.TriplesToStream
 calculate-metrics org.culturegraph.mf.stream.pipe.stat.CooccurrenceMetricCalculator
 
 jscript org.culturegraph.mf.stream.pipe.JScriptObjectPipe