Modifies the Counter module to allow pipe lining

Additionally, this commits makes the module available in Flux. That
is not really useful but helps finding the module.

(cherry picked from commit eaf9153)

Author: cboehme

src/main/java/org/culturegraph/mf/stream/sink/Counter.java renamed to src/main/java/org/culturegraph/mf/stream/pipe/Counter.java

@@ -13,25 +13,33 @@
  *  See the License for the specific language governing permissions and
  *  limitations under the License.
  */
-package org.culturegraph.mf.stream.sink;
+package org.culturegraph.mf.stream.pipe;
 
-import org.culturegraph.mf.framework.DefaultStreamReceiver;
+import org.culturegraph.mf.framework.DefaultStreamPipe;
+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;
 
 
 /**
  * Counts the number of records and fields read. Used mainly for test cases and
  * debugging.
- * 
- * @author Markus Michael Geipel, Christoph Böhme
- * 
+ *
+ * @author Markus Michael Geipel
+ * @author Christoph Böhme
+ * @author Michael Büchner
+ *
  */
-public final class Counter extends DefaultStreamReceiver {
-	
-	private boolean closed;
+@Description("Counts the number of records and fields read.")
+@In(StreamReceiver.class)
+@Out(StreamReceiver.class)
+public final class Counter extends DefaultStreamPipe<StreamReceiver> {
+
 	private int numRecords;
 	private int numEntities;
 	private int numLiterals;
-	
+
 	/**
 	 * @return the numRecords
 	 */
@@ -52,53 +60,62 @@ public int getNumEntities() {
 	public int getNumLiterals() {
 		return numLiterals;
 	}
-	
-	/**
-	 * 
-	 * @return true if the stream is closed
-	 */
-	public boolean isClosed() {
-		return closed;
-	}
 
 	@Override
 	public void startRecord(final String identifier) {
-		assert !closed;
+		assert !isClosed();
 		++numRecords;
+		if(getReceiver() != null) {
+			getReceiver().startRecord(identifier);
+		}
 	}
 
 	@Override
 	public void startEntity(final String name) {
-		assert !closed;
+		assert !isClosed();
 		++numEntities;
+		if(getReceiver() != null) {
+			getReceiver().startEntity(name);
+		}
 	}
 
 	@Override
 	public void literal(final String name, final String value) {
-		assert !closed;
+		assert !isClosed();
 		++numLiterals;
+		if(getReceiver() != null) {
+			getReceiver().literal(name, value);
+		}
+	}
+
+	@Override
+	public void endRecord() {
+		if(getReceiver() != null) {
+			getReceiver().endRecord();
+		}
 	}
 
 	@Override
-	public void resetStream() {
-		closed = false;
+	public void endEntity() {
+		if(getReceiver() != null) {
+			getReceiver().endEntity();
+		}
+	}
+
+	@Override
+	public void onResetStream() {
 		numRecords = 0;
 		numEntities = 0;
 		numLiterals = 0;
 	}
-	
-	@Override
-	public void closeStream() {
-		closed = true;
-	}
-	
+
 	@Override
 	public String toString() {
 		String streamClosed = "";
-		if (closed) {
+		if (isClosed()) {
 			streamClosed =" Stream has been closed.";
 		}
-		
+
 		return "counted " + numRecords + " records, " + numEntities + " entities, " + numLiterals + " literals." + streamClosed;
 	}
 }

src/main/resources/flux-commands.properties

@@ -84,6 +84,8 @@ generic-xml org.culturegraph.mf.stream.reader.GenericXmlReader
 
 match org.culturegraph.mf.stream.pipe.StringMatcher
 
+stream-count org.culturegraph.mf.stream.pipe.Counter
+
 extract-literals org.culturegraph.mf.stream.converter.LiteralExtractor
 object-to-literal org.culturegraph.mf.stream.converter.ObjectToLiteral
 

src/test/java/org/culturegraph/mf/stream/converter/bib/SimpleMabReaderTest.java

@@ -25,8 +25,7 @@
 import org.culturegraph.mf.exceptions.FormatException;
 import org.culturegraph.mf.stream.DataFilePath;
 import org.culturegraph.mf.stream.converter.LineReader;
-import org.culturegraph.mf.stream.converter.bib.MabDecoder;
-import org.culturegraph.mf.stream.sink.Counter;
+import org.culturegraph.mf.stream.pipe.Counter;
 import org.culturegraph.mf.stream.sink.StreamValidator;
 import org.culturegraph.mf.stream.source.ResourceOpener;
 import org.culturegraph.mf.types.Event;
@@ -40,58 +39,58 @@
  * @see MabDecoder
  */
 public final class SimpleMabReaderTest {
-	
+
 	private static final int NUM_RECORDS = 10;
 	private static final int NUM_LITERALS = 520;
-	
+
 
 	@Test
 	public void testRead(){
 		final ResourceOpener opener = new ResourceOpener();
 		final Counter countStreamReceiver = opener.setReceiver(new LineReader())
 				.setReceiver(new MabDecoder())
 				.setReceiver(new Counter());
-		
+
 		opener.process(DataFilePath.TITLE_MAB);
 		opener.closeStream();
-		
+
 		Assert.assertEquals("Number of read records is incorrect", NUM_RECORDS, countStreamReceiver.getNumRecords());
 		Assert.assertEquals("Number of read literals is incorrect", NUM_LITERALS, countStreamReceiver.getNumLiterals());
 	}
-	
+
 	@Test
 	public void testGetId() throws IOException {
 		final InputStream inputStream = Thread.currentThread()
 				.getContextClassLoader().getResourceAsStream(DataFilePath.TITLE_MAB);
-		
+
 		final BufferedReader breader = new BufferedReader(new InputStreamReader(inputStream));
-		
+
 		String line = breader.readLine();
 		while (line != null) {
 			if(!line.isEmpty()){
 				Assert.assertNotNull(MabDecoder.extractIdFromRecord(line));
 			}
 			line = breader.readLine();
 		}
-		
+
 		breader.close();
 	}
-	
+
 	@Test
 	public void testSkipEmptyStrings() {
 		final MabDecoder decoder = new MabDecoder();
-		
+
 		final List<Event> expected = Collections.emptyList();
 		final StreamValidator validator = new StreamValidator(expected);
-		
+
 		decoder.setReceiver(validator);
-		
+
 		try {
 			decoder.process(" ");
 			decoder.closeStream();
-		} catch (FormatException e) {
+		} catch (final FormatException e) {
 			Assert.fail(e.toString());
 		}
 	}
-	
+
 }

src/test/java/org/culturegraph/mf/stream/converter/bib/SimpleMarcReaderTest.java

@@ -25,9 +25,7 @@
 import org.culturegraph.mf.exceptions.FormatException;
 import org.culturegraph.mf.stream.DataFilePath;
 import org.culturegraph.mf.stream.converter.LineReader;
-import org.culturegraph.mf.stream.converter.bib.MabDecoder;
-import org.culturegraph.mf.stream.converter.bib.MarcDecoder;
-import org.culturegraph.mf.stream.sink.Counter;
+import org.culturegraph.mf.stream.pipe.Counter;
 import org.culturegraph.mf.stream.sink.StreamValidator;
 import org.culturegraph.mf.stream.source.ResourceOpener;
 import org.culturegraph.mf.types.Event;
@@ -41,34 +39,34 @@
  * @see MabDecoder
  */
 public final class SimpleMarcReaderTest {
-	
+
 	private static final int NUM_RECORDS = 10;
 	private static final int NUM_LITERALS = 745;
-	
 
 
-		
+
+
 	@Test
 	public void testRead(){
 		final ResourceOpener opener = new ResourceOpener();
 
 		final Counter countStreamReceiver = opener.setReceiver(new LineReader())
 				.setReceiver(new MarcDecoder())
 				.setReceiver(new Counter());
-		
+
 		opener.process(DataFilePath.TITLE_MARC);
 		opener.closeStream();
-		
+
 		Assert.assertEquals("Number of read records is incorrect", NUM_RECORDS, countStreamReceiver.getNumRecords());
 		Assert.assertEquals("Number of read literals is incorrect", NUM_LITERALS, countStreamReceiver.getNumLiterals());
 	}
-	
+
 	@Test
 	public void testGetId() throws IOException {
 		final InputStream inputStream = Thread.currentThread()
 				.getContextClassLoader().getResourceAsStream(DataFilePath.TITLE_MARC);
 		final BufferedReader breader = new BufferedReader(new InputStreamReader(inputStream));
-		
+
 		String line = breader.readLine();
 		while (line != null) {
 			if(!line.isEmpty()){
@@ -78,22 +76,22 @@ public void testGetId() throws IOException {
 		}
 		breader.close();
 	}
-	
+
 	@Test
 	public void testSkipEmptyStrings() {
 		final MarcDecoder decoder = new MarcDecoder();
-		
+
 		final List<Event> expected = Collections.emptyList();
 		final StreamValidator validator = new StreamValidator(expected);
-		
+
 		decoder.setReceiver(validator);
-		
+
 		try {
 			decoder.process(" ");
 			decoder.closeStream();
-		} catch (FormatException e) {
+		} catch (final FormatException e) {
 			Assert.fail(e.toString());
 		}
 	}
-	
+
 }

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

@@ -23,15 +23,14 @@
 import org.culturegraph.mf.morph.DataFilePath;
 import org.culturegraph.mf.stream.reader.PicaReader;
 import org.culturegraph.mf.stream.reader.Reader;
-import org.culturegraph.mf.stream.sink.Counter;
 import org.junit.Test;
 
 
 /**
  * Tests {@link Splitter}.
- * 
+ *
  * @author Markus Michael Geipel
- * 
+ *
  */
 public final class SplitterTest {
 
@@ -56,5 +55,5 @@ public void testCorrectTeeFunction() throws IOException {
 		assertEquals(NUM_TN_RECORDS, countingWriterTn.getNumRecords());
 		assertEquals(NUM_TP_RECORDS, countingWriterTp.getNumRecords());
 	}
-	
+
 }