org.springframework.cloud.stream.binder.kinesis

Class KinesisMessageChannelBinder

java.lang.Object

org.springframework.cloud.stream.binder.AbstractBinder<org.springframework.messaging.MessageChannel,C,P>

org.springframework.cloud.stream.binder.AbstractMessageChannelBinder<org.springframework.cloud.stream.binder.ExtendedConsumerProperties<KinesisConsumerProperties>,org.springframework.cloud.stream.binder.ExtendedProducerProperties<KinesisProducerProperties>,KinesisStreamProvisioner>

org.springframework.cloud.stream.binder.kinesis.KinesisMessageChannelBinder

All Implemented Interfaces:

org.springframework.beans.factory.Aware, org.springframework.beans.factory.InitializingBean, org.springframework.cloud.stream.binder.Binder<org.springframework.messaging.MessageChannel,org.springframework.cloud.stream.binder.ExtendedConsumerProperties<KinesisConsumerProperties>,org.springframework.cloud.stream.binder.ExtendedProducerProperties<KinesisProducerProperties>>, org.springframework.cloud.stream.binder.ExtendedBindingProperties<KinesisConsumerProperties,KinesisProducerProperties>, org.springframework.cloud.stream.binder.ExtendedPropertiesBinder<org.springframework.messaging.MessageChannel,KinesisConsumerProperties,KinesisProducerProperties>, org.springframework.cloud.stream.binder.PollableConsumerBinder<org.springframework.messaging.MessageHandler,org.springframework.cloud.stream.binder.ExtendedConsumerProperties<KinesisConsumerProperties>>, org.springframework.context.ApplicationContextAware, org.springframework.context.ApplicationEventPublisherAware

public class KinesisMessageChannelBinder
extends org.springframework.cloud.stream.binder.AbstractMessageChannelBinder<org.springframework.cloud.stream.binder.ExtendedConsumerProperties<KinesisConsumerProperties>,org.springframework.cloud.stream.binder.ExtendedProducerProperties<KinesisProducerProperties>,KinesisStreamProvisioner>
implements org.springframework.cloud.stream.binder.ExtendedPropertiesBinder<org.springframework.messaging.MessageChannel,KinesisConsumerProperties,KinesisProducerProperties>

The Spring Cloud Stream Binder implementation for AWS Kinesis.

Author:

Peter Oates, Artem Bilan, Arnaud Lecollaire, Dirk Bonhomme, Asiel Caballero

Nested Class Summary

Nested classes/interfaces inherited from class org.springframework.cloud.stream.binder.AbstractMessageChannelBinder

org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.ErrorInfrastructure, org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.PolledConsumerResources

Field Summary

Fields inherited from class org.springframework.cloud.stream.binder.AbstractMessageChannelBinder

provisioningProvider

Fields inherited from class org.springframework.cloud.stream.binder.AbstractBinder

logger

Constructor Summary

Constructors

Constructor and Description

KinesisMessageChannelBinder(KinesisBinderConfigurationProperties configurationProperties, KinesisStreamProvisioner provisioningProvider, com.amazonaws.services.kinesis.AmazonKinesisAsync amazonKinesis, com.amazonaws.auth.AWSCredentialsProvider awsCredentialsProvider, com.amazonaws.services.dynamodbv2.AmazonDynamoDB dynamoDBClient, com.amazonaws.services.dynamodbv2.AmazonDynamoDBStreams dynamoDBStreams, com.amazonaws.services.cloudwatch.AmazonCloudWatch cloudWatchClient)

Method Summary

