streaming implementations of bufferable content handles #1250

ehennum · ehennum · commit 1a23a8354145 · 2020-09-08T18:25:43.000-07:00
diff --git a/marklogic-client-api/src/main/java/com/marklogic/client/io/InputSourceHandle.java b/marklogic-client-api/src/main/java/com/marklogic/client/io/InputSourceHandle.java
@@ -52,7 +52,7 @@
  */
 public class InputSourceHandle
   extends BaseHandle<InputStream, OutputStreamSender>
-  implements OutputStreamSender, BufferableHandle, ContentHandle<InputSource>,
+  implements OutputStreamSender, StreamingContentHandle<InputSource, InputStream>,
     XMLReadHandle, XMLWriteHandle,
     StructureReadHandle, StructureWriteHandle, CtsQueryWriteHandle,
     Closeable
@@ -248,6 +248,27 @@ public byte[] toBuffer() {
       throw new MarkLogicIOException(e);
     }
   }
+  @Override
+  public InputSource toContent(InputStream serialization) {
+    return (serialization == null) ? null :
+            new InputSource(new InputStreamReader(serialization, StandardCharsets.UTF_8));
+  }
+  @Override
+  public InputSource bytesToContent(byte[] buffer) {
+    return (buffer == null || buffer.length == 0) ? null :
+            toContent(new ByteArrayInputStream(buffer));
+  }
+  @Override
+  public byte[] contentToBytes(InputSource content) {
+    try {
+      ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+      sendContent(content).write(buffer);
+      return buffer.toByteArray();
+    } catch (IOException e) {
+      throw new MarkLogicIOException("Could not convert InputSource to byte[] array", e);
+    }
+  }
+
   /**
    * Buffers the SAX input source and returns the buffer
    * as an XML string.
@@ -397,6 +418,10 @@ protected OutputStreamSender sendContent() {
 
     return this;
   }
+  protected OutputStreamSender sendContent(InputSource content) {
+    return (content == null) ? null :
+            new OutputStreamSenderImpl(makeTransformer(), makeReader(true), content);
+  }
   @Override
   public void write(OutputStream out) throws IOException {
     try {
@@ -471,4 +496,28 @@ public void close() {
       }
     }
   }
+
+  private static class OutputStreamSenderImpl implements OutputStreamSender {
+    private final TransformerFactory transformerFactory;
+    private final XMLReader xmlReader;
+    private final InputSource content;
+    private OutputStreamSenderImpl(TransformerFactory transformerFactory, XMLReader xmlReader, InputSource content) {
+      this.transformerFactory = transformerFactory;
+      this.xmlReader = xmlReader;
+      this.content  = content;
+    }
+
+    @Override
+    public void write(OutputStream out) throws IOException {
+      try {
+        this.transformerFactory.newTransformer().transform(
+                new SAXSource(this.xmlReader, this.content),
+                new StreamResult(new OutputStreamWriter(out, StandardCharsets.UTF_8))
+        );
+      } catch (TransformerException e) {
+        logger.error("Failed to transform input source into result",e);
+        throw new MarkLogicIOException(e);
+      }
+    }
+  }
 }
diff --git a/marklogic-client-api/src/main/java/com/marklogic/client/io/JacksonParserHandle.java b/marklogic-client-api/src/main/java/com/marklogic/client/io/JacksonParserHandle.java
@@ -17,6 +17,7 @@
 
 import java.io.*;
 
+import com.marklogic.client.io.marker.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -26,14 +27,6 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.marklogic.client.MarkLogicIOException;
 import com.marklogic.client.impl.JacksonBaseHandle;
-import com.marklogic.client.io.marker.BufferableHandle;
-import com.marklogic.client.io.marker.ContentHandle;
-import com.marklogic.client.io.marker.ContentHandleFactory;
-import com.marklogic.client.io.marker.CtsQueryWriteHandle;
-import com.marklogic.client.io.marker.JSONReadHandle;
-import com.marklogic.client.io.marker.JSONWriteHandle;
-import com.marklogic.client.io.marker.StructureReadHandle;
-import com.marklogic.client.io.marker.StructureWriteHandle;
 
 /**
  * <p>An adapter for using the streaming capabilities of the Jackson Open Source library.
@@ -45,8 +38,7 @@
  */
 public class JacksonParserHandle
   extends JacksonBaseHandle<JsonParser>
-  implements ContentHandle<JsonParser>,
-    OutputStreamSender, BufferableHandle,
+  implements OutputStreamSender, StreamingContentHandle<JsonParser, InputStream>,
     JSONReadHandle, JSONWriteHandle,
     StructureReadHandle, StructureWriteHandle, CtsQueryWriteHandle,
     Closeable
diff --git a/marklogic-client-api/src/main/java/com/marklogic/client/io/ReaderHandle.java b/marklogic-client-api/src/main/java/com/marklogic/client/io/ReaderHandle.java
@@ -25,26 +25,12 @@
 import java.io.Reader;
 import java.nio.charset.StandardCharsets;
 
