Improve Jaxb2Decoder

- Introcuces XmlEventDecoder which decodes from DataBuffer to
  javax.xml.stream.events.XMLEvent. It uses the Aalto async XML API if
  available, but falls back to a blocking default if not.

- Refacors Jaxb2Decoder to use said XmlEventDecoder, and split the
  stream of events into separate substreams by using the JAXB annotation
  value, one stream for each part of the tree that can be unmarshaled to
  the given type.

- Various improvements in the JAXB code.

spring-web-reactive/src/main/java/org/springframework/core/codec/support/Jaxb2Decoder.java

@@ -16,30 +16,28 @@
 
 package org.springframework.core.codec.support;
 
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import javax.xml.bind.JAXBContext;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Function;
+import javax.xml.XMLConstants;
 import javax.xml.bind.JAXBElement;
 import javax.xml.bind.JAXBException;
-import javax.xml.bind.UnmarshalException;
 import javax.xml.bind.Unmarshaller;
 import javax.xml.bind.annotation.XmlRootElement;
-import javax.xml.transform.Source;
-import javax.xml.transform.sax.SAXSource;
-import javax.xml.transform.stream.StreamSource;
+import javax.xml.bind.annotation.XmlSchema;
+import javax.xml.bind.annotation.XmlType;
+import javax.xml.namespace.QName;
+import javax.xml.stream.XMLEventReader;
+import javax.xml.stream.events.XMLEvent;
 
 import org.reactivestreams.Publisher;
-import org.xml.sax.InputSource;
-import org.xml.sax.SAXException;
-import org.xml.sax.XMLReader;
-import org.xml.sax.helpers.XMLReaderFactory;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.codec.CodecException;
 import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.core.io.buffer.support.DataBufferUtils;
-import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
 import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
 
@@ -47,87 +45,170 @@ import org.springframework.util.MimeTypeUtils;
  * Decode from a bytes stream of XML elements to a stream of {@code Object} (POJO).
  *
  * @author Sebastien Deleuze
