Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Handle XML exceptions #3003

Merged
merged 6 commits into from
Aug 27, 2023
Merged

Handle XML exceptions #3003

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
54d4b1f
initial without tailrec
JustinPihony Aug 25, 2023
1399dd4
remove errant println
JustinPihony Aug 25, 2023
adfdfd5
keep tail recursion
JustinPihony Aug 25, 2023
a255205
import exception
JustinPihony Aug 25, 2023
88971e0
merge
JustinPihony Aug 25, 2023
9114726
Formatting
johanandren Aug 27, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
131 changes: 70 additions & 61 deletions xml/src/main/scala/akka/stream/alpakka/xml/impl/StreamingXmlParser.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,8 +46,12 @@ private[xml] object StreamingXmlParser {

override def onPush(): Unit = {
val array = grab(in).toArray
parser.getInputFeeder.feedInput(array, 0, array.length)
advanceParser()
try {
parser.getInputFeeder.feedInput(array, 0, array.length)
advanceParser()
} catch {
case xmlException: javax.xml.stream.XMLStreamException => failStage(xmlException)
}
}

override def onPull(): Unit = advanceParser()
Expand All @@ -58,67 +62,72 @@ private[xml] object StreamingXmlParser {
else if (isAvailable(out)) advanceParser()
}

@tailrec private def advanceParser(): Unit =
private def advanceParser(): Unit = {
if (parser.hasNext) {
parser.next() match {
case AsyncXMLStreamReader.EVENT_INCOMPLETE if isClosed(in) && !started => completeStage()
case AsyncXMLStreamReader.EVENT_INCOMPLETE if isClosed(in) => failStage(withStreamingFinishedException)
case AsyncXMLStreamReader.EVENT_INCOMPLETE => pull(in)

case XMLStreamConstants.START_DOCUMENT =>
started = true
push(out, StartDocument)

case XMLStreamConstants.END_DOCUMENT =>
push(out, EndDocument)
completeStage()

case XMLStreamConstants.START_ELEMENT =>
val attributes = (0 until parser.getAttributeCount).map { i =>
val optNs = Option(parser.getAttributeNamespace(i)).filterNot(_ == "")
val optPrefix = Option(parser.getAttributePrefix(i)).filterNot(_ == "")
Attribute(name = parser.getAttributeLocalName(i),
value = parser.getAttributeValue(i),
prefix = optPrefix,
namespace = optNs)
}.toList
val namespaces = (0 until parser.getNamespaceCount).map { i =>
val namespace = parser.getNamespaceURI(i)
val optPrefix = Option(parser.getNamespacePrefix(i)).filterNot(_ == "")
Namespace(namespace, optPrefix)
}.toList
val optPrefix = Option(parser.getPrefix)
val optNs = optPrefix.flatMap(prefix => Option(parser.getNamespaceURI(prefix)))
push(
out,
StartElement(parser.getLocalName,
attributes,
optPrefix.filterNot(_ == ""),
optNs.filterNot(_ == ""),
namespaceCtx = namespaces)
)

case XMLStreamConstants.END_ELEMENT =>
push(out, EndElement(parser.getLocalName))

case XMLStreamConstants.CHARACTERS =>
push(out, Characters(parser.getText))

case XMLStreamConstants.PROCESSING_INSTRUCTION =>
push(out, ProcessingInstruction(Option(parser.getPITarget), Option(parser.getPIData)))

case XMLStreamConstants.COMMENT =>
push(out, Comment(parser.getText))

case XMLStreamConstants.CDATA =>
push(out, CData(parser.getText))

// Do not support DTD, SPACE, NAMESPACE, NOTATION_DECLARATION, ENTITY_DECLARATION, PROCESSING_INSTRUCTION
// ATTRIBUTE is handled in START_ELEMENT implicitly

case x =>
advanceParser()
try {
parser.next() match {
case AsyncXMLStreamReader.EVENT_INCOMPLETE if isClosed(in) && !started => completeStage()
case AsyncXMLStreamReader.EVENT_INCOMPLETE if isClosed(in) => failStage(withStreamingFinishedException)
case AsyncXMLStreamReader.EVENT_INCOMPLETE => pull(in)

case XMLStreamConstants.START_DOCUMENT =>
started = true
push(out, StartDocument)

case XMLStreamConstants.END_DOCUMENT =>
push(out, EndDocument)
completeStage()

case XMLStreamConstants.START_ELEMENT =>
val attributes = (0 until parser.getAttributeCount).map { i =>
val optNs = Option(parser.getAttributeNamespace(i)).filterNot(_ == "")
val optPrefix = Option(parser.getAttributePrefix(i)).filterNot(_ == "")
Attribute(name = parser.getAttributeLocalName(i),
value = parser.getAttributeValue(i),
prefix = optPrefix,
namespace = optNs)
}.toList
val namespaces = (0 until parser.getNamespaceCount).map { i =>
val namespace = parser.getNamespaceURI(i)
val optPrefix = Option(parser.getNamespacePrefix(i)).filterNot(_ == "")
Namespace(namespace, optPrefix)
}.toList
val optPrefix = Option(parser.getPrefix)
val optNs = optPrefix.flatMap(prefix => Option(parser.getNamespaceURI(prefix)))
push(
out,
StartElement(parser.getLocalName,
attributes,
optPrefix.filterNot(_ == ""),
optNs.filterNot(_ == ""),
namespaceCtx = namespaces)
)

case XMLStreamConstants.END_ELEMENT =>
push(out, EndElement(parser.getLocalName))

case XMLStreamConstants.CHARACTERS =>
push(out, Characters(parser.getText))

case XMLStreamConstants.PROCESSING_INSTRUCTION =>
push(out, ProcessingInstruction(Option(parser.getPITarget), Option(parser.getPIData)))

case XMLStreamConstants.COMMENT =>
push(out, Comment(parser.getText))

case XMLStreamConstants.CDATA =>
push(out, CData(parser.getText))

// Do not support DTD, SPACE, NAMESPACE, NOTATION_DECLARATION, ENTITY_DECLARATION, PROCESSING_INSTRUCTION
// ATTRIBUTE is handled in START_ELEMENT implicitly

case x =>
advanceParser()
}
} catch {
case xmlException: javax.xml.stream.XMLStreamException => failStage(xmlException)
}
} else completeStage()
}
}
}
9 changes: 9 additions & 0 deletions xml/src/test/scala/docs/scaladsl/XmlProcessingSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,13 +12,16 @@ import akka.stream.scaladsl.{Flow, Keep, Sink, Source}
import akka.util.ByteString
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.BeforeAndAfterAll
import org.scalatest.exceptions.TestFailedException

import scala.collection.immutable
import scala.concurrent.duration._
import scala.concurrent.{Await, Future}
import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpec

import javax.xml.stream.XMLStreamException

class XmlProcessingSpec extends AnyWordSpec with Matchers with ScalaFutures with BeforeAndAfterAll with LogCapturing {
implicit val system: ActorSystem = ActorSystem("Test")
implicit val defaultPatience: PatienceConfig = PatienceConfig(timeout = 2.seconds, interval = 50.millis)
Expand All @@ -36,6 +39,12 @@ class XmlProcessingSpec extends AnyWordSpec with Matchers with ScalaFutures with
Source.single("").runWith(parse).futureValue should ===(Vector())
}

"fail if invalid XML" in {
val doubleDocType = "<!DOCTYPE><!DOCTYPE>"
val failure = intercept[TestFailedException] {Source.single(doubleDocType).runWith(parse).futureValue}
failure.cause.get shouldBe a[XMLStreamException]
}

"properly parse simple XML" in {
// #parser-usage
val doc = "<doc><elem>elem1</elem><elem>elem2</elem></doc>"
Expand Down