Skip to content

Commit 603202f

Browse files
author
Adrian Chlebosz
committed
GH-2068: Introduce amendments after code review
1 parent 9003aa5 commit 603202f

File tree

2 files changed

+30
-36
lines changed

2 files changed

+30
-36
lines changed

‎spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/ConvertingAndDelegatingMessageListenerAdapter.java‎ renamed to ‎spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/ConvertingMessageListener.java‎

Lines changed: 15 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@
1818

1919
import java.util.Arrays;
2020
import java.util.Map;
21-
import java.util.Objects;
2221
import java.util.stream.Collectors;
2322

2423
import org.apache.kafka.clients.consumer.Consumer;
@@ -35,6 +34,7 @@
3534
import org.springframework.messaging.converter.MessageConverter;
3635
import org.springframework.messaging.converter.SimpleMessageConverter;
3736
import org.springframework.messaging.support.GenericMessage;
37+
import org.springframework.util.Assert;
3838

3939
/**
4040
* A {@link AcknowledgingConsumerAwareMessageListener} adapter that implements
@@ -49,9 +49,9 @@
4949
* @since 3.0
5050
* @see AcknowledgingConsumerAwareMessageListener
5151
*/
52-
public class ConvertingAndDelegatingMessageListenerAdapter<T, U, V> implements AcknowledgingConsumerAwareMessageListener<T, U> {
52+
public class ConvertingMessageListener<T, U, V> implements AcknowledgingConsumerAwareMessageListener<T, U> {
5353

54-
private final Object delegate;
54+
private final MessageListener<T, V> delegate;
5555
private final MessageConverter messageConverter;
5656
private final Class<V> desiredValueType;
5757

@@ -63,9 +63,9 @@ public class ConvertingAndDelegatingMessageListenerAdapter<T, U, V> implements A
6363
* @param delegate the {@link MessageListener} to use when passing converted {@link ConsumerRecord} further.
6464
* @param desiredValueType the {@link Class} setting desired type of {@link ConsumerRecord}'s value.
6565
*/
66-
public ConvertingAndDelegatingMessageListenerAdapter(Object delegate, Class<V> desiredValueType) {
67-
validateMessageListener(delegate);
68-
Objects.requireNonNull(desiredValueType);
66+
public ConvertingMessageListener(MessageListener<T, V> delegate, Class<V> desiredValueType) {
67+
Assert.notNull(delegate, "Delegate message listener cannot be null");
68+
Assert.notNull(desiredValueType, "Desired value type cannot be null");
6969

7070
this.delegate = delegate;
7171
this.desiredValueType = desiredValueType;
@@ -81,38 +81,30 @@ public ConvertingAndDelegatingMessageListenerAdapter(Object delegate, Class<V> d
8181
* @param messageConverter the {@link MessageConverter} to use for conversion.
8282
* @param desiredValueType the {@link Class} setting desired type of {@link ConsumerRecord}'s value.
8383
*/
84-
public ConvertingAndDelegatingMessageListenerAdapter(Object delegate, MessageConverter messageConverter, Class<V> desiredValueType) {
85-
validateMessageListener(delegate);
86-
Objects.requireNonNull(messageConverter);
87-
Objects.requireNonNull(desiredValueType);
84+
public ConvertingMessageListener(MessageListener<T, V> delegate, MessageConverter messageConverter, Class<V> desiredValueType) {
85+
Assert.notNull(delegate, "Delegate message listener cannot be null");
86+
Assert.notNull(messageConverter, "Message converter cannot be null");
87+
Assert.notNull(desiredValueType, "Desired value type cannot be null");
8888

8989
this.delegate = delegate;
9090
this.messageConverter = messageConverter;
9191
this.desiredValueType = desiredValueType;
9292
}
9393

94-
private void validateMessageListener(Object messageListener) {
95-
Objects.requireNonNull(messageListener);
96-
if (!(messageListener instanceof MessageListener)) {
97-
throw new IllegalArgumentException("Passed message listener must be of MessageListener type");
98-
}
99-
}
100-
10194
@Override
10295
public void onMessage(ConsumerRecord<T, U> data, Acknowledgment acknowledgment, Consumer<?, ?> consumer) { // NOSONAR
10396
ConsumerRecord<T, V> convertedConsumerRecord = convertConsumerRecord(data);
10497
if (this.delegate instanceof AcknowledgingConsumerAwareMessageListener) {
105-
((AcknowledgingConsumerAwareMessageListener<T, V>) this.delegate).onMessage(convertedConsumerRecord, acknowledgment, consumer);
98+
this.delegate.onMessage(convertedConsumerRecord, acknowledgment, consumer);
10699
}
107100
else if (this.delegate instanceof ConsumerAwareMessageListener) {
108-
((ConsumerAwareMessageListener<T, V>) this.delegate).onMessage(convertedConsumerRecord, consumer);
101+
this.delegate.onMessage(convertedConsumerRecord, consumer);
109102
}
110103
else if (this.delegate instanceof AcknowledgingMessageListener) {
111-
((AcknowledgingMessageListener<T, V>) this.delegate).onMessage(convertedConsumerRecord, acknowledgment);
112-
}
113-
else if (this.delegate instanceof MessageListener) {
114-
((MessageListener<T, V>) this.delegate).onMessage(convertedConsumerRecord);
104+
this.delegate.onMessage(convertedConsumerRecord, acknowledgment);
115105
}
106+
107+
this.delegate.onMessage(convertedConsumerRecord);
116108
}
117109

118110
private ConsumerRecord<T, V> convertConsumerRecord(ConsumerRecord<T, U> data) { // NOSONAR

‎spring-kafka/src/test/java/org/springframework/kafka/listener/adapter/ConvertingAndDelegatingMessageListenerAdapterTest.java‎ renamed to ‎spring-kafka/src/test/java/org/springframework/kafka/listener/adapter/ConvertingMessageListenerTests.java‎

Lines changed: 15 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -34,21 +34,21 @@
3434
* @since 3.0.0
3535
*
3636
*/
37-
class ConvertingAndDelegatingMessageListenerAdapterTest {
37+
class ConvertingMessageListenerTests {
3838

3939
private final ObjectMapper mapper = new ObjectMapper();
4040

4141
@Test
4242
public void testMessageListenerIsInvokedWithConvertedSimpleRecord() {
4343
var consumerRecord = new ConsumerRecord<>("foo", 0, 0, "key", 0);
4444

45-
var messageListener = (MessageListener<String, Integer>) (data) -> assertThat(data.value()).isNotNull();
46-
var adapter = new ConvertingAndDelegatingMessageListenerAdapter<String, Integer, Integer>(
47-
messageListener,
45+
var delegateListener = (MessageListener<String, Integer>) (data) -> assertThat(data.value()).isNotNull();
46+
var convertingMessageListener = new ConvertingMessageListener<String, Integer, Integer>(
47+
delegateListener,
4848
Integer.class
4949
);
5050

51-
adapter.onMessage(consumerRecord, null, null);
51+
convertingMessageListener.onMessage(consumerRecord, null, null);
5252
}
5353

5454
@Test
@@ -57,17 +57,17 @@ public void testMessageListenerIsInvokedWithRecordConvertedByCustomConverter() t
5757
var toBeConvertedJson = mapper.writeValueAsString(toBeConverted);
5858
var consumerRecord = new ConsumerRecord<>("foo", 0, 0, "key", toBeConvertedJson);
5959

60-
var messageListener = (MessageListener<String, ToBeConverted>) (data) -> {
60+
var delegateListener = (MessageListener<String, ToBeConverted>) (data) -> {
6161
assertThat(data.value()).isNotNull();
6262
assertThat(data.value().getA()).isEqualTo("foo");
6363
};
64-
var adapter = new ConvertingAndDelegatingMessageListenerAdapter<String, String, ToBeConverted>(
65-
messageListener,
64+
var convertingMessageListener = new ConvertingMessageListener<String, String, ToBeConverted>(
65+
delegateListener,
6666
new MappingJackson2MessageConverter(),
6767
ToBeConverted.class
6868
);
6969

70-
adapter.onMessage(consumerRecord, null, null);
70+
convertingMessageListener.onMessage(consumerRecord, null, null);
7171
}
7272

7373
@Test
@@ -76,16 +76,18 @@ public void testConversionFailsWhileUsingDefaultConverterForComplexObject() thro
7676
var toBeConvertedJson = mapper.writeValueAsString(toBeConverted);
7777
var consumerRecord = new ConsumerRecord<>("foo", 0, 0, "key", toBeConvertedJson);
7878

79-
var messageListener = (MessageListener<String, ToBeConverted>) (data) -> {
79+
var delegateListener = (MessageListener<String, ToBeConverted>) (data) -> {
8080
assertThat(data.value()).isNotNull();
8181
assertThat(data.value().getA()).isEqualTo("foo");
8282
};
83-
var adapter = new ConvertingAndDelegatingMessageListenerAdapter<String, String, ToBeConverted>(
84-
messageListener,
83+
var convertingMessageListener = new ConvertingMessageListener<String, String, ToBeConverted>(
84+
delegateListener,
8585
ToBeConverted.class
8686
);
8787

88-
assertThatThrownBy(() -> adapter.onMessage(consumerRecord, null, null)).isInstanceOf(MessageConversionException.class);
88+
assertThatThrownBy(
89+
() -> convertingMessageListener.onMessage(consumerRecord, null, null)
90+
).isInstanceOf(MessageConversionException.class);
8991
}
9092

9193
private static class ToBeConverted {

0 commit comments

Comments
 (0)