MLE-17095 Refactor, removed some duplication

Author: rjrudin

src/main/java/com/marklogic/spark/ContextSupport.java

@@ -145,6 +145,10 @@ public final String getStringOption(String option) {
         return hasOption(option) ? properties.get(option).trim() : null;
     }
 
+    public final boolean isStreamingFiles() {
+        return "true".equalsIgnoreCase(getStringOption(Options.STREAM_FILES));
+    }
+
     public Map<String, String> getProperties() {
         return properties;
     }

src/main/java/com/marklogic/spark/reader/document/DocumentContext.java

@@ -43,7 +43,7 @@ Set<DocumentManager.Metadata> getRequestedMetadata() {
     }
 
     boolean contentWasRequested() {
-        if ("true".equals(getStringOption(Options.STREAM_FILES))) {
+        if (isStreamingFiles()) {
             return false;
         }
         if (!hasOption(Options.READ_DOCUMENTS_CATEGORIES)) {

src/main/java/com/marklogic/spark/reader/document/DocumentScanBuilder.java

@@ -17,7 +17,7 @@ class DocumentScanBuilder implements ScanBuilder, SupportsPushDownLimit {
 
     DocumentScanBuilder(CaseInsensitiveStringMap options, StructType schema) {
         this.context = new DocumentContext(options, schema);
-        if ("true".equalsIgnoreCase(this.context.getStringOption(Options.STREAM_FILES)) && Util.MAIN_LOGGER.isInfoEnabled()) {
+        if (this.context.isStreamingFiles() && Util.MAIN_LOGGER.isInfoEnabled()) {
             Util.MAIN_LOGGER.info("Will defer reading documents from MarkLogic so they can be streamed to files during the writer phase.");
         }
     }

src/main/java/com/marklogic/spark/reader/document/ForestReader.java

@@ -54,7 +54,7 @@ class ForestReader implements PartitionReader<InternalRow> {
     ForestReader(ForestPartition forestPartition, DocumentContext context) {
         this.forestPartition = forestPartition;
         this.limit = context.getLimit();
-        this.isStreamingFiles = "true".equalsIgnoreCase(context.getStringOption(Options.STREAM_FILES));
+        this.isStreamingFiles = context.isStreamingFiles();
 
         DatabaseClient client = context.isDirectConnection() ?
             context.connectToMarkLogic(forestPartition.getHost()) :

src/main/java/com/marklogic/spark/reader/file/ArchiveFileReader.java

@@ -44,9 +44,7 @@ public enum StreamingMode {
     ArchiveFileReader(FilePartition filePartition, FileContext fileContext) {
         this(
             filePartition, fileContext,
-            // Will refactor this later to avoid duplication of this comparison.
-            // Should be a nice little method in FileContext.
-            "true".equalsIgnoreCase(fileContext.getStringOption(Options.STREAM_FILES)) ? StreamingMode.STREAM_DURING_READER_PHASE : null
+            fileContext.isStreamingFiles() ? StreamingMode.STREAM_DURING_READER_PHASE : null
         );
     }
 

src/main/java/com/marklogic/spark/reader/file/GenericFileReader.java

@@ -32,7 +32,7 @@ class GenericFileReader implements PartitionReader<InternalRow> {
     GenericFileReader(FilePartition filePartition, FileContext fileContext) {
         this.filePartition = filePartition;
         this.fileContext = fileContext;
-        this.isStreaming = "true".equalsIgnoreCase(fileContext.getStringOption(Options.STREAM_FILES));
+        this.isStreaming = fileContext.isStreamingFiles();
     }
 
     @Override

src/main/java/com/marklogic/spark/writer/DocumentRowConverter.java

@@ -42,7 +42,7 @@ class DocumentRowConverter implements RowConverter {
         this.uriTemplate = writeContext.getStringOption(Options.WRITE_URI_TEMPLATE);
         this.documentFormat = writeContext.getDocumentFormat();
         this.objectMapper = new ObjectMapper();
-        this.isStreamingFromFiles = writeContext.hasOption(Options.STREAM_FILES);
+        this.isStreamingFromFiles = writeContext.isStreamingFiles();
     }
 
     @Override

src/main/java/com/marklogic/spark/writer/WriteBatcherDataWriter.java

@@ -71,7 +71,7 @@ class WriteBatcherDataWriter implements DataWriter<InternalRow> {
         this.docBuilder = this.writeContext.newDocBuilder();
         this.databaseClient = writeContext.connectToMarkLogic();
         this.rowConverter = determineRowConverter();
-        this.isStreamingFiles = "true".equals(writeContext.getStringOption(Options.STREAM_FILES));
+        this.isStreamingFiles = writeContext.isStreamingFiles();
         this.documentManager = this.isStreamingFiles ? databaseClient.newDocumentManager() : null;
 
         if (writeContext.isAbortOnFailure()) {

src/main/java/com/marklogic/spark/writer/file/ContentWriter.java

@@ -42,8 +42,9 @@ class ContentWriter {
     private final GenericDocumentManager documentManager;
 
     ContentWriter(Map<String, String> properties) {
-        this.encoding = determineEncoding(properties);
-        this.prettyPrint = "true".equalsIgnoreCase(properties.get(Options.WRITE_FILES_PRETTY_PRINT));
+        ContextSupport context = new ContextSupport(properties);
+        this.encoding = determineEncoding(context);
+        this.prettyPrint = "true".equalsIgnoreCase(context.getStringOption(Options.WRITE_FILES_PRETTY_PRINT));
         if (this.prettyPrint) {
             this.objectMapper = new ObjectMapper();
             this.transformer = newTransformer();
@@ -52,9 +53,8 @@ class ContentWriter {
             this.objectMapper = null;
         }
 
-        this.isStreamingFiles = "true".equalsIgnoreCase(properties.get(Options.STREAM_FILES));
-        this.documentManager = this.isStreamingFiles ?
-            new ContextSupport(properties).connectToMarkLogic().newDocumentManager() : null;
+        this.isStreamingFiles = context.isStreamingFiles();
+        this.documentManager = this.isStreamingFiles ? context.connectToMarkLogic().newDocumentManager() : null;
     }
 
     void writeContent(InternalRow row, OutputStream outputStream) throws IOException {
@@ -82,9 +82,9 @@ void writeMetadata(InternalRow row, OutputStream outputStream) throws IOExceptio
         }
     }
 
-    private Charset determineEncoding(Map<String, String> properties) {
-        String encodingValue = properties.get(Options.WRITE_FILES_ENCODING);
-        if (encodingValue != null && encodingValue.trim().length() > 0) {
+    private Charset determineEncoding(ContextSupport context) {
+        if (context.hasOption(Options.WRITE_FILES_ENCODING)) {
+            String encodingValue = context.getStringOption(Options.WRITE_FILES_ENCODING);
             try {
                 return Charset.forName(encodingValue);
             } catch (Exception ex) {