diff --git a/README.md b/README.md index e589a7b..11fe449 100644 --- a/README.md +++ b/README.md @@ -564,11 +564,15 @@ The `KinesisMessageHandler` is an `AbstractMessageHandler` to perform put record The stream, partition key (or explicit hash key) and sequence number can be determined against request message via evaluation provided expressions or can be specified statically. They also can specified as `AwsHeaders.STREAM`, `AwsHeaders.PARTITION_KEY` and `AwsHeaders.SEQUENCE_NUMBER` respectively. -The `KinesisMessageHandler` can be configured with channels for sending a `Message` on send success (in which the payload is either -the `data` from the `PutRecordRequest` or the full `PutRecordsRequest`), or an `ErrorMessage` on send failure -(in which the payload is `AwsRequestFailureException`). A `com.amazonaws.handlers.AsyncHandler` can also be -provided to the `KinesisMessageHandler` for custom handling after sending record(s) to the stream, but doing so -precludes the usage of such channels. +The `KinesisMessageHandler` can be configured with the `outputChannel` for sending a `Message` on successful put operation. +The payload is the original request and additional `AwsHeaders.SHARD` and `AwsHeaders.SEQUENCE_NUMBER` headers are populated from the `PutRecordResult`. +If the request payload is a `PutRecordsRequest`, the full `PutRecordsResult` is populated in the `AwsHeaders.SERVICE_RESULT` header instead. + +When an async failure is happened on the put operation, the `ErrorMessage` is send to the `failureChannel`. +The payload is an `AwsRequestFailureException`. + +An `com.amazonaws.handlers.AsyncHandler` can also be provided to the `KinesisMessageHandler` for custom handling after putting record(s) to the stream. +This is called independently if `outputChannel` and/or `failureChannel` are provided. The `payload` of request message can be: @@ -581,21 +585,16 @@ The `payload` of request message can be: The Java Configuration for the message handler: ````java -@SpringBootApplication -public static class MyConfiguration { - - @Bean - @ServiceActivator(inputChannel = "kinesisSendChannel") - public MessageHandler kinesisMessageHandler(AmazonKinesis amazonKinesis, - MessageChannel channel, - MessageChannel errorChannel) { - KinesisMessageHandler kinesisMessageHandler = new KinesisMessageHandler(amazonKinesis); - kinesisMessageHandler.setPartitionKey("1"); - kinesisMessageHandler.setOutputChannel(channel); - kinesisMessageHandler.setSendFailureChannel(errorChannel); - return kinesisMessageHandler; - } - +@Bean +@ServiceActivator(inputChannel = "kinesisSendChannel") +public MessageHandler kinesisMessageHandler(AmazonKinesis amazonKinesis, + MessageChannel channel, + MessageChannel errorChannel) { + KinesisMessageHandler kinesisMessageHandler = new KinesisMessageHandler(amazonKinesis); + kinesisMessageHandler.setPartitionKey("1"); + kinesisMessageHandler.setOutputChannel(channel); + kinesisMessageHandler.setFailureChannel(errorChannel); + return kinesisMessageHandler; } ```` 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 247e13e..3ccc074 100644 --- a/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java +++ b/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java @@ -33,6 +33,7 @@ import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.expression.ValueExpression; import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.handler.AbstractMessageProducingHandler; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.integration.support.DefaultErrorMessageStrategy; import org.springframework.integration.support.ErrorMessageStrategy; import org.springframework.messaging.Message; @@ -43,6 +44,7 @@ import org.springframework.util.StringUtils; import com.amazonaws.AmazonWebServiceRequest; import com.amazonaws.AmazonWebServiceResult; +import com.amazonaws.ResponseMetadata; import com.amazonaws.handlers.AsyncHandler; import com.amazonaws.services.kinesis.AmazonKinesisAsync; import com.amazonaws.services.kinesis.model.PutRecordRequest; @@ -55,9 +57,11 @@ 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 */ public class KinesisMessageHandler extends AbstractMessageProducingHandler { @@ -84,9 +88,9 @@ public class KinesisMessageHandler extends AbstractMessageProducingHandler { private Expression sendTimeoutExpression = new ValueExpression<>(DEFAULT_SEND_TIMEOUT); - private MessageChannel sendFailureChannel; + private MessageChannel failureChannel; - private String sendFailureChannelName; + private String failureChannelName; private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy(); @@ -170,38 +174,38 @@ public class KinesisMessageHandler extends AbstractMessageProducingHandler { } /** - * Set the failure channel. After a send failure, an {@link ErrorMessage} will be sent + * Set the failure channel. After a failure on put, an {@link ErrorMessage} will be sent * to this channel with a payload of a {@link AwsRequestFailureException} with the * failed message and cause. - * @param sendFailureChannel the failure channel. + * @param failureChannel the failure channel. * @since 1.1.0 */ - public void setSendFailureChannel(MessageChannel sendFailureChannel) { - this.sendFailureChannel = sendFailureChannel; + public void setFailureChannel(MessageChannel failureChannel) { + this.failureChannel = failureChannel; } - protected MessageChannel getSendFailureChannel() { - if (this.sendFailureChannel != null) { - return this.sendFailureChannel; + protected MessageChannel getFailureChannel() { + if (this.failureChannel != null) { + return this.failureChannel; } - else if (this.sendFailureChannelName != null) { - this.sendFailureChannel = getChannelResolver().resolveDestination(this.sendFailureChannelName); - return this.sendFailureChannel; + else if (this.failureChannelName != null) { + this.failureChannel = getChannelResolver().resolveDestination(this.failureChannelName); + return this.failureChannel; } return null; } /** - * Set the failure channel name. After a send failure, an {@link ErrorMessage} will be + * Set the failure channel name. After a failure on put, an {@link ErrorMessage} will be * sent to this channel name with a payload of a {@link AwsRequestFailureException} * with the failed message and cause. - * @param sendFailureChannelName the failure channel name. + * @param failureChannelName the failure channel name. * @since 1.1.0 */ - public void setSendFailureChannelName(String sendFailureChannelName) { - this.sendFailureChannelName = sendFailureChannelName; + public void setFailureChannelName(String failureChannelName) { + this.failureChannelName = failureChannelName; } @Override @@ -217,17 +221,20 @@ public class KinesisMessageHandler extends AbstractMessageProducingHandler { if (message.getPayload() instanceof PutRecordsRequest) { - resultFuture = this.amazonKinesis.putRecordsAsync((PutRecordsRequest) message.getPayload(), - (AsyncHandler) getAsyncHandler(message, - (PutRecordsRequest) message.getPayload())); + AsyncHandler asyncHandler = + obtainAsyncHandler(message, (PutRecordsRequest) message.getPayload()); + + resultFuture = this.amazonKinesis.putRecordsAsync((PutRecordsRequest) message.getPayload(), asyncHandler); } else { final PutRecordRequest putRecordRequest = (message.getPayload() instanceof PutRecordRequest) ? (PutRecordRequest) message.getPayload() : buildPutRecordRequest(message); - resultFuture = this.amazonKinesis.putRecordAsync(putRecordRequest, - (AsyncHandler) getAsyncHandler(message, putRecordRequest)); + AsyncHandler asyncHandler = + obtainAsyncHandler(message, putRecordRequest); + + resultFuture = this.amazonKinesis.putRecordAsync(putRecordRequest, asyncHandler); } if (this.sync) { @@ -298,43 +305,53 @@ public class KinesisMessageHandler extends AbstractMessageProducingHandler { } @SuppressWarnings("rawtypes") - private AsyncHandler getAsyncHandler(final Message message, - final AmazonWebServiceRequest request) { - if (this.asyncHandler != null) { - return this.asyncHandler; - } - else { - return new AsyncHandler() { + private > AsyncHandler obtainAsyncHandler( + final Message message, final REQUEST request) { - @Override - public void onError(Exception ex) { - if (getSendFailureChannel() != null) { - KinesisMessageHandler.this.messagingTemplate.send(getSendFailureChannel(), - KinesisMessageHandler.this.errorMessageStrategy.buildErrorMessage( - new AwsRequestFailureException(message, request, ex), null)); - } + return new AsyncHandler() { + + @Override + public void onError(Exception ex) { + if (KinesisMessageHandler.this.asyncHandler != null) { + KinesisMessageHandler.this.asyncHandler.onError(ex); } - @Override - public void onSuccess(AmazonWebServiceRequest request, AmazonWebServiceResult result) { - Message resultMessage; + if (getFailureChannel() != null) { + KinesisMessageHandler.this.messagingTemplate.send(getFailureChannel(), + KinesisMessageHandler.this.errorMessageStrategy.buildErrorMessage( + new AwsRequestFailureException(message, request, ex), null)); + } + } + + @Override + @SuppressWarnings("unchecked") + public void onSuccess(REQUEST request, RESULT result) { + if (KinesisMessageHandler.this.asyncHandler != null) { + ((AsyncHandler) KinesisMessageHandler.this.asyncHandler) + .onSuccess(request, result); + } + + if (getOutputChannel() != null) { + AbstractIntegrationMessageBuilder messageBuilder = + getMessageBuilderFactory() + .fromMessage(message); if (result instanceof PutRecordResult) { - resultMessage = getMessageBuilderFactory().fromMessage(message) + messageBuilder .setHeader(AwsHeaders.SHARD, ((PutRecordResult) result).getShardId()) - .setHeader(AwsHeaders.SEQUENCE_NUMBER, ((PutRecordResult) result).getSequenceNumber()) - .build(); + .setHeader(AwsHeaders.SEQUENCE_NUMBER, ((PutRecordResult) result).getSequenceNumber()); } else { - resultMessage = getMessageBuilderFactory().fromMessage(message).build(); + messageBuilder.setHeader(AwsHeaders.SERVICE_RESULT, result); } - if (getOutputChannel() != null) { - KinesisMessageHandler.this.messagingTemplate.send(getOutputChannel(), resultMessage); - } + + KinesisMessageHandler.this.messagingTemplate.send(getOutputChannel(), messageBuilder.build()); } - }; - } + } + + }; } } + diff --git a/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java b/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java index 6409b4e..364825d 100644 --- a/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java +++ b/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -90,4 +90,9 @@ public abstract class AwsHeaders { */ public static final String CHECKPOINTER = PREFIX + "checkpointer"; + /** + * The {@value SERVICE_RESULT} header represents a {@link com.amazonaws.AmazonWebServiceResult}. + */ + public static final String SERVICE_RESULT = PREFIX + "serviceResult"; + } diff --git a/src/main/java/org/springframework/integration/aws/support/AwsRequestFailureException.java b/src/main/java/org/springframework/integration/aws/support/AwsRequestFailureException.java index a9f9cc8..6c94dbd 100644 --- a/src/main/java/org/springframework/integration/aws/support/AwsRequestFailureException.java +++ b/src/main/java/org/springframework/integration/aws/support/AwsRequestFailureException.java @@ -25,7 +25,8 @@ import com.amazonaws.AmazonWebServiceRequest; * An exception that is the payload of an {@code ErrorMessage} when a send fails. * * @author Jacob Severson - * @since 1.1.0 + * + * @since 1.1 */ public class AwsRequestFailureException extends MessagingException { @@ -44,7 +45,7 @@ public class AwsRequestFailureException extends MessagingException { @Override public String toString() { - return super.toString() + " [record=" + this.request + "]"; + return super.toString() + " [request=" + this.request + "]"; } } 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 271f245..2646e10 100644 --- a/src/test/java/org/springframework/integration/aws/outbound/KinesisMessageHandlerTests.java +++ b/src/test/java/org/springframework/integration/aws/outbound/KinesisMessageHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -53,10 +53,10 @@ import com.amazonaws.services.kinesis.model.PutRecordRequest; import com.amazonaws.services.kinesis.model.PutRecordResult; import com.amazonaws.services.kinesis.model.PutRecordsRequest; import com.amazonaws.services.kinesis.model.PutRecordsRequestEntry; -import com.amazonaws.services.kinesis.model.PutRecordsResult; /** * @author Artem Bilan + * * @since 1.1 */ @RunWith(SpringRunner.class) @@ -107,8 +107,11 @@ public class KinesisMessageHandlerTests { ArgumentCaptor putRecordRequestArgumentCaptor = ArgumentCaptor.forClass(PutRecordRequest.class); + ArgumentCaptor> asyncHandlerArgumentCaptor = + ArgumentCaptor.forClass((Class>) (Class) AsyncHandler.class); + verify(this.amazonKinesis).putRecordAsync(putRecordRequestArgumentCaptor.capture(), - eq((AsyncHandler) this.asyncHandler)); + asyncHandlerArgumentCaptor.capture()); PutRecordRequest putRecordRequest = putRecordRequestArgumentCaptor.getValue(); @@ -118,6 +121,13 @@ public class KinesisMessageHandlerTests { assertThat(putRecordRequest.getExplicitHashKey()).isNull(); assertThat(putRecordRequest.getData()).isEqualTo(ByteBuffer.wrap("message".getBytes())); + AsyncHandler asyncHandler = asyncHandlerArgumentCaptor.getValue(); + + RuntimeException testingException = new RuntimeException("testingException"); + asyncHandler.onError(testingException); + + verify(this.asyncHandler).onError(eq(testingException)); + message = new GenericMessage<>(new PutRecordsRequest() .withStreamName("myStream") .withRecords(new PutRecordsRequestEntry() @@ -128,8 +138,7 @@ public class KinesisMessageHandlerTests { ArgumentCaptor putRecordsRequestArgumentCaptor = ArgumentCaptor.forClass(PutRecordsRequest.class); - verify(this.amazonKinesis).putRecordsAsync(putRecordsRequestArgumentCaptor.capture(), - eq((AsyncHandler) this.asyncHandler)); + verify(this.amazonKinesis).putRecordsAsync(putRecordsRequestArgumentCaptor.capture(), any(AsyncHandler.class)); PutRecordsRequest putRecordsRequest = putRecordsRequestArgumentCaptor.getValue(); 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 ca4afee..4437da6 100644 --- a/src/test/java/org/springframework/integration/aws/outbound/KinesisProducingMessageHandlerTests.java +++ b/src/test/java/org/springframework/integration/aws/outbound/KinesisProducingMessageHandlerTests.java @@ -57,7 +57,8 @@ import com.amazonaws.services.kinesis.model.PutRecordsResult; /** * @author Jacob Severson - * @since 1.1.0 + * + * @since 1.1 */ @RunWith(SpringRunner.class) @DirtiesContext @@ -218,7 +219,7 @@ public class KinesisProducingMessageHandlerTests { KinesisMessageHandler kinesisMessageHandler = new KinesisMessageHandler(amazonKinesis()); kinesisMessageHandler.setSync(true); kinesisMessageHandler.setOutputChannel(successChannel()); - kinesisMessageHandler.setSendFailureChannel(errorChannel()); + kinesisMessageHandler.setFailureChannel(errorChannel()); kinesisMessageHandler.setConverter(new Converter() { private SerializingConverter serializingConverter = new SerializingConverter();