Add Spring Data projection support

This commit introduces a `ProjectingMessageConverter` that supports binding
`String` and `byte[]`-backed JSON strings to Spring Data Projection interfaces.
This allows very selective, and low-coupled bindings to data including
the lookup of values from multiple places inside the JSON document.
For example the following interface can be defined as message payload type:

```
interface SomeSample {

  @JsonPath({ "$.username", "$.user.name" })
  String getUsername();
}
```

Accessor methods will be used to lookup the property name as field in the
received JSON document by default.
The @JsonPath expression allows to customize the value lookup and even
to define multiple JSONPath expression to lookup values from multiple
places until an expression returns an actual value.

If the type we're supposed to unmarshal into is a plain class the
`ProjectingMessageConverter` delegates to the default `StringJsonMessageConverter`.

Added Spring Data and the transitively required Jayway JSONPath
library as optional build dependencies.

The `ProjectingMessageConverter` now delegates writes to
a `StringJsonMessageConverter`.

* Some code style polishing

Author: odrotbohm

build.gradle

@@ -73,13 +73,15 @@ subprojects { subproject ->
 		assertjVersion = '2.6.0'
 		hamcrestVersion = '1.3'
 		jacksonVersion = '2.9.1'
+		jaywayJsonPathVersion = '2.4.0'
 		junitVersion = '4.12'
 		kafkaVersion = '1.0.0'
 		mockitoVersion = '2.11.0'
 		scalaVersion = '2.11'
 		slf4jVersion = '1.7.25'
 		springRetryVersion = '1.2.1.RELEASE'
 		springVersion = '5.0.2.RELEASE'
+		springDataCommonsVersion = '2.0.2.RELEASE'
 
 		idPrefix = 'kafka'
 
@@ -170,6 +172,10 @@ project ('spring-kafka') {
 		compile ("com.fasterxml.jackson.core:jackson-core:$jacksonVersion", optional)
 		compile ("com.fasterxml.jackson.core:jackson-databind:$jacksonVersion", optional)
 
+		// Spring Data projection message binding support
+		compile ("org.springframework.data:spring-data-commons:$springDataCommonsVersion", optional)
+		compile ("com.jayway.jsonpath:json-path:$jaywayJsonPathVersion", optional)
+
 		testCompile project (":spring-kafka-test")
 		testCompile "org.assertj:assertj-core:$assertjVersion"
 		testCompile "org.springframework:spring-tx:$springVersion"

spring-kafka/src/main/java/org/springframework/kafka/support/converter/ProjectingMessageConverter.java

@@ -0,0 +1,114 @@
+/*
+ * Copyright 2018 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.kafka.support.converter;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.lang.reflect.Type;
+import java.nio.charset.StandardCharsets;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.data.projection.MethodInterceptorFactory;
+import org.springframework.data.projection.ProjectionFactory;
+import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
+import org.springframework.data.web.JsonProjectingMethodInterceptorFactory;
+import org.springframework.kafka.support.KafkaNull;
+import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.jayway.jsonpath.spi.mapper.JacksonMappingProvider;
+
+/**
+ * A {@link MessageConverter} implementation that uses a Spring Data
+ * {@link ProjectionFactory} to bind incoming messages to projection interfaces.
+ *
+ * @author Oliver Gierke
+ *
+ * @since 2.1.1
+ */
+public class ProjectingMessageConverter extends MessagingMessageConverter {
+
+	private final ProjectionFactory projectionFactory;
+
+	private final MessagingMessageConverter delegate;
+
+	/**
+	 * Creates a new {@link ProjectingMessageConverter} using the given {@link ObjectMapper}.
+	 * @param mapper must not be {@literal null}.
+	 */
+	public ProjectingMessageConverter(ObjectMapper mapper) {
+		Assert.notNull(mapper, "ObjectMapper must not be null");
+
+		JacksonMappingProvider provider = new JacksonMappingProvider(mapper);
+		MethodInterceptorFactory interceptorFactory = new JsonProjectingMethodInterceptorFactory(provider);
+
+		SpelAwareProxyProjectionFactory factory = new SpelAwareProxyProjectionFactory();
+		factory.registerMethodInvokerFactory(interceptorFactory);
+
+		this.projectionFactory = factory;
+		this.delegate = new StringJsonMessageConverter(mapper);
+	}
+
+	@Override
+	protected Object convertPayload(Message<?> message) {
+		return this.delegate.convertPayload(message);
+	}
+
+	@Override
+	protected Object extractAndConvertValue(ConsumerRecord<?, ?> record, Type type) {
+		Object value = record.value();
+
+		if (value == null) {
+			return KafkaNull.INSTANCE;
+		}
+
+		Class<?> rawType = ResolvableType.forType(type).resolve(Object.class);
+
+		if (!rawType.isInterface()) {
+			return this.delegate.extractAndConvertValue(record, type);
+		}
+
+		InputStream inputStream = new ByteArrayInputStream(getAsByteArray(value));
+
+		// The inputStream is closed underneath by the ObjectMapper#_readTreeAndClose()
+		return this.projectionFactory.createProjection(rawType, inputStream);
+	}
+
+	/**
+	 * Return the given source value as byte array.
+	 * @param source must not be {@literal null}.
+	 * @return the source instance as byte array.
+	 */
+	private static byte[] getAsByteArray(Object source) {
+		Assert.notNull(source, "Source must not be null");
+
+		if (source instanceof String) {
+			return String.class.cast(source).getBytes(StandardCharsets.UTF_8);
+		}
+
+		if (source instanceof byte[]) {
+			return byte[].class.cast(source);
+		}
+
+		throw new ConversionException(String.format("Unsupported payload type '%s'. Expected 'String' or 'byte[]'",
+				source.getClass()), null);
+	}
+
+}

spring-kafka/src/test/java/org/springframework/kafka/support/converter/ProjectingMessageConverterTests.java

@@ -0,0 +1,161 @@
+/*
+ * Copyright 2018 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.kafka.support.converter;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.doReturn;
+
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
+import org.springframework.data.web.JsonPath;
+import org.springframework.kafka.support.KafkaNull;
+import org.springframework.messaging.support.MessageBuilder;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.jayway.jsonpath.DocumentContext;
+
+/**
+ * @author Oliver Gierke
+ *
+ * @since 2.1.1
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class ProjectingMessageConverterTests {
+
+	private static final String STRING_PAYLOAD =
+			"{ \"username\" : \"SomeUsername\", \"user\" : { \"name\" : \"SomeName\"}}";
+
+	private static final byte[] BYTE_ARRAY_PAYLOAD = STRING_PAYLOAD.getBytes(StandardCharsets.UTF_8);
+
+	private final ProjectingMessageConverter converter = new ProjectingMessageConverter(new ObjectMapper());
+
+	@Mock
+	private ConsumerRecord<?, ?> record;
+
+	@Rule
+	public ExpectedException exception = ExpectedException.none();
+
+	@Test
+	public void rejectsNullObjectMapper() {
+		this.exception.expect(IllegalArgumentException.class);
+		new ProjectingMessageConverter(null);
+	}
+
+	@Test
+	public void returnsKafkaNullForNullPayload() {
+		doReturn(null).when(this.record).value();
+
+		assertThat(this.converter.extractAndConvertValue(this.record, Object.class)).isEqualTo(KafkaNull.INSTANCE);
+	}
+
+	@Test
+	public void createsProjectedPayloadForInterface() {
+		assertProjectionProxy(STRING_PAYLOAD);
+		assertProjectionProxy(BYTE_ARRAY_PAYLOAD);
+	}
+
+	@Test
+	public void usesJacksonToCreatePayloadForClass() {
+		assertSimpleObject(STRING_PAYLOAD);
+		assertSimpleObject(BYTE_ARRAY_PAYLOAD);
+	}
+
+	@Test
+	public void rejectsInvalidPayload() {
+		this.exception.expect(ConversionException.class);
+		this.exception.expectMessage(Object.class.getName());
+
+		assertProjectionProxy(new Object());
+	}
+
+	@Test
+	public void writesProjectedPayloadUsingJackson() {
+		Map<String, Object> values = new HashMap<>();
+		values.put("username", "SomeUsername");
+		values.put("name", "SomeName");
+
+		Sample sample = new SpelAwareProxyProjectionFactory().createProjection(Sample.class, values);
+
+		Object payload = this.converter.convertPayload(MessageBuilder.withPayload(sample).build());
+
+		DocumentContext json = com.jayway.jsonpath.JsonPath.parse(payload.toString());
+
+		assertThat(json.read("$.username", String.class)).isEqualTo("SomeUsername");
+		assertThat(json.read("$.name", String.class)).isEqualTo("SomeName");
+	}
+
+	private void assertProjectionProxy(Object payload) {
+		doReturn(payload).when(this.record).value();
+
+		Object value = this.converter.extractAndConvertValue(this.record, Sample.class);
+
+		assertThat(value).isInstanceOf(Sample.class);
+
+		Sample sample = (Sample) value;
+
+		assertThat(sample.getName()).isEqualTo("SomeName");
+		assertThat(sample.getUsername()).isEqualTo("SomeUsername");
+	}
+
+	private void assertSimpleObject(Object payload) {
+		doReturn(payload).when(this.record).value();
+
+		Object value = this.converter.extractAndConvertValue(this.record, AnotherSample.class);
+
+		assertThat(value).isInstanceOf(AnotherSample.class);
+
+		AnotherSample sample = (AnotherSample) value;
+
+		assertThat(sample.user.name).isEqualTo("SomeName");
+		assertThat(sample.username).isEqualTo("SomeUsername");
+	}
+
+	interface Sample {
+
+		String getUsername();
+
+		@JsonPath("$.user.name")
+		String getName();
+
+	}
+
+	public static class AnotherSample {
+
+		public String username;
+
+		public User user;
+
+		public static class User {
+
+			public String name;
+
+		}
+
+	}
+
+}