Modifier and Type	Method and Description
protected org.springframework.integration.core.MessageProducer	createConsumerEndpoint(org.springframework.cloud.stream.provisioning.ConsumerDestination destination, String group, org.springframework.cloud.stream.binder.ExtendedConsumerProperties<KinesisConsumerProperties> properties)
protected org.springframework.messaging.MessageHandler	createProducerMessageHandler(org.springframework.cloud.stream.provisioning.ProducerDestination destination, org.springframework.cloud.stream.binder.ExtendedProducerProperties<KinesisProducerProperties> producerProperties, org.springframework.messaging.MessageChannel errorChannel)
com.amazonaws.services.kinesis.AmazonKinesisAsync	getAmazonKinesis()
String	getDefaultsPrefix()
protected org.springframework.integration.support.ErrorMessageStrategy	getErrorMessageStrategy()
KinesisConsumerProperties	getExtendedConsumerProperties(String channelName)
KinesisProducerProperties	getExtendedProducerProperties(String channelName)
Class<? extends org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider>	getExtendedPropertiesEntryClass()
List<String>	getStreamsInUse()
protected void	onInit()
protected void	postProcessOutputChannel(org.springframework.messaging.MessageChannel outputChannel, org.springframework.cloud.stream.binder.ExtendedProducerProperties<KinesisProducerProperties> producerProperties)
void	setCheckpointStore(org.springframework.integration.metadata.ConcurrentMetadataStore checkpointStore)
void	setExtendedBindingProperties(KinesisExtendedBindingProperties extendedBindingProperties)
void	setKinesisClientLibConfigurations(List<com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration> kinesisClientLibConfigurations)
void	setKinesisProducerConfiguration(com.amazonaws.services.kinesis.producer.KinesisProducerConfiguration kinesisProducerConfiguration)
void	setLockRegistry(org.springframework.integration.support.locks.LockRegistry lockRegistry)

Methods inherited from class org.springframework.cloud.stream.binder.AbstractMessageChannelBinder

afterUnbindConsumer, afterUnbindProducer, bindPollableConsumer, createPolledConsumerResources, createProducerMessageHandler, doBindConsumer, doBindProducer, errorsBaseName, errorsBaseName, getApplicationEventPublisher, getContainerCustomizer, getDefaultErrorMessageHandler, getErrorBridgeName, getErrorBridgeName, getErrorMessageHandler, getErrorMessageHandlerName, getErrorRecovererName, getMessageSourceCustomizer, getPolledConsumerErrorMessageHandler, getPolledConsumerRecoveryCallback, postProcessPollableSource, registerErrorInfrastructure, registerErrorInfrastructure, setApplicationEventPublisher, setConsumerEndpointCustomizer, setProducerMessageHandlerCustomizer, useNativeEncoding

Methods inherited from class org.springframework.cloud.stream.binder.AbstractBinder

afterPropertiesSet, applyPrefix, bindConsumer, bindProducer, buildRetryTemplate, constructDLQName, getApplicationContext, getBeanFactory, getEvaluationContext, groupedName, setApplicationContext

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.springframework.cloud.stream.binder.Binder

bindConsumer, bindProducer

Methods inherited from interface org.springframework.cloud.stream.binder.ExtendedBindingProperties

getBindings

Constructor Detail

KinesisMessageChannelBinder

public KinesisMessageChannelBinder(KinesisBinderConfigurationProperties configurationProperties,
                                   KinesisStreamProvisioner provisioningProvider,
                                   com.amazonaws.services.kinesis.AmazonKinesisAsync amazonKinesis,
                                   com.amazonaws.auth.AWSCredentialsProvider awsCredentialsProvider,
                                   @Nullable
                                   com.amazonaws.services.dynamodbv2.AmazonDynamoDB dynamoDBClient,
                                   @Nullable
                                   com.amazonaws.services.dynamodbv2.AmazonDynamoDBStreams dynamoDBStreams,
                                   @Nullable
                                   com.amazonaws.services.cloudwatch.AmazonCloudWatch cloudWatchClient)

Method Detail

setExtendedBindingProperties

public void setExtendedBindingProperties(KinesisExtendedBindingProperties extendedBindingProperties)

setCheckpointStore

public void setCheckpointStore(org.springframework.integration.metadata.ConcurrentMetadataStore checkpointStore)

setLockRegistry

public void setLockRegistry(org.springframework.integration.support.locks.LockRegistry lockRegistry)

setKinesisProducerConfiguration

public void setKinesisProducerConfiguration(com.amazonaws.services.kinesis.producer.KinesisProducerConfiguration kinesisProducerConfiguration)

setKinesisClientLibConfigurations

public void setKinesisClientLibConfigurations(List<com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration> kinesisClientLibConfigurations)

getExtendedConsumerProperties

