Added header, footer and separator settings to ObjectWriter.

These settings allow to prefix the output with a header string and to
append a footer at the end of the stream. Additionally, the separator
between objects can be now changed. Previously, this was always a
newline character.

The default settings of header, footer and separator are chosen so that
the output of ObjectWriter is the same as in the old version of the
module (i.e. a new line character is append to each object).

In addition to these new options the encoding and compression options
can now be set via Flux.

This commit offers a solution for issue #152: The output of the JSON
encoder can be turned into a JSON array by setting header to "[",
separator to ", " and footer to "]".

Author: cboehme

src/main/java/org/culturegraph/mf/stream/sink/AbstractObjectWriter.java

@@ -0,0 +1,62 @@
+/*
+ *  Copyright 2013 Christoph Böhme
+ *
+ *  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.sink;
+
+/**
+ * Common functions for object writers.
+ * 
+ * @author Christoph Böhme
+ *
+ * @param <T>
+ * 		object type
+ */
+public abstract class AbstractObjectWriter<T> implements ConfigurableObjectWriter<T> {
+
+	private String header = DEFAULT_HEADER;
+	private String footer = DEFAULT_FOOTER;
+	private String separator = DEFAULT_SEPARATOR;
+
+	@Override
+	public final String getHeader() {
+		return header;
+	}
+
+	@Override
+	public final void setHeader(final String header) {
+		this.header = header;
+	}
+
+	@Override
+	public final String getFooter() {
+		return footer;
+	}
+
+	@Override
+	public final void setFooter(final String footer) {
+		this.footer = footer;
+	}
+
+	@Override
+	public final String getSeparator() {
+		return separator;
+	}
+
+	@Override
+	public final void setSeparator(final String separator) {
+		this.separator = separator;
+	}
+
+}

src/main/java/org/culturegraph/mf/stream/sink/ConfigurableObjectWriter.java

@@ -0,0 +1,114 @@
+/*
+ *  Copyright 2013 Christoph Böhme
+ *
+ *  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.sink;
+
+import org.culturegraph.mf.framework.ObjectReceiver;
+import org.culturegraph.mf.util.FileCompression;
+
+/**
+ * Back end implementations for {@link ObjectWriter} should offer
+ * a default set of configuration options. These are defined by
+ * this interface.
+ * 
+ * @author Christoph Böhme
+ *
+ * @param <T> object type
+ */
+public interface ConfigurableObjectWriter<T> extends ObjectReceiver<T> {
+	
+	String DEFAULT_HEADER = "";
+	String DEFAULT_FOOTER = "\n";
+	String DEFAULT_SEPARATOR = "\n";
+	
+	/**
+	 * Returns the encoding used by the underlying writer.
+	 * 
+	 * @return current encoding
+	 */
+	String getEncoding();
+
+	/**
+	 * Sets the encoding used by the underlying writer.
+	 * 
+	 * @param encoding
+	 *            name of the encoding
+	 */
+	void setEncoding(String encoding);
+
+	/**
+	 * Returns the compression mode.
+	 * 
+	 * @return current compression mode
+	 */
+	FileCompression getCompression();
+
+	/**
+	 * Sets the compression mode.
+	 * 
+	 * @param compression
+	 */
+	void setCompression(final FileCompression compression);
+
+	/**
+	 * Sets the compression mode.
+	 * 
+	 * @param compression
+	 */
+	void setCompression(final String compression);
+	
+	/**
+	 * Returns the header which is output before the first object.
+	 * 
+	 * @return header string
+	 */
+	String getHeader();
+	
+	/**
+	 * Sets the header which is output before the first object.
+	 * 
+	 * @param header new header string
+	 */
+	void setHeader(final String header);
+	
+	/**
+	 * Returns the footer which is output after the last object.
+	 * 
+	 * @return footer string
+	 */
+	String getFooter();
+	
+	/**
+	 * Sets the footer which is output after the last object.
+	 * 
+	 * @param footer new footer string
+	 */
+	void setFooter(final String footer);
+	
+	/**
+	 * Returns the separator which is output between objects.
+	 * 
+	 * @return separator string
+	 */
+	String getSeparator();
+	
+	/**
+	 * Sets the separator which is output between objects.
+	 * 
+	 * @param separator new separator string
+	 */
+	void setSeparator(final String separator);
+
+}

src/main/java/org/culturegraph/mf/stream/sink/ObjectFileWriter.java

@@ -24,31 +24,33 @@
 import java.util.regex.Pattern;
 
 import org.culturegraph.mf.exceptions.MetafactureException;
