From 7a95aca874c1036242d6b48853aa5f6e9480239b Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 12 Sep 2019 15:50:27 -0400 Subject: [PATCH] GH-109: Kinesis MH: Use `MessageConverter` 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) {` --- .../ConvertingFromMessageConverter.java | 51 +++++++++ .../aws/outbound/KinesisMessageHandler.java | 25 ++++- .../aws/outbound/KplMessageHandler.java | 104 ++++++++++++++---- .../outbound/KinesisMessageHandlerTests.java | 13 ++- .../KinesisProducingMessageHandlerTests.java | 21 ++-- 5 files changed, 178 insertions(+), 36 deletions(-) create mode 100644 src/main/java/org/springframework/integration/aws/outbound/ConvertingFromMessageConverter.java diff --git a/src/main/java/org/springframework/integration/aws/outbound/ConvertingFromMessageConverter.java b/src/main/java/org/springframework/integration/aws/outbound/ConvertingFromMessageConverter.java new file mode 100644 index 0000000..49bf0a1 --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/outbound/ConvertingFromMessageConverter.java @@ -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 delegate; + + ConvertingFromMessageConverter(Converter 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(); + } + +} diff --git a/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java b/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java index 778a8de..ef70d69 100644 --- a/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java +++ b/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java @@ -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 { private final AmazonKinesisAsync amazonKinesis; - private Converter converter = new SerializingConverter(); + private MessageConverter messageConverter = new ConvertingFromMessageConverter(new SerializingConverter()); private Expression streamExpression; @@ -79,10 +82,21 @@ public class KinesisMessageHandler extends AbstractAwsMessageHandler { * 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 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 { } } 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); diff --git a/src/main/java/org/springframework/integration/aws/outbound/KplMessageHandler.java b/src/main/java/org/springframework/integration/aws/outbound/KplMessageHandler.java index f2863cb..698226a 100644 --- a/src/main/java/org/springframework/integration/aws/outbound/KplMessageHandler.java +++ b/src/main/java/org/springframework/integration/aws/outbound/KplMessageHandler.java @@ -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 { +public class KplMessageHandler extends AbstractAwsMessageHandler implements Lifecycle { private final KinesisProducer kinesisProducer; - private Converter converter = new SerializingConverter(); + private MessageConverter messageConverter = new ConvertingFromMessageConverter(new SerializingConverter()); private Expression streamExpression; @@ -77,6 +83,12 @@ public class KplMessageHandler extends AbstractAwsMessageHandler { private OutboundMessageMapper 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 { * 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 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 { + "Consider to use 'OutboundMessageMapper' 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 { } } 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); diff --git a/src/test/java/org/springframework/integration/aws/outbound/KinesisMessageHandlerTests.java b/src/test/java/org/springframework/integration/aws/outbound/KinesisMessageHandlerTests.java index cbbcad3..b8bc370 100644 --- a/src/test/java/org/springframework/integration/aws/outbound/KinesisMessageHandlerTests.java +++ b/src/test/java/org/springframework/integration/aws/outbound/KinesisMessageHandlerTests.java @@ -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() { + 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; diff --git a/src/test/java/org/springframework/integration/aws/outbound/KinesisProducingMessageHandlerTests.java b/src/test/java/org/springframework/integration/aws/outbound/KinesisProducingMessageHandlerTests.java index 228098f..16e4b85 100644 --- a/src/test/java/org/springframework/integration/aws/outbound/KinesisProducingMessageHandlerTests.java +++ b/src/test/java/org/springframework/integration/aws/outbound/KinesisProducingMessageHandlerTests.java @@ -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() { + 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; }