quick fix for #108.

Author: mgeipel

examples/beacon/create/create.flux

@@ -1,8 +1,9 @@
 //creates a beacon file based on a pica+ dump of the DNB CBS data.
 
+default type = "ALL";
 default out = dump + "-" + type + ".beacon";
 default header = FLUX_DIR + "header.txt";
-default type = "ALL";
+
 
 //read header
 "reading header " + header | write("stdout");
@@ -28,6 +29,7 @@ template("${s}")|
 @Y;
 
 @Y|
+wait-for-inputs("2")|
 write(out);
 
 

src/main/java/org/culturegraph/mf/stream/pipe/CloseSupressor.java

@@ -0,0 +1,77 @@
+/*
+ *  Copyright 2013 Deutsche Nationalbibliothek
+ *
+ *  Licensed 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.culturegraph.mf.stream.pipe;
+
+import org.culturegraph.mf.framework.ObjectPipe;
+import org.culturegraph.mf.framework.ObjectReceiver;
+import org.culturegraph.mf.framework.annotations.Description;
+import org.culturegraph.mf.framework.annotations.In;
+import org.culturegraph.mf.framework.annotations.Out;
+
+/**
+ * 
+ * @param <T>
+ *            object type
+ * 
+ * @author Markus Geipel
+ * 
+ */
+@Description("")
+@In(Object.class)
+@Out(Object.class)
+public final class CloseSupressor<T> implements ObjectPipe<T, ObjectReceiver<T>> {
+
+	private ObjectReceiver<T> receiver;
+	private final int numCloses;
+	private int count;
+
+	public CloseSupressor(final int numCloses) {
+		this.numCloses = numCloses;
+	}
+	
+	public CloseSupressor(final String numCloses) {
+		this.numCloses = Integer.parseInt(numCloses);
+	}
+
+	@Override
+	public void process(final T obj) {
+		if (receiver != null) {
+			receiver.process(obj);
+		}
+	}
+
+	@Override
+	public <R extends ObjectReceiver<T>> R setReceiver(final R receiver) {
+		this.receiver = receiver;
+		return receiver;
+	}
+
+	@Override
+	public void resetStream() {
+		count = 0;
+		if (receiver != null) {
+			receiver.resetStream();
+		}
+	}
+
+	@Override
+	public void closeStream() {
+		++count;
+		if (count == numCloses && receiver != null) {
+			receiver.closeStream();
+		}
+	}
+}

src/main/resources/flux-commands.properties

@@ -16,6 +16,8 @@ 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
 
+calculate-metrics org.culturegraph.mf.stream.pipe.stat.CooccurrenceMetricCalculator
+
 jscript org.culturegraph.mf.stream.pipe.JScriptObjectPipe
 
 as-lines	org.culturegraph.mf.stream.converter.LineReader
@@ -66,6 +68,7 @@ filter-duplicate-objects org.culturegraph.mf.stream.pipe.DuplicateObjectFilter
 
 object-tee org.culturegraph.mf.stream.pipe.ObjectTee
 stream-tee org.culturegraph.mf.stream.pipe.StreamTee
+wait-for-inputs	org.culturegraph.mf.stream.pipe.CloseSupressor
 
 stream-to-xml	org.culturegraph.mf.stream.sink.SimpleXmlWriter
 rdf-macros	org.culturegraph.mf.stream.pipe.RdfMacroPipe

src/test/java/org/culturegraph/mf/stream/pipe/CloseSuppressorTest.java

@@ -0,0 +1,45 @@
+/*
+ *  Copyright 2013 Deutsche Nationalbibliothek
+ *
+ *  Licensed 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.culturegraph.mf.stream.pipe;
+
+import org.culturegraph.mf.framework.ObjectReceiver;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+/**
+ * Tests for {@link CloseSupressor}.
+ * 
+ * @author Markus Geipel
+ */
+public final class CloseSuppressorTest {
+
+	@SuppressWarnings("unchecked")
+	@Test
+	public void testSuppression() {
+
+		final CloseSupressor<Object> supressor = new CloseSupressor<Object>(3);
+		final ObjectReceiver<Object> receiver = Mockito.mock(ObjectReceiver.class);
+		supressor.setReceiver(receiver);
+		supressor.closeStream();
+		supressor.closeStream();
+		Mockito.verifyZeroInteractions(receiver);
+		supressor.closeStream();
+		Mockito.verify(receiver).closeStream();
+		supressor.closeStream();
+		Mockito.verifyNoMoreInteractions(receiver);
+	}
+
+}