Fix #191: only emit endRecord() after triple was received

Only emit `endRecord()` in `resetStream()` and `closeStream()` if a
triple was received and a `startRecord()` event emitted.

Author: cboehme

src/main/java/org/culturegraph/mf/stream/pipe/sort/TripleCollect.java

@@ -1,5 +1,5 @@
 /*
- *  Copyright 2013, 2014 Deutsche Nationalbibliothek
+ *  Copyright 2013, 2014, 2016 Deutsche Nationalbibliothek
  *
  *  Licensed under the Apache License, Version 2.0 the "License";
  *  you may not use this file except in compliance with the License.
@@ -27,17 +27,18 @@
 
 /**
  * Collects named values to form records.
- * 
+ *
  * @author markus geipel
- * 
+ *
  */
 @Description("Collects named values to form records. The name becomes the id, the value is split by 'separator' into name and value")
 @In(Triple.class)
 @Out(StreamReceiver.class)
 public final class TripleCollect extends DefaultObjectPipe<Triple, StreamReceiver> {
+
 	private final FormetaParser parser = new FormetaParser();
 	private final PartialRecordEmitter emitter = new PartialRecordEmitter();
-	
+
 	private String currentSubject;
 
 	public TripleCollect() {
@@ -50,7 +51,7 @@ public void process(final Triple triple) {
 			currentSubject = triple.getSubject();
 			getReceiver().startRecord(currentSubject);
 		}
-		
+
 		if (currentSubject.equals(triple.getSubject())) {
 			decodeTriple(triple);
 		} else {
@@ -74,16 +75,20 @@ public void decodeTriple(final Triple triple) {
 
 	@Override
 	protected void onResetStream() {
-		currentSubject = null;
-		getReceiver().endRecord();
+		if (currentSubject != null) {
+			currentSubject = null;
+			getReceiver().endRecord();
+		}
 	}
 
 	@Override
 	protected void onCloseStream() {
-		currentSubject = null;
-		getReceiver().endRecord();
+		if (currentSubject != null) {
+			currentSubject = null;
+			getReceiver().endRecord();
+		}
 	}
-	
+
 	@Override
 	protected void onSetReceiver() {
 		emitter.setReceiver(getReceiver());

src/test/java/org/culturegraph/mf/stream/pipe/sort/TripleCollectTest.java

@@ -1,5 +1,5 @@
 /*
- *  Copyright 2013, 2014 Deutsche Nationalbibliothek
+ *  Copyright 2013, 2014, 2016 Deutsche Nationalbibliothek
  *
  *  Licensed under the Apache License, Version 2.0 the "License";
  *  you may not use this file except in compliance with the License.
@@ -15,21 +15,28 @@
  */
 package org.culturegraph.mf.stream.pipe.sort;
 
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
 import org.culturegraph.mf.formeta.Formeta;
 import org.culturegraph.mf.framework.StreamReceiver;
 import org.culturegraph.mf.types.Triple;
 import org.culturegraph.mf.types.Triple.ObjectType;
+import org.junit.Before;
 import org.junit.Test;
 import org.mockito.InOrder;
+import org.mockito.Mock;
 import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
 
 /**
  * Tests {@link TripleCollect}
- * 
+ *
  * @author Markus Geipel
- * 
+ *
  */
 public final class TripleCollectTest {
+
 	private static final String VALUE = "value";
 	private static final String VALUE1 = "value1";
 	private static final String VALUE2 = "value2";
@@ -38,45 +45,67 @@ public final class TripleCollectTest {
 	private static final String REC_ID = "id";
 	private static final String REC_ALT_ID = "altid";
 
+	@Mock
+	private StreamReceiver receiver;
+
+	private TripleCollect collect;
+
+	@Before
+	public void initMocks() {
+		MockitoAnnotations.initMocks(this);
+		collect = new TripleCollect();
+		collect.setReceiver(receiver);
+	}
+
 	@Test
 	public void testShouldBuildRecords() {
-		final StreamReceiver receiver = Mockito.mock(StreamReceiver.class);
-		final TripleCollect collect = new TripleCollect();
-		collect.setReceiver(receiver);
-		
 		collect.process(new Triple(REC_ID, NAME, VALUE1));
 		collect.process(new Triple(REC_ID, NAME, VALUE2));
 		collect.process(new Triple(REC_ALT_ID, NAME, VALUE1));
-		
+		collect.closeStream();
+
 		final InOrder ordered = Mockito.inOrder(receiver);
-		
+
 		ordered.verify(receiver).startRecord(REC_ID);
 		ordered.verify(receiver).literal(NAME, VALUE1);
 		ordered.verify(receiver).literal(NAME, VALUE2);
 		ordered.verify(receiver).endRecord();
 		ordered.verify(receiver).startRecord(REC_ALT_ID);
 		ordered.verify(receiver).literal(NAME, VALUE1);
+		ordered.verify(receiver).endRecord();
 	}
-	
+
 	@Test
 	public void testShouldDecodeEntities() {
-		final StreamReceiver receiver = Mockito.mock(StreamReceiver.class);
-		final TripleCollect collect = new TripleCollect();
-		collect.setReceiver(receiver);
-		
 		collect.process(new Triple(REC_ID,  ENTITY_NAME, Formeta.GROUP_START +NAME + Formeta.NAME_VALUE_SEPARATOR + VALUE
 				+ Formeta.ITEM_SEPARATOR + ENTITY_NAME + Formeta.GROUP_START + NAME
 				+ Formeta.NAME_VALUE_SEPARATOR + VALUE + Formeta.GROUP_END + Formeta.GROUP_END,
 				ObjectType.ENTITY));
-		
+		collect.closeStream();
+
 		final InOrder ordered = Mockito.inOrder(receiver);
-		
+
 		ordered.verify(receiver).startRecord(REC_ID);
 		ordered.verify(receiver).startEntity(ENTITY_NAME);
 		ordered.verify(receiver).literal(NAME, VALUE);
 		ordered.verify(receiver).startEntity(ENTITY_NAME);
 		ordered.verify(receiver).literal(NAME, VALUE);
 		ordered.verify(receiver, Mockito.times(2)).endEntity();
+		ordered.verify(receiver).endRecord();
+	}
 
+	@Test
+	public void shouldNotEmitEndRecordOnCloseStreamIfNoTriplesWereReceived() {
+		collect.closeStream();
+
+		verify(receiver, never()).endRecord();
+	}
+
+	@Test
+	public void shouldNotEmitEndRecordOnResetStreamIfNoTriplesWereReceived() {
+		collect.resetStream();
+
+		verify(receiver, never()).endRecord();
 	}
+
 }