+import com.marklogic.client.io.marker.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.marklogic.client.MarkLogicIOException;
 import com.marklogic.client.impl.Utilities;
-import com.marklogic.client.io.marker.BufferableHandle;
-import com.marklogic.client.io.marker.ContentHandle;
-import com.marklogic.client.io.marker.ContentHandleFactory;
-import com.marklogic.client.io.marker.CtsQueryWriteHandle;
-import com.marklogic.client.io.marker.JSONReadHandle;
-import com.marklogic.client.io.marker.JSONWriteHandle;
-import com.marklogic.client.io.marker.QuadsWriteHandle;
-import com.marklogic.client.io.marker.StructureReadHandle;
-import com.marklogic.client.io.marker.StructureWriteHandle;
-import com.marklogic.client.io.marker.TextReadHandle;
-import com.marklogic.client.io.marker.TextWriteHandle;
-import com.marklogic.client.io.marker.TriplesReadHandle;
-import com.marklogic.client.io.marker.TriplesWriteHandle;
-import com.marklogic.client.io.marker.XMLReadHandle;
-import com.marklogic.client.io.marker.XMLWriteHandle;
 
 /**
  * <p>A Reader Handle represents a character content as a reader
@@ -55,7 +41,7 @@
  */
 public class ReaderHandle
   extends BaseHandle<InputStream, OutputStreamSender>
-  implements OutputStreamSender, BufferableHandle, ContentHandle<Reader>,
+  implements OutputStreamSender, StreamingContentHandle<Reader, InputStream>,
     JSONReadHandle, JSONWriteHandle,
     TextReadHandle, TextWriteHandle,
     XMLReadHandle, XMLWriteHandle,
@@ -198,6 +184,28 @@ public byte[] toBuffer() {
       throw new MarkLogicIOException(e);
     }
   }
+  @Override
+  public Reader toContent(InputStream serialization) {
+    return (serialization == null) ? null :
+            new InputStreamReader(serialization, StandardCharsets.UTF_8);
+  }
+  @Override
+  public Reader bytesToContent(byte[] buffer) {
+    return (buffer == null || buffer.length == 0) ? null :
+            toContent(new ByteArrayInputStream(buffer));
+  }
+  @Override
+  public byte[] contentToBytes(Reader content) {
+    if (content == null) return null;
+    try {
+      ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+      Utilities.write(content, buffer);
+      return buffer.toByteArray();
+    } catch (IOException e) {
+      throw new MarkLogicIOException("Could not convert Reader to byte[] array", e);
+    }
+  }
+
   /**
    * Buffers the character stream and returns the buffer as a string.
    */
diff --git a/marklogic-client-api/src/main/java/com/marklogic/client/io/SourceHandle.java b/marklogic-client-api/src/main/java/com/marklogic/client/io/SourceHandle.java
@@ -31,18 +31,11 @@
 import javax.xml.transform.stream.StreamResult;
 import javax.xml.transform.stream.StreamSource;
 
+import com.marklogic.client.io.marker.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.marklogic.client.MarkLogicIOException;
-import com.marklogic.client.io.marker.BufferableHandle;
-import com.marklogic.client.io.marker.ContentHandle;
-import com.marklogic.client.io.marker.ContentHandleFactory;
-import com.marklogic.client.io.marker.CtsQueryWriteHandle;
-import com.marklogic.client.io.marker.StructureReadHandle;
-import com.marklogic.client.io.marker.StructureWriteHandle;
-import com.marklogic.client.io.marker.XMLReadHandle;
-import com.marklogic.client.io.marker.XMLWriteHandle;
 
 /**
  * <p>A Source Handle represents XML content as a transform source for reading
@@ -52,7 +45,7 @@
  */
 public class SourceHandle
   extends BaseHandle<InputStream, OutputStreamSender>
-  implements OutputStreamSender, BufferableHandle, ContentHandle<Source>,
+  implements OutputStreamSender, StreamingContentHandle<Source, InputStream>,
     XMLReadHandle, XMLWriteHandle,
     StructureReadHandle, StructureWriteHandle, CtsQueryWriteHandle,
     Closeable
@@ -259,6 +252,29 @@ public byte[] toBuffer() {
       throw new MarkLogicIOException(e);
     }
   }
+  @Override
+  public Source toContent(InputStream serialization) {
+    return (serialization == null) ? null :
+            new StreamSource(new InputStreamReader(serialization, StandardCharsets.UTF_8));
+  }
+  @Override
+  public Source bytesToContent(byte[] buffer) {
+    return (buffer == null || buffer.length == 0) ? null :
+            toContent(new ByteArrayInputStream(buffer));
+  }
+  @Override
+  public byte[] contentToBytes(Source content) {
+    if (content == null) return null;
+    try {
+      ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+      makeTransformer().newTransformer().transform(content,
+              new StreamResult(new OutputStreamWriter(buffer, StandardCharsets.UTF_8)));
+      return buffer.toByteArray();
+    } catch (TransformerException e) {
+      throw new MarkLogicIOException("Could not convert Source to byte[] array", e);
+    }
+  }
+
   /**
    * Buffers the transform source and returns the buffer as a string.
    */
