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

Jump to bottom

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 all 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
22 changes: 18 additions & 4 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 @@ -13,6 +13,7 @@ import com.fasterxml.aalto.{AsyncByteArrayFeeder, AsyncXMLInputFactory, AsyncXML
import com.fasterxml.aalto.stax.InputFactoryImpl
import com.fasterxml.aalto.util.IllegalCharHandler.ReplacingIllegalCharHandler

import javax.xml.stream.XMLStreamException
import scala.annotation.tailrec

private[xml] object StreamingXmlParser {
Expand Down Expand Up @@ -73,16 +74,29 @@ private[xml] object StreamingXmlParser {
val bs = transform.getByteString(a)
context = transform.getContext(a)
val array = bs.toArray
parser.getInputFeeder.feedInput(array, 0, array.length)
advanceParser()
try {
parser.getInputFeeder.feedInput(array, 0, array.length)
advanceParser()
} catch {
case xmlException: XMLStreamException => failStage(xmlException)
}
}

override def onPull(): Unit = advanceParser()
override def onPull(): Unit =
try {
advanceParser()
} catch {
case xmlException: XMLStreamException => failStage(xmlException)
}

override def onUpstreamFinish(): Unit = {
parser.getInputFeeder.endOfInput()
if (!parser.hasNext) completeStage()
else if (isAvailable(out)) advanceParser()
else if (isAvailable(out)) try {
advanceParser()
} catch {
case xmlException: XMLStreamException => failStage(xmlException)
}
}

@tailrec private def advanceParser(): Unit =
Expand Down
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, Framing, 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