Add StreamDeferrer module

The `StreamDeferrer` defers forwarding of all events until an
end-record event is received. It is useful for working with
duplicated streams.

Author: cboehme

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

@@ -0,0 +1,92 @@
+/*
+ * Copyright 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.
+ * 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.DefaultStreamPipe;
+import org.culturegraph.mf.framework.StreamReceiver;
+import org.culturegraph.mf.framework.annotations.Description;
+import org.culturegraph.mf.framework.annotations.FluxCommand;
+import org.culturegraph.mf.framework.annotations.In;
+import org.culturegraph.mf.framework.annotations.Out;
+
+/**
+ * Defers all stream events until an <i>end-record</i> is received. Once the
+ * event is received {@code StreamDeferrer} forwards all events it has received
+ * before the <i>end-record</i> event to the next downstream module. The
+ * <i>end-record</i> event is forwarded immediately.
+ * <p>
+ * The {@code StreamDeferrer} is useful when merging two or more streams of
+ * events into one. The module is intended to be added at the end of each flow.
+ * There it ensures that the events for each record are forwarded consecutively
+ * to the merged flow without them being mixed with events belonging to a record
+ * from another stream.
+ *
+ * @author Christoph Böhme
+ * @see org.culturegraph.mf.stream.pipe.StreamTee StreamTee, duplicates event
+ * streams
+ */
+@Description("Defers all stream events until an end-record event is received")
+@FluxCommand("defer-stream")
+@In(StreamReceiver.class)
+@Out(StreamReceiver.class)
+public class StreamDeferrer extends DefaultStreamPipe<StreamReceiver> {
+
+    private final StreamBuffer buffer = new StreamBuffer();
+
+    @Override
+    public void startRecord(final String identifier) {
+        buffer.clear();
+        buffer.startRecord(identifier);
+    }
+
+    @Override
+    public void endRecord() {
+        buffer.endRecord();
+        buffer.replay();
+    }
+
+    @Override
+    public void startEntity(final String name) {
+        buffer.startEntity(name);
+    }
+
+    @Override
+    public void endEntity() {
+        buffer.endEntity();
+    }
+
+    @Override
+    public void literal(final String name, final String value) {
+        buffer.literal(name, value);
+    }
+
+    @Override
+    protected void onSetReceiver() {
+        buffer.setReceiver(getReceiver());
+    }
+
+    @Override
+    protected void onResetStream() {
+        buffer.clear();
+    }
+
+    @Override
+    protected void onCloseStream() {
+        buffer.clear();
+    }
+
+}

src/main/resources/flux-commands.properties

@@ -1,3 +1,19 @@
+#
+# Copyright 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.
+# 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.
+#
+
 # Sources:
 open-file org.culturegraph.mf.stream.source.FileOpener
 open-gzip org.culturegraph.mf.stream.source.GzipOpener
@@ -74,6 +90,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
+defer-stream org.culturegraph.mf.stream.pipe.StreamDeferrer
 
 stream-to-xml	org.culturegraph.mf.stream.converter.xml.SimpleXmlEncoder
 rdf-macros	org.culturegraph.mf.stream.pipe.RdfMacroPipe

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

@@ -0,0 +1,95 @@
+/*
+ * Copyright 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.
+ * 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.StreamReceiver;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.InOrder;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import static org.mockito.Mockito.*;
+
+/**
+ * Tests for class {@link StreamDeferrer}.
+ *
+ * @author Christoph Böhme
+ */
+public class StreamDeferrerTest {
+
+    @Mock
+    private StreamReceiver receiver;
+
+    private StreamDeferrer streamDeferrer;
+
+    @Before
+    public void init() {
+        MockitoAnnotations.initMocks(this);
+        streamDeferrer = new StreamDeferrer();
+        streamDeferrer.setReceiver(receiver);
+    }
+
+    @Test
+    public void shouldDeferStreamEventsUntilEndRecordIsReceived() {
+        streamDeferrer.startRecord("1");
+        streamDeferrer.literal("l", "v");
+        streamDeferrer.startEntity("e");
+        streamDeferrer.endEntity();
+
+        verifyZeroInteractions(receiver);
+
+        streamDeferrer.endRecord();
+
+        InOrder ordered = inOrder(receiver);
+        ordered.verify(receiver).startRecord("1");
+        ordered.verify(receiver).literal("l", "v");
+        ordered.verify(receiver).startEntity("e");
+        ordered.verify(receiver).endEntity();
+        ordered.verify(receiver).endRecord();
+    }
+
+    @Test
+    public void shouldDiscardDeferredEventsIfAnotherStartRecordIsReceived() {
+        streamDeferrer.startRecord("1");
+        streamDeferrer.literal("l1", "v1");
+        streamDeferrer.startRecord("2");
+        streamDeferrer.literal("l2", "v2");
+        streamDeferrer.endRecord();
+
+        InOrder ordered = inOrder(receiver);
+        ordered.verify(receiver, never()).startRecord("1");
+        ordered.verify(receiver, never()).literal("l1", "v1");
+        ordered.verify(receiver).startRecord("2");
+        ordered.verify(receiver).literal("l2", "v2");
+        ordered.verify(receiver).endRecord();
+    }
+
+    @Test
+    public void shouldDiscardDeferredEventsOnResetStream() {
+        streamDeferrer.startRecord("1");
+        streamDeferrer.literal("l", "v");
+        streamDeferrer.resetStream();
+        streamDeferrer.endRecord();
+
+        InOrder ordered = inOrder(receiver);
+        ordered.verify(receiver, never()).startRecord("1");
+        ordered.verify(receiver, never()).literal("l", "v");
+        ordered.verify(receiver).endRecord();
+    }
+
+}