-import org.culturegraph.mf.framework.ObjectReceiver;
 import org.culturegraph.mf.util.FileCompression;
 
 /**
  * @param <T>
  *            object type
  * 
  * @author Markus Geipel
+ * @author Christoph Böhme
  * 
  */
-
-public final class ObjectFileWriter<T> implements ObjectReceiver<T> {
+public final class ObjectFileWriter<T> extends AbstractObjectWriter<T>  {
 
 	private static final String VAR = "${i}";
 	private static final Pattern VAR_PATTERN = Pattern.compile(VAR, Pattern.LITERAL);
 
 	private String path;
 	private int count;
 	private Writer writer;
+	private boolean firstObject;
 	private boolean closed;
 
 	private String encoding = "UTF-8";
 	private FileCompression compression = FileCompression.AUTO;
-
+	
 	public ObjectFileWriter(final String path) {
+		super();
+		
 		this.path = path;
 		startNewFile();
 
@@ -58,67 +60,43 @@ public ObjectFileWriter(final String path) {
 		}
 	}
 
-	/**
-	 * Returns the encoding used to open the resource.
-	 * 
-	 * @return current default setting
-	 */
+	@Override
 	public String getEncoding() {
 		return encoding;
 	}
 
-	/**
-	 * Sets the encoding used to open the resource.
-	 * 
-	 * @param encoding
-	 *            new encoding
-	 */
+	@Override
 	public void setEncoding(final String encoding) {
 		this.encoding = encoding;
 	}
 
+	@Override
 	public FileCompression getCompression() {
 		return compression;
 	}
 
+	@Override
 	public void setCompression(final FileCompression compression) {
 		this.compression = compression;
 	}
 
+	@Override
 	public void setCompression(final String compression) {
 		setCompression(FileCompression.valueOf(compression.toUpperCase()));
 	}
 
-	private void startNewFile() {
-		final Matcher matcher = VAR_PATTERN.matcher(this.path);
-		final String path = matcher.replaceAll(String.valueOf(count));
-		try {
-			final OutputStream file = new FileOutputStream(path);
-			try {
-				final OutputStream compressor = compression.createCompressor(file, path);
-				try {
-					writer = new OutputStreamWriter(compressor, encoding);
-					closed = false;
-				} catch (IOException e) {
-					compressor.close();
-					throw e;
-				}
-			} catch (IOException e) {
-				file.close();
-				throw e;
-			}
-		} catch (IOException e) {
-			throw new MetafactureException("Error creating file '" + path + "'.", e);
-		}
-	}
-
 	@Override
 	public void process(final T obj) {
 		assert !closed;
 		try {
+			if (firstObject) {
+				writer.write(getHeader());
+				firstObject = false;
+			} else {
+				writer.write(getSeparator());
+			}
 			writer.write(obj.toString());
-			writer.append('\n');
-		} catch (IOException e) {
+		} catch (final IOException e) {
 			throw new MetafactureException(e);
 		}
 	}
@@ -127,8 +105,11 @@ public void process(final T obj) {
 	public void resetStream() {
 		if (!closed) {
 			try {
+				if (!firstObject) {
+					writer.write(getFooter());
+				}
 				writer.close();
-			} catch (IOException e) {
+			} catch (final IOException e) {
 				throw new MetafactureException(e);
 			} finally {
 				closed = true;
@@ -142,12 +123,40 @@ public void resetStream() {
 	public void closeStream() {
 		if (!closed) {
 			try {
+				if (!firstObject) {
+					writer.write(getFooter());
+				}
 				writer.close();
-			} catch (IOException e) {
+			} catch (final IOException e) {
 				throw new MetafactureException(e);
 			} finally {
 				closed = true;
 			}
 		}
 	}
+
+	private void startNewFile() {
+		final Matcher matcher = VAR_PATTERN.matcher(this.path);
+		final String path = matcher.replaceAll(String.valueOf(count));
+		try {
+			final OutputStream file = new FileOutputStream(path);
+			try {
+				final OutputStream compressor = compression.createCompressor(file, path);
+				try {
+					writer = new OutputStreamWriter(compressor, encoding);
+					firstObject = true;
+					closed = false;
+				} catch (final IOException e) {
+					compressor.close();
+					throw e;
+				}
+			} catch (final IOException e) {
+				file.close();
+				throw e;
+			}
+		} catch (final IOException e) {
+			throw new MetafactureException("Error creating file '" + path + "'.", e);
+		}
+	}
+
 }