Spring/spring-integration-aws
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-109: Kinesis MH: Use MessageConverter

Browse Source 
Fixes https://github.com/spring-projects/spring-integration-aws/issues/109

The plain `Converter` is not able to deal with `MessageHeaders` as a
context for target payload conversion.

* Use `MessageConverter` abstraction to get a request message context
for target conversion
* Deprecate a `setConverter()` in favor of a new `setMessageConverter()`
* Use delegating `ConvertingFromMessageConverter` for default serialization
behavior and for backward compatibility
* Add `flushDuration` option into `KplMessageHandler` for calling a
`kinesisProducer.flush()` periodically
* Call `kinesisProducer.flush()` immediately after `addUserRecord()`
`if (this.flushDuration.toMillis() <= 0) {`
This commit is contained in:
Artem Bilan
2019-09-12 15:50:27 -04:00
parent d9196c32e5
commit 7a95aca874
5 changed files with 178 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
src/main/java/org/springframework/integration/aws/outbound/ConvertingFromMessageConverter.java Normal file
View File

@@ -0,0 +1,51 @@
/*
* Copyright 2017-2019 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
*
* https://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.integration.aws.outbound;
import org.springframework.core.convert.converter.Converter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.util.Assert;
/**
* A simple {@link MessageConverter} that delegates to a {@link Converter}.
*
* @author Artem Bilan
*
* @since 2.3
*/
class ConvertingFromMessageConverter implements MessageConverter {
private final Converter<Object, ?> delegate;
ConvertingFromMessageConverter(Converter<Object, ?> delegate) {
Assert.notNull(delegate, "'delegate' must not be null");
this.delegate = delegate;
}
@Override
public Object fromMessage(Message<?> message, Class<?> targetClass) {
return this.delegate.convert(message.getPayload());
}
@Override
public Message<?> toMessage(Object payload, MessageHeaders headers) {
throw new UnsupportedOperationException();
}
}

25
src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java
View File