+ * @author Arjen Poutsma
  * @see Jaxb2Encoder
  */
 public class Jaxb2Decoder extends AbstractDecoder<Object> {
 
-	private final ConcurrentMap<Class<?>, JAXBContext> jaxbContexts = new ConcurrentHashMap<>(64);
+	/**
+	 * The default value for JAXB annotations.
+	 * @see XmlRootElement#name()
+	 * @see XmlRootElement#namespace()
+	 * @see XmlType#name()
+	 * @see XmlType#namespace()
+	 */
+	private final static String JAXB_DEFAULT_ANNOTATION_VALUE = "##default";
 
+	private final XmlEventDecoder xmlEventDecoder = new XmlEventDecoder();
+
+	private final JaxbContextContainer jaxbContexts = new JaxbContextContainer();
 
 	public Jaxb2Decoder() {
 		super(MimeTypeUtils.APPLICATION_XML, MimeTypeUtils.TEXT_XML);
 	}
 
+	@Override
+	public boolean canDecode(ResolvableType type, MimeType mimeType, Object... hints) {
+		if (super.canDecode(type, mimeType, hints)) {
+			Class<?> outputClass = type.getRawClass();
+			return outputClass.isAnnotationPresent(XmlRootElement.class) ||
+					outputClass.isAnnotationPresent(XmlType.class);
+		}
+		else {
+			return false;
+		}
+	}
 
 	@Override
 	public Flux<Object> decode(Publisher<DataBuffer> inputStream, ResolvableType type,
 			MimeType mimeType, Object... hints) {
-
 		Class<?> outputClass = type.getRawClass();
-		try {
-			Source source = processSource(
-					new StreamSource(DataBufferUtils.toInputStream(inputStream)));
-			Unmarshaller unmarshaller = createUnmarshaller(outputClass);
-			if (outputClass.isAnnotationPresent(XmlRootElement.class)) {
-				return Flux.just(unmarshaller.unmarshal(source));
-			}
-			else {
-				JAXBElement<?> jaxbElement = unmarshaller.unmarshal(source, outputClass);
-				return Flux.just(jaxbElement.getValue());
-			}
-		}
-		catch (UnmarshalException ex) {
-			return Flux.error(
-			  new CodecException("Could not unmarshal to [" + outputClass + "]: " + ex.getMessage(), ex));
-		}
-		catch (JAXBException ex) {
-			return Flux.error(new CodecException("Could not instantiate JAXBContext: " +
-					ex.getMessage(), ex));
-		}
+		Flux<XMLEvent> xmlEventFlux =
+				this.xmlEventDecoder.decode(inputStream, null, mimeType);
+
+		QName typeName = toQName(outputClass);
+		Flux<List<XMLEvent>> splitEvents = split(xmlEventFlux, typeName);
+
+		return splitEvents.map(events -> unmarshal(events, outputClass));
 	}
 
-	protected Source processSource(Source source) {
-		if (source instanceof StreamSource) {
-			StreamSource streamSource = (StreamSource) source;
-			InputSource inputSource = new InputSource(streamSource.getInputStream());
-			try {
-				XMLReader xmlReader = XMLReaderFactory.createXMLReader();
-				return new SAXSource(xmlReader, inputSource);
-			}
-			catch (SAXException ex) {
-				throw new CodecException("Error while processing the source", ex);
-			}
+	/**
+	 * Returns the qualified name for the given class, according to the mapping rules
+	 * in the JAXB specification.
+	 */
+	QName toQName(Class<?> outputClass) {
+		String localPart;
+		String namespaceUri;
+
+		if (outputClass.isAnnotationPresent(XmlRootElement.class)) {
+			XmlRootElement annotation = outputClass.getAnnotation(XmlRootElement.class);
+			localPart = annotation.name();
+			namespaceUri = annotation.namespace();
+		}
+		else if (outputClass.isAnnotationPresent(XmlType.class)) {
+			XmlType annotation = outputClass.getAnnotation(XmlType.class);
+			localPart = annotation.name();
+			namespaceUri = annotation.namespace();
 		}
 		else {
-			return source;
+			throw new IllegalArgumentException("Outputclass [" + outputClass + "] is " +
+					"neither annotated with @XmlRootElement nor @XmlType");
 		}
+
+		if (JAXB_DEFAULT_ANNOTATION_VALUE.equals(localPart)) {
+			localPart = ClassUtils.getShortNameAsProperty(outputClass);
+		}
+		if (JAXB_DEFAULT_ANNOTATION_VALUE.equals(namespaceUri)) {
+			Package outputClassPackage = outputClass.getPackage();
+			if (outputClassPackage != null &&
+					outputClassPackage.isAnnotationPresent(XmlSchema.class)) {
+				XmlSchema annotation = outputClassPackage.getAnnotation(XmlSchema.class);
+				namespaceUri = annotation.namespace();
+			}
+			else {
+				namespaceUri = XMLConstants.NULL_NS_URI;
+			}
+		}
+		return new QName(namespaceUri, localPart);
 	}
 
-	protected final Unmarshaller createUnmarshaller(Class<?> clazz) throws JAXBException {
+	/**
+	 * Split a flux of {@link XMLEvent}s into a flux of XMLEvent lists, one list for each
+	 * branch of the tree that starts with the given qualified name.
+	 * That is, given the XMLEvents shown
+	 * {@linkplain XmlEventDecoder here},
+	 * and the {@code desiredName} "{@code child}", this method
+	 * returns a flux of two lists, each of which containing the events of a particular
+	 * branch of the tree that starts with "{@code child}".
+	 * <ol>
+	 * <li>The first list, dealing with the first branch of the tree
+	 * <ol>
+	 * <li>{@link javax.xml.stream.events.StartElement} {@code child}</li>
+	 * <li>{@link javax.xml.stream.events.Characters} {@code foo}</li>
+	 * <li>{@link javax.xml.stream.events.EndElement} {@code child}</li>
+	 * </ol>
+	 * <li>The second list, dealing with the second branch of the tree
+	 * <ol>
+	 * <li>{@link javax.xml.stream.events.StartElement} {@code child}</li>
+	 * <li>{@link javax.xml.stream.events.Characters} {@code bar}</li>
+	 * <li>{@link javax.xml.stream.events.EndElement} {@code child}</li>
+	 * </ol>
+	 * </li>
+	 * </ol>
+	 */
+	Flux<List<XMLEvent>> split(Flux<XMLEvent> xmlEventFlux, QName desiredName) {
+		return xmlEventFlux
+				.flatMap(new Function<XMLEvent, Publisher<? extends List<XMLEvent>>>() {
+
+					private List<XMLEvent> events = null;
+
+					private int elementDepth = 0;
+
+					private int barrier = Integer.MAX_VALUE;
+
+					@Override
+					public Publisher<? extends List<XMLEvent>> apply(XMLEvent event) {
+						if (event.isStartElement()) {
+							if (this.barrier == Integer.MAX_VALUE) {
+								QName startElementName = event.asStartElement().getName();
+								if (desiredName.equals(startElementName)) {
+									this.events = new ArrayList<XMLEvent>();
+									this.barrier = this.elementDepth;
+								}
+							}
+							this.elementDepth++;
+						}
+						if (this.elementDepth > this.barrier) {
+							this.events.add(event);
+						}
+						if (event.isEndElement()) {
+							this.elementDepth--;
+							if (this.elementDepth == this.barrier) {
+								this.barrier = Integer.MAX_VALUE;
+								return Mono.just(this.events);
+							}
+						}
+						return Mono.empty();
+					}
+				});
+	}
+
+	private Object unmarshal(List<XMLEvent> eventFlux, Class<?> outputClass) {
 		try {
-			JAXBContext jaxbContext = getJaxbContext(clazz);
-			return jaxbContext.createUnmarshaller();
+			Unmarshaller unmarshaller = this.jaxbContexts.createUnmarshaller(outputClass);
+			XMLEventReader eventReader = new ListBasedXMLEventReader(eventFlux);
+			if (outputClass.isAnnotationPresent(XmlRootElement.class)) {
+				return unmarshaller.unmarshal(eventReader);
+			}
+			else {
+				JAXBElement<?> jaxbElement =
+						unmarshaller.unmarshal(eventReader, outputClass);
+				return jaxbElement.getValue();
+			}
 		}
 		catch (JAXBException ex) {
-			throw new CodecException("Could not create Unmarshaller for class " +
-					"[" + clazz + "]: " + ex.getMessage(), ex);
+			throw new CodecException(ex.getMessage(), ex);
 		}
 	}
 
-	protected final JAXBContext getJaxbContext(Class<?> clazz) {
-		Assert.notNull(clazz, "'clazz' must not be null");
-		JAXBContext jaxbContext = this.jaxbContexts.get(clazz);
-		if (jaxbContext == null) {
-			try {
-				jaxbContext = JAXBContext.newInstance(clazz);
-				this.jaxbContexts.putIfAbsent(clazz, jaxbContext);
-			}
-			catch (JAXBException ex) {
-				throw new CodecException("Could not instantiate JAXBContext for class " +
-						"[" + clazz + "]: " + ex.getMessage(), ex);
-			}
-		}
-		return jaxbContext;
-	}
-
 }

spring-web-reactive/src/main/java/org/springframework/core/codec/support/Jaxb2Encoder.java

@@ -18,12 +18,11 @@ package org.springframework.core.codec.support;
 
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import javax.xml.bind.JAXBContext;
 import javax.xml.bind.JAXBException;
 import javax.xml.bind.MarshalException;
 import javax.xml.bind.Marshaller;
+import javax.xml.bind.annotation.XmlRootElement;
+import javax.xml.bind.annotation.XmlType;
 
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
@@ -32,7 +31,6 @@ import org.springframework.core.ResolvableType;
 import org.springframework.core.codec.CodecException;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferAllocator;
-import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
@@ -41,27 +39,43 @@ import org.springframework.util.MimeTypeUtils;
  * Encode from an {@code Object} stream to a byte stream of XML elements.
  *
  * @author Sebastien Deleuze
+ * @author Arjen Poutsma
  * @see Jaxb2Decoder
  */
 public class Jaxb2Encoder extends AbstractEncoder<Object> {
 
-	private final ConcurrentMap<Class<?>, JAXBContext> jaxbContexts = new ConcurrentHashMap<>(64);
+	private final JaxbContextContainer jaxbContexts = new JaxbContextContainer();
 
 	public Jaxb2Encoder() {
 		super(MimeTypeUtils.APPLICATION_XML, MimeTypeUtils.TEXT_XML);
 	}
 
+	@Override
+	public boolean canEncode(ResolvableType type, MimeType mimeType, Object... hints) {
+		if (super.canEncode(type, mimeType, hints)) {
+			Class<?> outputClass = type.getRawClass();
+			return outputClass.isAnnotationPresent(XmlRootElement.class) ||
+					outputClass.isAnnotationPresent(XmlType.class);
+		}
+		else {
+			return false;
+		}
+
+	}
+
 	@Override
 	public Flux<DataBuffer> encode(Publisher<?> inputStream,
 			DataBufferAllocator allocator, ResolvableType type, MimeType mimeType,
 			Object... hints) {
 
-		return Flux.from(inputStream).map(value -> {
+		return Flux.from(inputStream).
+				take(1). // only map 1 value to ensure valid XML output
+				map(value -> {
 			try {
 				DataBuffer buffer = allocator.allocateBuffer(1024);
 				OutputStream outputStream = buffer.asOutputStream();
 				Class<?> clazz = ClassUtils.getUserClass(value);
-				Marshaller marshaller = createMarshaller(clazz);
+				Marshaller marshaller = jaxbContexts.createMarshaller(clazz);
 				marshaller.setProperty(Marshaller.JAXB_ENCODING, StandardCharsets.UTF_8.name());
 				marshaller.marshal(value, outputStream);
 				return buffer;
@@ -75,32 +89,7 @@ public class Jaxb2Encoder extends AbstractEncoder<Object> {
 		});
 	}
 
-	protected final Marshaller createMarshaller(Class<?> clazz) {
-		try {
-			JAXBContext jaxbContext = getJaxbContext(clazz);
-			return jaxbContext.createMarshaller();
-		}
-		catch (JAXBException ex) {
-			throw new CodecException("Could not create Marshaller for class " +
-					"[" + clazz + "]: " + ex.getMessage(), ex);
-		}
-	}
 
-	protected final JAXBContext getJaxbContext(Class<?> clazz) {
-		Assert.notNull(clazz, "'clazz' must not be null");
-		JAXBContext jaxbContext = this.jaxbContexts.get(clazz);
-		if (jaxbContext == null) {
-			try {
-				jaxbContext = JAXBContext.newInstance(clazz);
-				this.jaxbContexts.putIfAbsent(clazz, jaxbContext);
-			}
-			catch (JAXBException ex) {
-				throw new CodecException("Could not instantiate JAXBContext for class " +
-						"[" + clazz + "]: " + ex.getMessage(), ex);
-			}
-		}
-		return jaxbContext;
-	}
 
 }
 

spring-web-reactive/src/main/java/org/springframework/core/codec/support/JaxbContextContainer.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2002-2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.core.codec.support;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import javax.xml.bind.JAXBContext;
+import javax.xml.bind.JAXBException;
+import javax.xml.bind.Marshaller;
+import javax.xml.bind.Unmarshaller;
+
+import org.springframework.util.Assert;
+
+/**
+ * @author Arjen Poutsma
+ */
+final class JaxbContextContainer {
+
+	private final ConcurrentMap<Class<?>, JAXBContext> jaxbContexts =
+			new ConcurrentHashMap<>(64);
+
+	public Marshaller createMarshaller(Class<?> clazz) throws JAXBException {
+		JAXBContext jaxbContext = getJaxbContext(clazz);
+		return jaxbContext.createMarshaller();
+	}
+
+	public Unmarshaller createUnmarshaller(Class<?> clazz) throws JAXBException {
+		JAXBContext jaxbContext = getJaxbContext(clazz);
+		return jaxbContext.createUnmarshaller();
+	}
+
+	private JAXBContext getJaxbContext(Class<?> clazz) throws JAXBException {
+		Assert.notNull(clazz, "'clazz' must not be null");
+		JAXBContext jaxbContext = this.jaxbContexts.get(clazz);
+		if (jaxbContext == null) {
+			jaxbContext = JAXBContext.newInstance(clazz);
+			this.jaxbContexts.putIfAbsent(clazz, jaxbContext);
+		}
+		return jaxbContext;
+	}
+
+}

spring-web-reactive/src/main/java/org/springframework/core/codec/support/ListBasedXMLEventReader.java

@@ -0,0 +1,151 @@
+/*
+ * Copyright 2002-2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.core.codec.support;
+
+import java.util.List;
+import java.util.NoSuchElementException;
+import javax.xml.stream.XMLEventReader;
+import javax.xml.stream.XMLStreamConstants;
+import javax.xml.stream.XMLStreamException;
+import javax.xml.stream.events.Characters;
+import javax.xml.stream.events.XMLEvent;
+
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+
+/**
+ * TODO: move to org.springframework.util.xml when merging, hidden behind StaxUtils
+ *
+ * @author Arjen Poutsma
+ */
+class ListBasedXMLEventReader implements XMLEventReader {
+
+	private final XMLEvent[] events;
+
+	private int cursor = 0;
+
+	public ListBasedXMLEventReader(List<XMLEvent> events) {
+		Assert.notNull(events, "'events' must not be null");
+		this.events = events.toArray(new XMLEvent[events.size()]);
+	}
+
+	@Override
+	public boolean hasNext() {
+		Assert.notNull(events, "'events' must not be null");
+		return cursor != events.length;
+	}
+
+	@Override
+	public XMLEvent nextEvent() {
+		if (cursor < events.length) {
+			return events[cursor++];
+		}
+		else {
+			throw new NoSuchElementException();
+		}
+	}
+
+	@Override
+	public XMLEvent peek() {
+		if (cursor < events.length) {
+			return events[cursor];
+		}
+		else {
+			return null;
+		}
+	}
+
+	@Override
+	public Object next() {
+		return nextEvent();
+	}
+
+	/**
+	 * Throws an {@code UnsupportedOperationException} when called.
+	 * @throws UnsupportedOperationException when called
+	 */
+	@Override
+	public void remove() {
+		throw new UnsupportedOperationException(
+				"remove not supported on " + ClassUtils.getShortName(getClass()));
+	}
+
+	@Override
+	public String getElementText() throws XMLStreamException {
+		if (!peek().isStartElement()) {
+			throw new XMLStreamException("Not at START_ELEMENT");
+		}
+
+		StringBuilder builder = new StringBuilder();
+		while (true) {
+			XMLEvent event = nextEvent();
+			if (event.isEndElement()) {
+				break;
+			}
+			else if (!event.isCharacters()) {
+				throw new XMLStreamException(
+						"Unexpected event [" + event + "] in getElementText()");
+			}
+			Characters characters = event.asCharacters();
+			if (!characters.isIgnorableWhiteSpace()) {
+				builder.append(event.asCharacters().getData());
+			}
+		}
+		return builder.toString();
+	}
+
+	@Override
+	public XMLEvent nextTag() throws XMLStreamException {
+		while (true) {
+			XMLEvent event = nextEvent();
+			switch (event.getEventType()) {
+				case XMLStreamConstants.START_ELEMENT:
+				case XMLStreamConstants.END_ELEMENT:
+					return event;
+				case XMLStreamConstants.END_DOCUMENT:
+					return null;
+				case XMLStreamConstants.SPACE:
+				case XMLStreamConstants.COMMENT:
+				case XMLStreamConstants.PROCESSING_INSTRUCTION:
+					continue;
+				case XMLStreamConstants.CDATA:
+				case XMLStreamConstants.CHARACTERS:
+					if (!event.asCharacters().isWhiteSpace()) {
+						throw new XMLStreamException(
+								"Non-ignorable whitespace CDATA or CHARACTERS event in nextTag()");
+					}
+					break;
+				default:
+					throw new XMLStreamException("Received event [" + event +
+							"], instead of START_ELEMENT or END_ELEMENT.");
+			}
+		}
+	}
+
+	/**
+	 * Throws an {@code IllegalArgumentException} when called.
+	 * @throws IllegalArgumentException when called.
+	 */
+	@Override
+	public Object getProperty(String name) throws IllegalArgumentException {
+		throw new IllegalArgumentException("Property not supported: [" + name + "]");
+	}
+
+	@Override
+	public void close() {
+	}
+}

spring-web-reactive/src/main/java/org/springframework/core/codec/support/XmlEventDecoder.java

@@ -0,0 +1,145 @@
+/*
+ * Copyright 2002-2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.core.codec.support;
+
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Function;
+import javax.xml.stream.XMLEventReader;
+import javax.xml.stream.XMLInputFactory;
+import javax.xml.stream.XMLStreamException;
+import javax.xml.stream.events.XMLEvent;
+import javax.xml.stream.util.XMLEventAllocator;
+
+import com.fasterxml.aalto.AsyncByteBufferFeeder;
+import com.fasterxml.aalto.AsyncXMLInputFactory;
+import com.fasterxml.aalto.AsyncXMLStreamReader;
+import com.fasterxml.aalto.evt.EventAllocatorImpl;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.support.DataBufferUtils;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.MimeType;
+import org.springframework.util.MimeTypeUtils;
+
+/**
+ * Decodes a {@link DataBuffer} stream into a stream of {@link XMLEvent}s. That is, given
+ * the following XML:
+ * <pre>{@code
+ * <root>
+ *     <child>foo</child>
+ *     <child>bar</child>
+ * </root>}
+ * </pre>
+ * this method with result in a flux with the following events:
+ * <ol>
+ * <li>{@link javax.xml.stream.events.StartDocument}</li>
+ * <li>{@link javax.xml.stream.events.StartElement} {@code root}</li>
+ * <li>{@link javax.xml.stream.events.StartElement} {@code child}</li>
+ * <li>{@link javax.xml.stream.events.Characters} {@code foo}</li>
+ * <li>{@link javax.xml.stream.events.EndElement} {@code child}</li>
+ * <li>{@link javax.xml.stream.events.StartElement} {@code child}</li>
+ * <li>{@link javax.xml.stream.events.Characters} {@code bar}</li>
+ * <li>{@link javax.xml.stream.events.EndElement} {@code child}</li>
+ * <li>{@link javax.xml.stream.events.EndElement} {@code root}</li>
+ * </ol>
+ *
+ * Note that this decoder is not registered by default, but used internally by other
+ * decoders who are.
+ *
+ * @author Arjen Poutsma
+ */
+public class XmlEventDecoder extends AbstractDecoder<XMLEvent> {
+
+	private static final boolean aaltoPresent = ClassUtils
+			.isPresent("com.fasterxml.aalto.AsyncXMLStreamReader",
+					XmlEventDecoder.class.getClassLoader());
+
+	private static final XMLInputFactory inputFactory = XMLInputFactory.newFactory();
+
+	public XmlEventDecoder() {
+		super(MimeTypeUtils.APPLICATION_XML, MimeTypeUtils.TEXT_XML);
+	}
+
+	@Override
+	public Flux<XMLEvent> decode(Publisher<DataBuffer> inputStream, ResolvableType type,
+			MimeType mimeType, Object... hints) {
+		if (aaltoPresent) {
+			return Flux.from(inputStream).flatMap(new AaltoDataBufferToXmlEvent());
+		}
+		else {
+			try {
+				InputStream blockingStream = DataBufferUtils.toInputStream(inputStream);
+
+				XMLEventReader eventReader =
+						inputFactory.createXMLEventReader(blockingStream);
+
+				return Flux.fromIterable((Iterable<XMLEvent>) () -> eventReader);
+			}
+			catch (XMLStreamException ex) {
+				return Flux.error(ex);
+			}
+		}
+	}
+
+	/*
+	 * Separate static class to isolate Aalto dependency.
+	 */
+	private static class AaltoDataBufferToXmlEvent
+			implements Function<DataBuffer, Publisher<? extends XMLEvent>> {
+
+		private static final AsyncXMLInputFactory inputFactory =
+				(AsyncXMLInputFactory) XmlEventDecoder.inputFactory;
+
+		private final AsyncXMLStreamReader<AsyncByteBufferFeeder> streamReader =
+				inputFactory.createAsyncForByteBuffer();
+
+		private final XMLEventAllocator eventAllocator =
+				EventAllocatorImpl.getDefaultInstance();
+
+		@Override
+		public Publisher<? extends XMLEvent> apply(DataBuffer dataBuffer) {
+			try {
+				streamReader.getInputFeeder().feedInput(dataBuffer.asByteBuffer());
+				List<XMLEvent> events = new ArrayList<>();
+				while (true) {
+					if (streamReader.next() == AsyncXMLStreamReader.EVENT_INCOMPLETE) {
+						// no more events with what currently has been fed to the reader
+						break;
+					}
+					else {
+						XMLEvent event = eventAllocator.allocate(streamReader);
+						events.add(event);
+						if (event.isEndDocument()) {
+							break;
+						}
+					}
+				}
+				return Flux.fromIterable(events);
+			}
+			catch (XMLStreamException ex) {
+				return Mono.error(ex);
+			}
+
+		}
+	}
+}

spring-web-reactive/src/main/java/org/springframework/web/reactive/method/annotation/RequestMappingHandlerAdapter.java

@@ -31,6 +31,7 @@ import org.springframework.beans.factory.InitializingBean;
 import org.springframework.core.codec.Decoder;
 import org.springframework.core.codec.support.ByteBufferDecoder;
 import org.springframework.core.codec.support.JacksonJsonDecoder;
+import org.springframework.core.codec.support.Jaxb2Decoder;
 import org.springframework.core.codec.support.JsonObjectDecoder;
 import org.springframework.core.codec.support.StringDecoder;
 import org.springframework.core.convert.ConversionService;
@@ -100,7 +101,7 @@ public class RequestMappingHandlerAdapter implements HandlerAdapter, Initializin
 		if (ObjectUtils.isEmpty(this.argumentResolvers)) {
 
 			List<Decoder<?>> decoders = Arrays.asList(new ByteBufferDecoder(),
-					new StringDecoder(),
+					new StringDecoder(), new Jaxb2Decoder(),
 					new JacksonJsonDecoder(new JsonObjectDecoder()));
 
 			this.argumentResolvers.add(new RequestParamArgumentResolver());