XML: add ability to provide a specific XMLOutputFactory

xml/src/main/scala/akka/stream/alpakka/xml/impl/StreamingXmlWriter.scala

@@ -15,19 +15,20 @@ import javax.xml.stream.XMLOutputFactory
 /**
  * INTERNAL API
  */
-@InternalApi private[xml] class StreamingXmlWriter(charset: Charset)
+@InternalApi private[xml] class StreamingXmlWriter(charset: Charset, xmlOutputFactory: XMLOutputFactory)
     extends GraphStage[FlowShape[ParseEvent, ByteString]] {
+
+  def this(charset: Charset) = this(charset, XMLOutputFactory.newInstance())
+
   val in: Inlet[ParseEvent] = Inlet("XMLWriter.in")
   val out: Outlet[ByteString] = Outlet("XMLWriter.out")
   override val shape: FlowShape[ParseEvent, ByteString] = FlowShape(in, out)
 
-  private val xMLOutputFactory = XMLOutputFactory.newInstance()
-
   override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
     new GraphStageLogic(shape) with InHandler with OutHandler {
       val byteStringBuilder = new ByteStringBuilder()
 
-      val output = xMLOutputFactory.createXMLStreamWriter(byteStringBuilder.asOutputStream, charset.name())
+      val output = xmlOutputFactory.createXMLStreamWriter(byteStringBuilder.asOutputStream, charset.name())
 
       setHandlers(in, out, this)
 

xml/src/main/scala/akka/stream/alpakka/xml/javadsl/XmlWriting.scala

@@ -12,6 +12,8 @@ import akka.stream.alpakka.xml.impl
 import akka.stream.scaladsl.Flow
 import akka.util.ByteString
 
+import javax.xml.stream.XMLOutputFactory
+
 object XmlWriting {
 
   /**
@@ -28,4 +30,20 @@ object XmlWriting {
   def writer(charset: Charset): akka.stream.javadsl.Flow[ParseEvent, ByteString, NotUsed] =
     Flow.fromGraph(new impl.StreamingXmlWriter(charset)).asJava
 
+  /**
+   * Writer Flow that takes a stream of XML events similar to SAX and write ByteStrings.
+   * @param xmlOutputFactory factory from which to get an XMLStreamWriter
+   */
+  def writer(xmlOutputFactory: XMLOutputFactory): akka.stream.javadsl.Flow[ParseEvent, ByteString, NotUsed] =
+    Flow.fromGraph(new impl.StreamingXmlWriter(StandardCharsets.UTF_8, xmlOutputFactory)).asJava
+
+  /**
+   * Writer Flow that takes a stream of XML events similar to SAX and write ByteStrings.
+   * @param charset encoding of the stream
+   * @param xmlOutputFactory factory from which to get an XMLStreamWriter
+   */
+  def writer(charset: Charset,
+             xmlOutputFactory: XMLOutputFactory): akka.stream.javadsl.Flow[ParseEvent, ByteString, NotUsed] =
+    Flow.fromGraph(new impl.StreamingXmlWriter(charset, xmlOutputFactory)).asJava
+
 }

xml/src/main/scala/akka/stream/alpakka/xml/scaladsl/XmlWriting.scala

@@ -12,6 +12,8 @@ import akka.stream.alpakka.xml.impl
 import akka.stream.scaladsl.Flow
 import akka.util.ByteString
 
+import javax.xml.stream.XMLOutputFactory
+
 object XmlWriting {
 
   /**
@@ -21,6 +23,22 @@ object XmlWriting {
   def writer(charset: Charset): Flow[ParseEvent, ByteString, NotUsed] =
     Flow.fromGraph(new impl.StreamingXmlWriter(charset))
 
+  /**
+   * Writer Flow that takes a stream of XML events similar to SAX and write ByteStrings.
+   * encoding UTF-8
+   * @param xmlOutputFactory factory from which to get an XMLStreamWriter
+   */
+  def writer(xmlOutputFactory: XMLOutputFactory): Flow[ParseEvent, ByteString, NotUsed] =
+    Flow.fromGraph(new impl.StreamingXmlWriter(StandardCharsets.UTF_8, xmlOutputFactory))
+
+  /**
+   * Writer Flow that takes a stream of XML events similar to SAX and write ByteStrings.
+   * @param charset charset of encoding
+   * @param xmlOutputFactory factory from which to get an XMLStreamWriter
+   */
+  def writer(charset: Charset, xmlOutputFactory: XMLOutputFactory): Flow[ParseEvent, ByteString, NotUsed] =
+    Flow.fromGraph(new impl.StreamingXmlWriter(charset, xmlOutputFactory))
+
   /**
    * Writer Flow that takes a stream of XML events similar to SAX and write ByteStrings.
    * encoding UTF-8

xml/src/test/java/docs/javadsl/XmlWritingTest.java

@@ -28,6 +28,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import javax.xml.stream.XMLOutputFactory;
+
 import static org.junit.Assert.assertEquals;
 
 public class XmlWritingTest {
@@ -121,6 +123,41 @@ public void xmlWriterNamespace()
         .get(5, TimeUnit.SECONDS);
   }
 
+  @Test
+  public void xmlWriterProvidedFactory()
+      throws InterruptedException, ExecutionException, TimeoutException {
+
+    final XMLOutputFactory xmlOutputFactory = XMLOutputFactory.newInstance();
+    // #writer
+    final Sink<ParseEvent, CompletionStage<String>> write =
+        Flow.of(ParseEvent.class)
+            .via(XmlWriting.writer(xmlOutputFactory))
+            .map(ByteString::utf8String)
+            .toMat(Sink.fold("", (acc, el) -> acc + el), Keep.right());
+    // #writer
+
+    final String doc =
+        "<?xml version='1.0' encoding='UTF-8'?><doc><elem>elem1</elem><elem>elem2</elem></doc>";
+    final List<ParseEvent> docList = new ArrayList<>();
+    docList.add(StartDocument.getInstance());
+    docList.add(StartElement.create("doc", Collections.emptyMap()));
+    docList.add(StartElement.create("elem", Collections.emptyMap()));
+    docList.add(Characters.create("elem1"));
+    docList.add(EndElement.create("elem"));
+    docList.add(StartElement.create("elem", Collections.emptyMap()));
+    docList.add(Characters.create("elem2"));
+    docList.add(EndElement.create("elem"));
+    docList.add(EndElement.create("doc"));
+    docList.add(EndDocument.getInstance());
+
+    final CompletionStage<String> resultStage = Source.from(docList).runWith(write, materializer);
+
+    resultStage
+        .thenAccept((str) -> assertEquals(doc, str))
+        .toCompletableFuture()
+        .get(5, TimeUnit.SECONDS);
+  }
+
   @BeforeClass
   public static void setup() throws Exception {
     system = ActorSystem.create();

xml/src/test/scala/docs/scaladsl/XmlWritingSpec.scala

@@ -9,6 +9,7 @@ import akka.stream.{ActorMaterializer, Materializer}
 import akka.stream.alpakka.xml._
 import akka.stream.alpakka.xml.scaladsl.XmlWriting
 import akka.stream.scaladsl.{Flow, Keep, Sink, Source}
+import javax.xml.stream.XMLOutputFactory
 import org.scalatest.concurrent.PatienceConfiguration.Timeout
 import org.scalatest.concurrent.ScalaFutures
 import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpec}
@@ -205,6 +206,31 @@ class XmlWritingSpec extends WordSpec with Matchers with BeforeAndAfterAll with
       resultFuture.futureValue(Timeout(20.seconds)) should ===(doc)
     }
 
+    "properly work with a provided XMLOutputFactory" in {
+      val listEl: List[ParseEvent] = List(
+        StartDocument,
+        StartElement("doc"),
+        StartElement("elem"),
+        Characters("elem1"),
+        EndElement("elem"),
+        StartElement("elem"),
+        Characters("elem2"),
+        EndElement("elem"),
+        EndElement("doc"),
+        EndDocument
+      )
+
+      val doc = "<?xml version='1.0' encoding='UTF-8'?><doc><elem>elem1</elem><elem>elem2</elem></doc>"
+      val outputFactory = XMLOutputFactory.newInstance()
+      val writer: Sink[ParseEvent, Future[String]] = Flow[ParseEvent]
+        .via(XmlWriting.writer(outputFactory))
+        .map[String](_.utf8String)
+        .toMat(Sink.fold[String, String]("")((t, u) => t + u))(Keep.right)
+      val resultFuture: Future[String] = Source.fromIterator[ParseEvent](() => listEl.iterator).runWith(writer)
+
+      resultFuture.futureValue(Timeout(20.seconds)) should ===(doc)
+    }
+
   }
 
   override protected def afterAll(): Unit = system.terminate()