#76 Added WriteToFileConsumer

Author: rjrudin

src/main/java/com/marklogic/client/ext/datamovement/consumer/WriteToFileConsumer.java

@@ -0,0 +1,71 @@
+package com.marklogic.client.ext.datamovement.consumer;
+
+import com.marklogic.client.document.DocumentRecord;
+import com.marklogic.client.ext.helper.LoggingObject;
+import com.marklogic.client.io.InputStreamHandle;
+import org.springframework.util.FileCopyUtils;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.function.Consumer;
+
+/**
+ * Consumer implementation that is intended to be used with DMSDK's ExportListener. Writes each document to a File based
+ * on the directory passed to this class's constructor plus the document's URI.
+ */
+public class WriteToFileConsumer extends LoggingObject implements Consumer<DocumentRecord> {
+
+	private File baseDir;
+	private boolean logErrors = true;
+
+	public WriteToFileConsumer(File baseDir) {
+		this.baseDir = baseDir;
+	}
+
+	@Override
+	public void accept(DocumentRecord documentRecord) {
+		String uri = documentRecord.getUri();
+		File outputFile = getOutputFile(documentRecord);
+		if (logger.isDebugEnabled()) {
+			logger.debug("Writing document with URI " + uri + " to file: " + outputFile.getAbsolutePath());
+		}
+		try {
+			writeDocumentToFile(documentRecord, outputFile);
+		} catch (IOException e) {
+			String message = "Unable to write document to file; URI: " + uri + "; file: " + outputFile.getAbsolutePath();
+			if (logErrors) {
+				logger.warn(message, e);
+			} else {
+				throw new RuntimeException(message, e);
+			}
+		}
+	}
+
+	protected File getOutputFile(DocumentRecord documentRecord) {
+		return new File(baseDir, documentRecord.getUri());
+	}
+
+	protected void writeDocumentToFile(DocumentRecord documentRecord, File file) throws IOException {
+		file.getParentFile().mkdirs();
+		FileOutputStream fos = new FileOutputStream(file);
+		try {
+			InputStreamHandle handle = documentRecord.getContent(new InputStreamHandle());
+			FileCopyUtils.copy(handle.get(), fos);
+		} finally {
+			fos.close();
+		}
+	}
+
+	protected File getBaseDir() {
+		return baseDir;
+	}
+
+	protected boolean isLogErrors() {
+		return logErrors;
+	}
+
+	public void setLogErrors(boolean logErrors) {
+		this.logErrors = logErrors;
+	}
+}

src/test/java/com/marklogic/client/ext/datamovement/listener/ManageCollectionsTest.java

@@ -1,16 +1,19 @@
 package com.marklogic.client.ext.datamovement.listener;
 
 import com.marklogic.client.datamovement.DeleteListener;
+import com.marklogic.client.datamovement.ExportListener;
 import com.marklogic.client.datamovement.QueryBatch;
 import com.marklogic.client.datamovement.QueryBatchListener;
 import com.marklogic.client.ext.AbstractIntegrationTest;
 import com.marklogic.client.ext.batch.RestBatchWriter;
 import com.marklogic.client.ext.batch.SimpleDocumentWriteOperation;
 import com.marklogic.client.ext.datamovement.QueryBatcherTemplate;
 import com.marklogic.client.ext.datamovement.UrisQueryQueryBatcherBuilder;
+import com.marklogic.client.ext.datamovement.consumer.WriteToFileConsumer;
 import com.marklogic.client.ext.helper.ClientHelper;
 import org.junit.Test;
 
+import java.io.File;
 import java.util.Arrays;
 import java.util.List;
 
@@ -46,6 +49,16 @@ public void processEvent(QueryBatch batch) {
 		));
 		writer.waitForCompletion();
 
+		// Do a quick test of exporting the data
+		ExportListener exportListener = new ExportListener();
+		File exportDir = new File("build/export-test");
+		exportDir.mkdirs();
+		WriteToFileConsumer l = new WriteToFileConsumer(exportDir);
+		exportListener.onDocumentReady(l);
+		qbt.applyOnCollections(exportListener, COLLECTION);
+		assertTrue(new File(exportDir, firstUri).exists());
+		assertTrue(new File(exportDir, secondUri).exists());
+
 		// Set collections
 		qbt.applyOnCollections(new SetCollectionsListener(COLLECTION, "red"), COLLECTION);
 		assertUriInCollections(firstUri, COLLECTION, "red");
@@ -101,6 +114,7 @@ public void processEvent(QueryBatch batch) {
 		qbt.apply(new SetCollectionsListener(COLLECTION, "green"), builder);
 		assertUriInCollections(firstUri, COLLECTION, "blue");
 		assertUriInCollections(secondUri, COLLECTION, "blue");
+
 	}
 
 	private void assertUriInCollections(String uri, String... collections) {