Spring Cloud Stream AWS Kinesis Binder: Difference between revisions
No edit summary |
|||
| Line 28: | Line 28: | ||
:::[[Image:SpringCloudStreamAmazonKinesisBinder.png]] | :::[[Image:SpringCloudStreamAmazonKinesisBinder.png]] | ||
The payload passed to the message as follows: | The payload passed to the message as follows: | ||
| Line 49: | Line 47: | ||
} | } | ||
</syntaxhighlight> | </syntaxhighlight> | ||
As a side effect of simply creating the message, the "id" and "timestamp" headers are added by the [[Spring Messaging]] layer: | As a side effect of simply creating the message, the "id" and "timestamp" headers are added by the [[Spring Messaging]] layer: | ||
| Line 60: | Line 56: | ||
As part of the sending sequence, the headers are (by default) "embedded" and serialized together with the payload. Other header modes are possible (see org.springframework.cloud.stream.binder.HeaderMode). See org.springframework.cloud.stream.binder.AbstractMessageChannelBinder serializeAndEmbedHeadersIfApplicable(). | As part of the sending sequence, the headers are (by default) "embedded" and serialized together with the payload. Other header modes are possible (see org.springframework.cloud.stream.binder.HeaderMode). See org.springframework.cloud.stream.binder.AbstractMessageChannelBinder serializeAndEmbedHeadersIfApplicable(). | ||
At the Kinesis layer (org.springframework.integration.aws.outbound.KinesisMessageHandler) a new PutRecordRequest instance is created. This is where the following headers are added: | |||
* "aws_stream" (AwsHeaders.STREAM) | |||
* "aws_partitionKey" (AwsHeaders.PARTITION_KEY) | |||
* "aws_sequenceNumber" (AwsHeaders.SEQUENCE_NUMBER) | |||
=Consumer Group Support= | =Consumer Group Support= | ||
Revision as of 20:14, 18 December 2018
External
- Binder Implementation GitHub https://github.com/spring-cloud/spring-cloud-stream-binder-aws-kinesis
- AWS Kinesis Binder Reference Manual https://github.com/spring-cloud/spring-cloud-stream-binder-aws-kinesis/blob/master/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/overview.adoc
- Example GitHub https://github.com/spring-cloud/spring-cloud-stream-samples/tree/master/kinesis-samples
Internal
Overview
Dependencies
dependencies {
implementation('org.springframework.cloud:spring-cloud-stream-binder-kinesis:1.0.0.RELEASE')
}
Playground Examples
Message Propagation
The payload passed to the message as follows:
MessageChannel outputChannel = ...;
String payload = "..."
Message m = new GenericMessage<>(payload);
outputChannel.send(m);
is recovered at the other end as follows:
@StreamListener(InputChannelFactory.INPUT_CHANNEL_NAME)
public void handle(Message<String> e) {
String payload = e.getPayload();
}
As a side effect of simply creating the message, the "id" and "timestamp" headers are added by the Spring Messaging layer:
"id" -> "4a96cc04-1116-94de-268b-d442b0e86683" "timestamp" -> "1545163131036"
The "contentType" header is added by interceptor, in pre-send.
As part of the sending sequence, the headers are (by default) "embedded" and serialized together with the payload. Other header modes are possible (see org.springframework.cloud.stream.binder.HeaderMode). See org.springframework.cloud.stream.binder.AbstractMessageChannelBinder serializeAndEmbedHeadersIfApplicable().
At the Kinesis layer (org.springframework.integration.aws.outbound.KinesisMessageHandler) a new PutRecordRequest instance is created. This is where the following headers are added:
- "aws_stream" (AwsHeaders.STREAM)
- "aws_partitionKey" (AwsHeaders.PARTITION_KEY)
- "aws_sequenceNumber" (AwsHeaders.SEQUENCE_NUMBER)
Consumer Group Support
Also see:
Configuration
Stream Auto-Creation
1.0.0 cannot be configured to fail on inexistent stream, instead of creating it. See org.springframework.cloud.stream.binder.kinesis.provisioning.KinesisStreamProvisioner line 135.
Also see