public KinesisConsumerProperties getExtendedConsumerProperties(String channelName)

Specified by:

getExtendedConsumerProperties in interface org.springframework.cloud.stream.binder.ExtendedBindingProperties<KinesisConsumerProperties,KinesisProducerProperties>

getExtendedProducerProperties

public KinesisProducerProperties getExtendedProducerProperties(String channelName)

Specified by:

getExtendedProducerProperties in interface org.springframework.cloud.stream.binder.ExtendedBindingProperties<KinesisConsumerProperties,KinesisProducerProperties>

getDefaultsPrefix

public String getDefaultsPrefix()

Specified by:

getDefaultsPrefix in interface org.springframework.cloud.stream.binder.ExtendedBindingProperties<KinesisConsumerProperties,KinesisProducerProperties>

getExtendedPropertiesEntryClass

public Class<? extends org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider> getExtendedPropertiesEntryClass()

Specified by:

getExtendedPropertiesEntryClass in interface org.springframework.cloud.stream.binder.ExtendedBindingProperties<KinesisConsumerProperties,KinesisProducerProperties>

getAmazonKinesis

public com.amazonaws.services.kinesis.AmazonKinesisAsync getAmazonKinesis()

getStreamsInUse

public List<String> getStreamsInUse()

onInit

protected void onInit()
               throws Exception

Overrides:

onInit in class org.springframework.cloud.stream.binder.AbstractBinder<org.springframework.messaging.MessageChannel,org.springframework.cloud.stream.binder.ExtendedConsumerProperties<KinesisConsumerProperties>,org.springframework.cloud.stream.binder.ExtendedProducerProperties<KinesisProducerProperties>>

Throws:

Exception

createProducerMessageHandler

protected org.springframework.messaging.MessageHandler createProducerMessageHandler(org.springframework.cloud.stream.provisioning.ProducerDestination destination,
                                                                                    org.springframework.cloud.stream.binder.ExtendedProducerProperties<KinesisProducerProperties> producerProperties,
                                                                                    org.springframework.messaging.MessageChannel errorChannel)

Specified by:

createProducerMessageHandler in class org.springframework.cloud.stream.binder.AbstractMessageChannelBinder<org.springframework.cloud.stream.binder.ExtendedConsumerProperties<KinesisConsumerProperties>,org.springframework.cloud.stream.binder.ExtendedProducerProperties<KinesisProducerProperties>,KinesisStreamProvisioner>

postProcessOutputChannel

protected void postProcessOutputChannel(org.springframework.messaging.MessageChannel outputChannel,
                                        org.springframework.cloud.stream.binder.ExtendedProducerProperties<KinesisProducerProperties> producerProperties)

Overrides:

postProcessOutputChannel in class org.springframework.cloud.stream.binder.AbstractMessageChannelBinder<org.springframework.cloud.stream.binder.ExtendedConsumerProperties<KinesisConsumerProperties>,org.springframework.cloud.stream.binder.ExtendedProducerProperties<KinesisProducerProperties>,KinesisStreamProvisioner>

createConsumerEndpoint

protected org.springframework.integration.core.MessageProducer createConsumerEndpoint(org.springframework.cloud.stream.provisioning.ConsumerDestination destination,
                                                                                      String group,
                                                                                      org.springframework.cloud.stream.binder.ExtendedConsumerProperties<KinesisConsumerProperties> properties)

Specified by:

createConsumerEndpoint in class org.springframework.cloud.stream.binder.AbstractMessageChannelBinder<org.springframework.cloud.stream.binder.ExtendedConsumerProperties<KinesisConsumerProperties>,org.springframework.cloud.stream.binder.ExtendedProducerProperties<KinesisProducerProperties>,KinesisStreamProvisioner>

getErrorMessageStrategy

protected org.springframework.integration.support.ErrorMessageStrategy getErrorMessageStrategy()

Overrides:

getErrorMessageStrategy in class org.springframework.cloud.stream.binder.AbstractMessageChannelBinder<org.springframework.cloud.stream.binder.ExtendedConsumerProperties<KinesisConsumerProperties>,org.springframework.cloud.stream.binder.ExtendedProducerProperties<KinesisProducerProperties>,KinesisStreamProvisioner>