@@ -32,6 +32,7 @@ import org.springframework.integration.support.MutableMessage;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConversionException;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -49,7 +50,9 @@ import com.amazonaws.services.kinesis.model.PutRecordsResult;
*
* @author Artem Bilan
* @author Jacob Severson
*
* @since 1.1
*
* @see AmazonKinesisAsync#putRecord(PutRecordRequest)
* @see AmazonKinesisAsync#putRecords(PutRecordsRequest)
* @see com.amazonaws.handlers.AsyncHandler
@@ -58,7 +61,7 @@ public class KinesisMessageHandler extends AbstractAwsMessageHandler<Void> {
private final AmazonKinesisAsync amazonKinesis;
private Converter<Object, byte[]> converter = new SerializingConverter();
private MessageConverter messageConverter = new ConvertingFromMessageConverter(new SerializingConverter());
private Expression streamExpression;
@@ -79,10 +82,21 @@ public class KinesisMessageHandler extends AbstractAwsMessageHandler<Void> {
* Specify a {@link Converter} to serialize {@code payload} to the {@code byte[]} if
* that isn't {@code byte[]} already.
* @param converter the {@link Converter} to use; cannot be null.
* @deprecated since 2.3 in favor of {@link #setMessageConverter}
*/
@Deprecated
public void setConverter(Converter<Object, byte[]> converter) {
Assert.notNull(converter, "'converter' must not be null.");
this.converter = converter;
setMessageConverter(new ConvertingFromMessageConverter(converter));
}
/**
* Configure a {@link MessageConverter} for converting payload to {@code byte[]} for Kinesis record.
* @param messageConverter the {@link MessageConverter} to use.
* @since 2.3
*/
public void setMessageConverter(MessageConverter messageConverter) {
Assert.notNull(messageConverter, "'messageConverter' must not be null.");
this.messageConverter = messageConverter;
}
public void setStream(String stream) {
@@ -211,7 +225,10 @@ public class KinesisMessageHandler extends AbstractAwsMessageHandler<Void> {
}
}
else {
byte[] bytes = payload instanceof byte[] ? (byte[]) payload : this.converter.convert(payload);
byte[] bytes =
(byte[]) (payload instanceof byte[]
? payload
: this.messageConverter.fromMessage(message, byte[].class));
Assert.notNull(bytes, "payload cannot be null");
if (this.embeddedHeadersMapper != null) {
messageToEmbed = new MutableMessage<>(bytes, messageHeaders);

104
src/main/java/org/springframework/integration/aws/outbound/KplMessageHandler.java
View File

@@ -17,8 +17,11 @@
package org.springframework.integration.aws.outbound;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledFuture;
import org.springframework.context.Lifecycle;
import org.springframework.core.convert.converter.Converter;
import org.springframework.core.serializer.support.SerializingConverter;
import org.springframework.expression.Expression;
@@ -33,6 +36,7 @@ import org.springframework.integration.support.MutableMessage;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConversionException;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -56,16 +60,18 @@ import com.google.common.util.concurrent.MoreExecutors;
*
* @author Arnaud Lecollaire
* @author Artem Bilan
* @since 2.2.0
*
* @since 2.2
*
* @see AmazonKinesisAsync#putRecord(PutRecordRequest)
* @see AmazonKinesisAsync#putRecords(PutRecordsRequest)
* @see com.amazonaws.handlers.AsyncHandler
*/
public class KplMessageHandler extends AbstractAwsMessageHandler<Void> {
public class KplMessageHandler extends AbstractAwsMessageHandler<Void> implements Lifecycle {
private final KinesisProducer kinesisProducer;
private Converter<Object, byte[]> converter = new SerializingConverter();
private MessageConverter messageConverter = new ConvertingFromMessageConverter(new SerializingConverter());
private Expression streamExpression;
@@ -77,6 +83,12 @@ public class KplMessageHandler extends AbstractAwsMessageHandler<Void> {
private OutboundMessageMapper<byte[]> embeddedHeadersMapper;
private Duration flushDuration = Duration.ofMillis(0);
private volatile boolean running;
private volatile ScheduledFuture<?> flushFuture;
public KplMessageHandler(KinesisProducer kinesisProducer) {
Assert.notNull(kinesisProducer, "'kinesisProducer' must not be null.");
this.kinesisProducer = kinesisProducer;
@@ -86,10 +98,21 @@ public class KplMessageHandler extends AbstractAwsMessageHandler<Void> {
* Specify a {@link Converter} to serialize {@code payload} to the {@code byte[]} if
* that isn't {@code byte[]} already.
* @param converter the {@link Converter} to use; cannot be null.
@deprecated since 2.3 in favor of {@link #setMessageConverter}
*/
@Deprecated
public void setConverter(Converter<Object, byte[]> converter) {
Assert.notNull(converter, "'converter' must not be null.");
this.converter = converter;
setMessageConverter(new ConvertingFromMessageConverter(converter));
}
/**
* Configure a {@link MessageConverter} for converting payload to {@code byte[]} for Kinesis record.
* @param messageConverter the {@link MessageConverter} to use.
* @since 2.3
*/
public void setMessageConverter(MessageConverter messageConverter) {
Assert.notNull(messageConverter, "'messageConverter' must not be null.");
this.messageConverter = messageConverter;
}
public void setStream(String stream) {
@@ -159,25 +182,59 @@ public class KplMessageHandler extends AbstractAwsMessageHandler<Void> {
+ "Consider to use 'OutboundMessageMapper<byte[]>' for embedding headers into the record data.");
}
@Override
public synchronized void start() {
if (!this.running) {
if (this.flushDuration.toMillis() > 0) {
this.flushFuture = getTaskScheduler()
.scheduleAtFixedRate(this.kinesisProducer::flush, this.flushDuration);
}
this.running = true;
}
}
@Override
public synchronized void stop() {
if (this.running) {
this.running = false;
if (this.flushFuture != null) {
this.flushFuture.cancel(true);
}
}
}
@Override
public boolean isRunning() {
return this.running;
}
@Override
protected Future<?> handleMessageToAws(Message<?> message) {
if (message.getPayload() instanceof PutRecordsRequest) {
throw new UnsupportedOperationException("not implemented");
}
else if (message.getPayload() instanceof UserRecord) {
return handleUserRecord(message, buildPutRecordRequest(message), (UserRecord) message.getPayload());
}
else {
final PutRecordRequest putRecordRequest = (message.getPayload() instanceof PutRecordRequest)
? (PutRecordRequest) message.getPayload() : buildPutRecordRequest(message);
try {
if (message.getPayload() instanceof PutRecordsRequest) {
throw new UnsupportedOperationException("not implemented");
}
else if (message.getPayload() instanceof UserRecord) {
return handleUserRecord(message, buildPutRecordRequest(message), (UserRecord) message.getPayload());
}
else {
final PutRecordRequest putRecordRequest = (message.getPayload() instanceof PutRecordRequest)
? (PutRecordRequest) message.getPayload() : buildPutRecordRequest(message);
// convert the PutRecordRequest to a UserRecord
UserRecord userRecord = new UserRecord();
userRecord.setExplicitHashKey(putRecordRequest.getExplicitHashKey());
userRecord.setData(putRecordRequest.getData());
userRecord.setPartitionKey(putRecordRequest.getPartitionKey());
userRecord.setStreamName(putRecordRequest.getStreamName());
return handleUserRecord(message, putRecordRequest, userRecord);
// convert the PutRecordRequest to a UserRecord
UserRecord userRecord = new UserRecord();
userRecord.setExplicitHashKey(putRecordRequest.getExplicitHashKey());
userRecord.setData(putRecordRequest.getData());
userRecord.setPartitionKey(putRecordRequest.getPartitionKey());
userRecord.setStreamName(putRecordRequest.getStreamName());
return handleUserRecord(message, putRecordRequest, userRecord);
}
}
finally {
if (this.flushDuration.toMillis() <= 0) {
this.kinesisProducer.flush();
}
}
}
@@ -245,7 +302,10 @@ public class KplMessageHandler extends AbstractAwsMessageHandler<Void> {
}
}
else {
byte[] bytes = payload instanceof byte[] ? (byte[]) payload : this.converter.convert(payload);
byte[] bytes =
(byte[]) (payload instanceof byte[]
? payload
: this.messageConverter.fromMessage(message, byte[].class));
Assert.notNull(bytes, "payload cannot be null");
if (this.embeddedHeadersMapper != null) {
messageToEmbed = new MutableMessage<>(bytes, messageHeaders);

13
src/test/java/org/springframework/integration/aws/outbound/KinesisMessageHandlerTests.java
View File

@@ -34,7 +34,6 @@ import org.mockito.ArgumentCaptor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.convert.converter.Converter;
import org.springframework.core.serializer.support.SerializingConverter;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.aws.support.AwsHeaders;
@@ -44,6 +43,8 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.annotation.DirtiesContext;
@@ -178,12 +179,13 @@ public class KinesisMessageHandlerTests {
KinesisMessageHandler kinesisMessageHandler = new KinesisMessageHandler(amazonKinesis());
kinesisMessageHandler.setSync(true);
kinesisMessageHandler.setAsyncHandler(asyncHandler());
kinesisMessageHandler.setConverter(new Converter<Object, byte[]>() {
kinesisMessageHandler.setMessageConverter(new MessageConverter() {
private SerializingConverter serializingConverter = new SerializingConverter();
@Override
public byte[] convert(Object source) {
public Object fromMessage(Message<?> message, Class<?> targetClass) {
Object source = message.getPayload();
if (source instanceof String) {
return ((String) source).getBytes();
}
@@ -192,6 +194,11 @@ public class KinesisMessageHandlerTests {
}
}
@Override
public Message<?> toMessage(Object payload, MessageHeaders headers) {
return null;
}
});
kinesisMessageHandler.setEmbeddedHeadersMapper(new EmbeddedJsonHeadersMessageMapper("foo"));
return kinesisMessageHandler;

21
src/test/java/org/springframework/integration/aws/outbound/KinesisProducingMessageHandlerTests.java
View File

@@ -30,7 +30,6 @@ import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.convert.converter.Converter;
import org.springframework.core.serializer.support.SerializingConverter;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.aws.support.AwsHeaders;
@@ -41,7 +40,9 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.annotation.DirtiesContext;
@@ -173,10 +174,10 @@ public class KinesisProducingMessageHandlerTests {
handler.onSuccess(new PutRecordsRequest(), new PutRecordsResult());
return mock(Future.class);
}).willAnswer(invocation -> {
AsyncHandler<?, ?> handler = invocation.getArgument(1);
handler.onError(new RuntimeException("putRecordsRequestEx"));
return mock(Future.class);
});
AsyncHandler<?, ?> handler = invocation.getArgument(1);
handler.onError(new RuntimeException("putRecordsRequestEx"));
return mock(Future.class);
});
return mock;
}
@@ -198,12 +199,13 @@ public class KinesisProducingMessageHandlerTests {
kinesisMessageHandler.setSync(true);
kinesisMessageHandler.setOutputChannel(successChannel());
kinesisMessageHandler.setFailureChannel(errorChannel());
kinesisMessageHandler.setConverter(new Converter<Object, byte[]>() {
kinesisMessageHandler.setMessageConverter(new MessageConverter() {
private SerializingConverter serializingConverter = new SerializingConverter();
@Override
public byte[] convert(Object source) {
public Object fromMessage(Message<?> message, Class<?> targetClass) {
Object source = message.getPayload();
if (source instanceof String) {
return ((String) source).getBytes();
}
@@ -212,6 +214,11 @@ public class KinesisProducingMessageHandlerTests {
}
}
@Override
public Message<?> toMessage(Object payload, MessageHeaders headers) {
return null;
}
});
return kinesisMessageHandler;
}