Fix emitting record when stream is closed

By implementing the interfaces instead of extending DefaultObjectPipe the
"closeStream" method can be declared, pushing the record to the downstream
module.

- add test

Complements #296.

Author: dr0i

metafacture-strings/src/main/java/org/metafacture/strings/LineRecorder.java

@@ -1,5 +1,4 @@
-/*
- * Copyright 2019 hbz
+/* Copyright 2019 Pascal Christoph (hbz)
  *
  * Licensed under the Apache License, Version 2.0 the "License";
  * you may not use this file except in compliance with the License.
@@ -13,46 +12,81 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.metafacture.strings;
 
 import org.metafacture.framework.FluxCommand;
+import org.metafacture.framework.ObjectPipe;
 import org.metafacture.framework.ObjectReceiver;
 import org.metafacture.framework.annotations.Description;
 import org.metafacture.framework.annotations.In;
 import org.metafacture.framework.annotations.Out;
-import org.metafacture.framework.helpers.DefaultObjectPipe;
 
 /**
  * Collects strings and emits them as records when a line matches the pattern.
- * Appends to every incoming line a line feed so that the original structure is
- * preserved.
+ * Appends to every incoming line a line feed so that the original structure
+ * is preserved.
  *
  * @author Pascal Christoph (dr0i).
  *
  */
-@Description("Collects strings and emits them as records when a line matches the pattern.")
-@In(String.class)
-@Out(String.class)
-@FluxCommand("lines-to-records")
+@Description ( "Collects strings and emits them as records when a"
+        +" line matches the pattern or the stream is closed." )
+@In ( String.class )
+@Out ( String.class )
+@FluxCommand ( "lines-to-records" )
 public final class LineRecorder
-        extends DefaultObjectPipe<String, ObjectReceiver<String>> {
+        implements ObjectPipe<String,ObjectReceiver<String>> {
+
+    private final static int SB_CAPACITY=4096*7;
+    // empty line is the default
+    private String recordMarkerRegexp="^\\s*$";
+    StringBuilder record=new StringBuilder(
+            SB_CAPACITY);
+    ObjectReceiver<String> receiver;
+
+    public void setRecordMarkerRegexp ( final String regexp ) {
+        recordMarkerRegexp=regexp;
+    }
+
+    @Override
+    public void process ( final String line ) {
+        if(line.matches(
+                recordMarkerRegexp)){
+            getReceiver().process(
+                    record.toString());
+            record=new StringBuilder(
+                    SB_CAPACITY);
+        }else
+            record.append(
+                    line+"\n");
+    }
 
-    private final int SB_CAPACITY = 4096 * 7;
-    private String recordMarkerRegexp = "^\\s*$"; // empty line is default
-    StringBuilder record = new StringBuilder(SB_CAPACITY);
+    @Override
+    public void resetStream ( ) {
+        record=new StringBuilder(
+                SB_CAPACITY);
+    }
 
-    public void setRecordMarkerRegexp(final String regexp) {
-        this.recordMarkerRegexp = regexp;
+    @Override
+    public void closeStream ( ) {
+        getReceiver().process(
+                record.toString());
     }
 
     @Override
-    public void process(final String line) {
-        assert !isClosed();
-        if (line.matches(recordMarkerRegexp)) {
-            getReceiver().process(record.toString());
-            record = new StringBuilder(SB_CAPACITY);
-        } else
-            record.append(line + "\n");
+    public <R extends ObjectReceiver<String>> R setReceiver ( R receiver ) {
+        this.receiver=receiver;
+        return receiver;
+    }
+
+    /**
+     * Returns a reference to the downstream module.
+     *
+     * @return reference to the downstream module
+     */
+    protected final ObjectReceiver<String> getReceiver ( ) {
+        return receiver;
     }
 
 }

metafacture-strings/src/test/java/org/metafacture/strings/LineRecorderTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 Pascal Christoph
+ * Copyright 2019 Pascal Christoph (hbz)
  *
  * Licensed under the Apache License, Version 2.0 the "License";
  * you may not use this file except in compliance with the License.
@@ -99,4 +99,18 @@ public void shouldEmitRecordWithNonDefaultRecordMarker() {
         ordered.verifyNoMoreInteractions();
     }
 
+    @Test
+    public void shouldEmitLastRecordWithoutRecordMarkerWhenClosingStream() {
+        lineRecorder.process(RECORD3_PART1);
+        lineRecorder.process(RECORD3_PART2);
+        lineRecorder.closeStream();
+        final InOrder ordered = inOrder(receiver);
+        ordered.verify(receiver).process(
+                RECORD3_PART1 +
+                LINE_SEPARATOR +
+                RECORD3_PART2 +
+                LINE_SEPARATOR);
+        ordered.verifyNoMoreInteractions();
+    }
+
 }