Akka Streams with IO Streams

XmlPrettyPrinter.scala

import scala.concurrent._
import scala.util.Try

import java.io._
import java.util.concurrent.Executors

import akka.stream.io._
import akka.stream.scaladsl._
import akka.util.ByteString

object XmlPrettyPrinter {
  private[this] final val bufferSize = 4096

  protected[this] val ioEC = ExecutionContext.fromExecutor(Executors.newCachedThreadPool())

  def asFlow(spaces: Int = 2) = Flow[ByteString, ByteString]() { implicit b =>
    val inputIS = new PipedInputStream(bufferSize)
    val inputOS = new PipedOutputStream(inputIS)

    val outputIS = new PipedInputStream(bufferSize)
    val outputOS = new PipedOutputStream(outputIS)

    Future {
      try {
        prettyPrintXml(inputIS, outputOS, spaces)
      } finally {
        Try(inputIS.close())
        Try(outputOS.close())
      }
    } (ioEC)

    val sink = b.add(OutputStreamSink(() => inputOS))
    val src = b.add(InputStreamSource(() => outputIS))

    (sink, src)
  }

  protected[this] def prettyPrintXml(from: InputStream, to: OutputStream, spaces: Int = 2): Unit = {
    import javax.xml.transform._
    import javax.xml.transform.sax._
    import javax.xml.transform.stream._
    val transformer = TransformerFactory.newInstance().newTransformer()
    transformer.setOutputProperty(OutputKeys.INDENT, "yes")
    transformer.setOutputProperty("{http://xml.apache.org/xslt}indent-amount", spaces.toString)
    val result = new StreamResult(to)
    val source = new SAXSource(new org.xml.sax.InputSource(from))
    transformer.transform(source, result)
  }
}