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

Polishing:

Browse Source 
* Always call provided `AsyncHandler` from the internal instance
in the `KinesisMessageHandler`
* Make the `KinesisMessageHandler.obtainAsyncHandler()` as generic method
* Rename `sendFailureChannel` property just to the `failureChannel`
since the real operation is `put` not send
* Add `AwsHeaders.SERVICE_RESULT` to represent the service result, e.g.
in case of `PutRecordsRequest` in the `KinesisMessageHandler` to
send on success the whole `PutRecordsResult`
* Fix README to reflect the current reality of the code
* Fix `KinesisMessageHandlerTests` for provided `AsyncHandler` verification
This commit is contained in:
Artem Bilan
2017-11-16 10:33:36 -05:00
parent 281fad8330
commit d07441134b
6 changed files with 109 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
README.md
View File

@@ -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;
}
````

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

@@ -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<PutRecordsRequest, PutRecordsResult>) getAsyncHandler(message,
(PutRecordsRequest) message.getPayload()));
AsyncHandler<PutRecordsRequest, PutRecordsResult> 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<PutRecordRequest, PutRecordResult>) getAsyncHandler(message, putRecordRequest));
AsyncHandler<PutRecordRequest, PutRecordResult> 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<? extends AmazonWebServiceRequest, ?> getAsyncHandler(final Message<?> message,
final AmazonWebServiceRequest request) {
if (this.asyncHandler != null) {
return this.asyncHandler;
}
else {
return new AsyncHandler<AmazonWebServiceRequest, AmazonWebServiceResult>() {
private <REQUEST extends AmazonWebServiceRequest, RESULT extends AmazonWebServiceResult<?extends ResponseMetadata>> AsyncHandler<REQUEST, RESULT> 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<REQUEST, RESULT>() {
@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<REQUEST, RESULT>) 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());
}
};
}
}
};
}
}

7
src/main/java/org/springframework/integration/aws/support/AwsHeaders.java
View File

@@ -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";
}

5
src/main/java/org/springframework/integration/aws/support/AwsRequestFailureException.java
View File

@@ -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 + "]";
}
}

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

@@ -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<PutRecordRequest> putRecordRequestArgumentCaptor =
ArgumentCaptor.forClass(PutRecordRequest.class);
ArgumentCaptor<AsyncHandler<PutRecordRequest, PutRecordResult>> asyncHandlerArgumentCaptor =
ArgumentCaptor.forClass((Class<AsyncHandler<PutRecordRequest, PutRecordResult>>) (Class<?>) AsyncHandler.class);
verify(this.amazonKinesis).putRecordAsync(putRecordRequestArgumentCaptor.capture(),
eq((AsyncHandler<PutRecordRequest, PutRecordResult>) 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<PutRecordsRequest> putRecordsRequestArgumentCaptor =
ArgumentCaptor.forClass(PutRecordsRequest.class);
verify(this.amazonKinesis).putRecordsAsync(putRecordsRequestArgumentCaptor.capture(),
eq((AsyncHandler<PutRecordsRequest, PutRecordsResult>) this.asyncHandler));
verify(this.amazonKinesis).putRecordsAsync(putRecordsRequestArgumentCaptor.capture(), any(AsyncHandler.class));
PutRecordsRequest putRecordsRequest = putRecordsRequestArgumentCaptor.getValue();

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

@@ -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<Object, byte[]>() {
private SerializingConverter serializingConverter = new SerializingConverter();