Add support for binary serialisation using Apache Avro

morf-core/src/main/java/org/alfasoftware/morf/xml/BaseDataSetReader.java → morf-core/src/main/java/org/alfasoftware/morf/directory/BaseDataSetReader.java

@@ -13,13 +13,12 @@
  * limitations under the License.
  */
 
-package org.alfasoftware.morf.xml;
+package org.alfasoftware.morf.directory;
 
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 
-import org.alfasoftware.morf.xml.XmlStreamProvider.XmlInputStreamProvider;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 
@@ -28,7 +27,7 @@
  *
  * @author Copyright (c) Alfa Financial Software 2012
  */
-abstract class BaseDataSetReader implements XmlInputStreamProvider {
+public abstract class BaseDataSetReader implements DirectoryStreamProvider.DirectoryInputStreamProvider {
 
   /**
    * Maps table names to upper case
@@ -71,7 +70,7 @@ protected final String fileNameForTable(String tableName) {
 
 
   /**
-   * @see org.alfasoftware.morf.xml.XmlStreamProvider.XmlInputStreamProvider#availableStreamNames()
+   * @see org.alfasoftware.morf.directory.DirectoryStreamProvider.DirectoryInputStreamProvider#availableStreamNames()
    */
   @Override
   public Collection<String> availableStreamNames() {

morf-core/src/main/java/org/alfasoftware/morf/directory/DirectoryDataSet.java

@@ -0,0 +1,108 @@
+/* Copyright 2017 Alfa Financial Software
+ *
+ * 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.alfasoftware.morf.directory;
+
+import java.io.*;
+import java.nio.file.Files;
+import java.util.Arrays;
+
+/**
+ * Allows reading of a data set from a directory.
+ *
+ * @author Copyright (c) Alfa Financial Software 2010
+ */
+public class DirectoryDataSet extends BaseDataSetReader implements DirectoryStreamProvider.DirectoryInputStreamProvider, DirectoryStreamProvider.DirectoryOutputStreamProvider {
+
+  /**
+   * The directory to read from.
+   */
+  private final File directory;
+
+  /**
+   * Creates a directory based data set reader based on the directory <var>file</var>.
+   *
+   * @param directory The directory to access.
+   */
+  public DirectoryDataSet(File directory) {
+    super();
+    this.directory = directory;
+
+    if (!directory.isDirectory()) {
+      throw new IllegalArgumentException("[" + directory + "] is not a directory");
+    }
+
+    // read the files in the directory
+    // Do it here because DirectoryDataSet historically did not have to be "open" to be used.
+    for (File file : directory.listFiles()) {
+      if (file.getName().matches(".*\\.xml")) {
+        addTableName(file.getName().replaceAll("\\.xml", ""), file.getName());
+      }
+    }
+  }
+
+
+  /**
+   * @see DirectoryOutputStreamProvider#clearDestination()
+   */
+  @Override
+  public void clearDestination() {
+    for (File file : directory.listFiles()) {
+      // skip files/folders that start with . such as .svn
+      if (file.getName().startsWith(".")) continue;
+
+      deleteFileOrDirectory(file);
+    }
+  }
+
+
+  private void deleteFileOrDirectory(File file) {
+    if (file.isDirectory()) {
+      // clear it out (recursively) if needed...
+      if (!Files.isSymbolicLink(file.toPath())) {
+        Arrays.stream(file.listFiles()).forEach(this::deleteFileOrDirectory);
+      }
+    }
+    if (!file.delete()) {
+      throw new RuntimeException("Exception cleaning output directory, file [" + file + "]");
+    }
+  }
+
+
+  /**
+   * @see DirectoryInputStreamProvider#openInputStreamForTable(String)
+   */
+  @Override
+  public InputStream openInputStreamForTable(String tableName) {
+    try {
+      return new FileInputStream(new File(directory, fileNameForTable(tableName)));
+    } catch (FileNotFoundException e) {
+      throw new RuntimeException("Error opening output stream", e);
+    }
+  }
+
+
+  /**
+   * @see DirectoryOutputStreamProvider#openOutputStreamForTable(String)
+   */
+  @Override
+  public OutputStream openOutputStreamForTable(String tableName) {
+    try {
+      return new FileOutputStream(new File(directory, tableName + ".xml"));
+    } catch (FileNotFoundException e) {
+      throw new RuntimeException("Error opening output stream", e);
+    }
+  }
+}

morf-core/src/main/java/org/alfasoftware/morf/directory/DirectoryDataSetConsumer.java

@@ -0,0 +1,109 @@
+package org.alfasoftware.morf.directory;
+
+import org.alfasoftware.morf.dataset.DataSetConsumer;
+
+import java.io.File;
+import java.util.function.Function;
+
+public abstract class DirectoryDataSetConsumer implements DataSetConsumer {
+
+    /**
+     * Source of content handlers to send data to.
+     */
+    protected final DirectoryStreamProvider.DirectoryOutputStreamProvider directoryOutputStreamProvider;
+
+    /**
+     * What to do about clearing the destination.
+     */
+    protected final ClearDestinationBehaviour clearDestinationBehaviour;
+
+    /**
+     * Creates a data set consumer that will pipe the data set to the file system location
+     * specified by <var>file</var>.
+     *
+     * <p>The serialised output can be written to a single archive or multiple data files:</p>
+     * <ul>
+     * <li>If <var>file</var> identifies a directory then each table in the data set is
+     * serialised to a separate file within that directory.</li>
+     * <li>If <var>file</var> identifies a file name then the file will be created or replaced with
+     * a zip archive containing one file per table in the data set.</li>
+     * </ul>
+     *
+     * @param file The file system location to receive the data set.
+     * @param clearDestinationBehaviour Whether to clear the destination directory or not.
+     */
+    public DirectoryDataSetConsumer(File file,
+                                    DirectoryDataSetConsumer.ClearDestinationBehaviour clearDestinationBehaviour,
+                                    Function<File, DirectoryStreamProvider.DirectoryOutputStreamProvider> archiveDataSetWriterFunction) {
+        super();
+        if (file.isDirectory()) {
+            this.directoryOutputStreamProvider = new DirectoryDataSet(file);
+        } else {
+            this.directoryOutputStreamProvider = archiveDataSetWriterFunction.apply(file);
+        }
+        this.clearDestinationBehaviour = clearDestinationBehaviour;
+    }
+
+    /**
+     * Creates a data set consumer that will pipe the data set to the file system location
+     * specified by <var>file</var>.
+     *
+     * <p>The serialised output can be written to a single archive or multiple data files:</p>
+     * <ul>
+     * <li>If <var>file</var> identifies a directory then each table in the data set is
+     * serialised to a separate XML file within that directory.</li>
+     * <li>If <var>file</var> identifies a file name then the file will be created or replaced with
+     * a zip archive containing one XML file per table in the data set.</li>
+     * </ul>
+     *
+     * @param file The file system location to receive the data set.
+     */
+    public DirectoryDataSetConsumer(File file, Function<File, DirectoryStreamProvider.DirectoryOutputStreamProvider> archiveDataSetWriterFunction) {
+        this(file, DirectoryDataSetConsumer.ClearDestinationBehaviour.CLEAR, archiveDataSetWriterFunction);
+    }
+
+    public DirectoryDataSetConsumer(DirectoryStreamProvider.DirectoryOutputStreamProvider directoryOutputStreamProvider, ClearDestinationBehaviour clearDestinationBehaviour) {
+        this.directoryOutputStreamProvider = directoryOutputStreamProvider;
+        this.clearDestinationBehaviour = clearDestinationBehaviour;
+    }
+
+
+    /**
+     * @see org.alfasoftware.morf.dataset.DataSetConsumer#open()
+     */
+    @Override
+    public void open() {
+        directoryOutputStreamProvider.open();
+
+        if (clearDestinationBehaviour.equals(DirectoryDataSetConsumer.ClearDestinationBehaviour.CLEAR)) {
+            // we're outputting, so clear the destination of any previous runs
+            directoryOutputStreamProvider.clearDestination();
+        }
+    }
+
+    /**
+     * Fired when a dataset has ended.
+     *
+     * @see org.alfasoftware.morf.dataset.DataSetConsumer#close(org.alfasoftware.morf.dataset.DataSetConsumer.CloseState)
+     */
+    @Override
+    public void close(CloseState closeState) {
+        directoryOutputStreamProvider.close();
+    }
+
+
+    /**
+     * Controls the behaviour of the consumer when running against a directory.
+     */
+    public enum ClearDestinationBehaviour {
+        /**
+         * Clear the destination out before extracting (the default)
+         */
+        CLEAR,
+
+        /**
+         * Overwrite the destination
+         */
+        OVERWRITE
+    }
+}

morf-core/src/main/java/org/alfasoftware/morf/directory/DirectoryStreamProvider.java

@@ -0,0 +1,92 @@
+/* Copyright 2017 Alfa Financial Software
+ *
+ * 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.alfasoftware.morf.directory;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collection;
+
+/**
+ * Provides streams for accessing XML data sets.
+ *
+ * @author Copyright (c) Alfa Financial Software 2010
+ */
+public interface DirectoryStreamProvider {
+
+  /**
+   * Opens or creates any resources required to provide content handlers.
+   */
+  void open();
+
+  /**
+   * Closes, releases or finalises any resources.
+   */
+  void close();
+
+  /**
+   * Provides input streams for reading data sets from XML.
+   *
+   * @author Copyright (c) Alfa Financial Software 2010
+   */
+  interface DirectoryInputStreamProvider extends DirectoryStreamProvider {
+
+    /**
+     * Provides an input stream to read XML for a specific table.
+     *
+     * @param tableName The table for which a content handler is requried.
+     * @return An input stream from which the table XML can be read.
+     */
+    InputStream openInputStreamForTable(String tableName);
+
+    /**
+     * @return A collection of stream names that can be provided by calling {@link #openInputStreamForTable(String)}.
+     */
+    Collection<String> availableStreamNames();
+
+    /**
+     * Determines if a table exists.
+     *
+     * @param name The table name to be checked. The case of the name should be ignored.
+     * @return True if a table <var>name</var> exists. False otherwise.
+     */
+    boolean tableExists(String name);
+  }
+
+
+  /**
+   * Provides output streams for writing XML.
+   *
+   * @author Copyright (c) Alfa Financial Software 2010
+   */
+  interface DirectoryOutputStreamProvider extends DirectoryStreamProvider {
+
+    /**
+     * Provides an output stream to write XML for a specific table.
+     *
+     * <p><strong>Streams provided by this method must be closed in the normal
+     * way by the caller.</strong></p>
+     *
+     * @param tableName The table for which a content handler is requried.
+     * @return An output stream to which the table XML can be read.
+     */
+    OutputStream openOutputStreamForTable(String tableName);
+
+    /**
+     * Clear the destination of any previous output. Called after open().
+     */
+    void clearDestination();
+  }
+}

morf-core/src/main/java/org/alfasoftware/morf/xml/ArchiveDataSetReader.java

@@ -25,14 +25,15 @@
 import java.util.zip.ZipEntry;
 import java.util.zip.ZipFile;
 
-import org.alfasoftware.morf.xml.XmlStreamProvider.XmlInputStreamProvider;
+import org.alfasoftware.morf.directory.BaseDataSetReader;
+import org.alfasoftware.morf.directory.DirectoryStreamProvider;
 
 /**
  * Allows reading of data sets based on an archive (zip) file.
  *
  * @author Copyright (c) Alfa Financial Software 2010
  */
-class ArchiveDataSetReader extends BaseDataSetReader implements XmlInputStreamProvider {
+class ArchiveDataSetReader extends BaseDataSetReader implements DirectoryStreamProvider.DirectoryInputStreamProvider {
 
   /**
    * The file to read the archive from.