diff --git a/marklogic-client-api/src/main/java/com/marklogic/client/io/XMLEventReaderHandle.java b/marklogic-client-api/src/main/java/com/marklogic/client/io/XMLEventReaderHandle.java
@@ -21,7 +21,6 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
-import java.io.UnsupportedEncodingException;
 import java.nio.charset.StandardCharsets;
 
 import javax.xml.stream.FactoryConfigurationError;
@@ -32,19 +31,12 @@
 import javax.xml.stream.XMLResolver;
 import javax.xml.stream.XMLStreamException;
 
+import com.marklogic.client.io.marker.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.marklogic.client.MarkLogicIOException;
 import com.marklogic.client.MarkLogicInternalException;
-import com.marklogic.client.io.marker.BufferableHandle;
-import com.marklogic.client.io.marker.ContentHandle;
-import com.marklogic.client.io.marker.ContentHandleFactory;
-import com.marklogic.client.io.marker.CtsQueryWriteHandle;
-import com.marklogic.client.io.marker.StructureReadHandle;
-import com.marklogic.client.io.marker.StructureWriteHandle;
-import com.marklogic.client.io.marker.XMLReadHandle;
-import com.marklogic.client.io.marker.XMLWriteHandle;
 
 /**
  * <p>An XML Event Reader Handle represents XML content as an XML event reader
@@ -55,7 +47,7 @@
  */
 public class XMLEventReaderHandle
   extends BaseHandle<InputStream, OutputStreamSender>
-  implements OutputStreamSender, BufferableHandle, ContentHandle<XMLEventReader>,
+  implements OutputStreamSender, StreamingContentHandle<XMLEventReader, InputStream>,
     XMLReadHandle, XMLWriteHandle,
     StructureReadHandle, StructureWriteHandle, CtsQueryWriteHandle,
     Closeable
@@ -214,6 +206,39 @@ public byte[] toBuffer() {
       throw new MarkLogicIOException(e);
     }
   }
+  @Override
+  public XMLEventReader toContent(InputStream serialization) {
+    if (serialization == null) return null;
+    try {
+      XMLInputFactory factory = getFactory();
+      if (factory == null) {
+        throw new MarkLogicInternalException("Failed to make StAX input factory");
+      }
+
+      if (resolver != null)
+        factory.setXMLResolver(resolver);
+
+      return factory.createXMLEventReader(serialization, "UTF-8");
+    } catch (XMLStreamException e) {
+      logger.error("Failed to parse StAX events from input stream",e);
+      throw new MarkLogicInternalException(e);
+    } catch (FactoryConfigurationError e) {
+      logger.error("Failed to parse StAX events from input stream",e);
+      throw new MarkLogicInternalException(e);
+    }
+  }
+  @Override
+  public XMLEventReader bytesToContent(byte[] buffer) {
+    return (buffer == null || buffer.length == 0) ? null :
+            toContent(new ByteArrayInputStream(buffer));
+  }
+  @Override
+  public byte[] contentToBytes(XMLEventReader content) {
+    if (content == null) return null;
+    ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+    write(content, buffer);
+    return buffer.toByteArray();
+  }
   /**
    * Buffers the StAX event source and returns the buffer
    * as an XML string.
@@ -275,26 +300,10 @@ protected void receiveContent(InputStream content) {
     }
 
     this.underlyingStream = content;
-    try {
-      if (logger.isInfoEnabled())
-        logger.info("Parsing StAX events from input stream");
-
-      XMLInputFactory factory = getFactory();
-      if (factory == null) {
-        throw new MarkLogicInternalException("Failed to make StAX input factory");
-      }
 
-      if (resolver != null)
-        factory.setXMLResolver(resolver);
-
-      this.content = factory.createXMLEventReader(content, "UTF-8");
-    } catch (XMLStreamException e) {
-      logger.error("Failed to parse StAX events from input stream",e);
-      throw new MarkLogicInternalException(e);
-    } catch (FactoryConfigurationError e) {
-      logger.error("Failed to parse StAX events from input stream",e);
-      throw new MarkLogicInternalException(e);
-    }
+    if (logger.isInfoEnabled())
+      logger.info("Parsing StAX events from input stream");
+    this.content = toContent(content);
   }
   @Override
   protected OutputStreamSender sendContent() {
@@ -306,6 +315,9 @@ protected OutputStreamSender sendContent() {
   }
   @Override
   public void write(OutputStream out) throws IOException {
+    write(this.content, out);
+  }
+  private void write(XMLEventReader content, OutputStream out) {
     try {
       XMLOutputFactory factory = XMLOutputFactory.newFactory();
       XMLEventWriter   writer  = factory.createXMLEventWriter(out, "UTF-8");
diff --git a/marklogic-client-api/src/main/java/com/marklogic/client/io/XMLStreamReaderHandle.java b/marklogic-client-api/src/main/java/com/marklogic/client/io/XMLStreamReaderHandle.java
