Package com.adaptris.aws2.kinesis

Class KinesisSDKStreamProducer

java.lang.Object
com.adaptris.core.AdaptrisMessageWorkerImp
com.adaptris.core.AdaptrisMessageProducerImp
com.adaptris.core.ProduceOnlyProducerImp
com.adaptris.aws2.kinesis.KinesisSDKStreamProducer
All Implemented Interfaces:
AdaptrisComponent, AdaptrisMessageProducer, AdaptrisMessageSender, AdaptrisMessageWorker, ComponentLifecycle, ComponentLifecycleExtension, MessageEventGenerator
@ComponentProfile(summary="Produce to Amazon Kinesis using the SDK", tag="amazon,aws,kinesis,producer", recommended=AWSKinesisSDKConnection.class) public class KinesisSDKStreamProducer extends ProduceOnlyProducerImp
Producer to amazon kinesis using the SDK.

This may be the preferred approach over using KPL if you're running in environment where you don't want other processes to be spawned (for example: containerised).

In the adapter configuration file this class is aliased as aws-kinesis-sdk-stream-producer which is the preferred alternative to the fully qualified classname when building your configuration.

  • Field Details

  • Constructor Details

    • KinesisSDKStreamProducer

      public KinesisSDKStreamProducer()

  • Method Details

    • prepare

      public void prepare() throws CoreException
      Throws:
      CoreException

    • withStream

      public <T extends KinesisSDKStreamProducer> T withStream(String s)

    • setPartitionKey

      public void setPartitionKey(String partitionKey)

    • withPartitionKey

      public <T extends KinesisSDKStreamProducer> T withPartitionKey(String s)

    • withRequestBuilder

      public <T extends KinesisSDKStreamProducer> T withRequestBuilder(RequestBuilder r)

    • withBatchWindow

      public <T extends KinesisSDKStreamProducer> T withBatchWindow(Integer i)

    • withCreateIfNotExists

      public <T extends KinesisSDKStreamProducer> T withCreateIfNotExists(boolean createIfNotExists)

    • withShardCount

      public <T extends KinesisSDKStreamProducer> T withShardCount(int shardCount)
      If createIfNotExists is true, when creating the stream, this is the shard count
      Type Parameters:
      T -
      Parameters:
      shardCount -
      Returns:

    • withCreateStreamMaxWaitTimeMillis

      public <T extends KinesisSDKStreamProducer> T withCreateStreamMaxWaitTimeMillis(long millis)
      If createIfNotExists is true, when creating the stream, this is the max wait time for the stream to become active
      Type Parameters:
      T -
      Parameters:
      millis -
      Returns:

    • withCreateStreamPollTimeMillis

      public <T extends KinesisSDKStreamProducer> T withCreateStreamPollTimeMillis(long millis)
      If createIfNotExists is true, when creating the stream, this is the wait time between polling to check if the stream is active
      Type Parameters:
      T -
      Parameters:
      millis -
      Returns:

    • doProduce

      protected void doProduce(AdaptrisMessage msg, String endpoint) throws ProduceException
      Specified by:
      doProduce in class ProduceOnlyProducerImp
      Throws:
      ProduceException

    • doSend

      protected void doSend(software.amazon.awssdk.services.kinesis.KinesisClient kinesisClient, String endpoint, List<software.amazon.awssdk.services.kinesis.model.PutRecordsRequestEntry> putRecordsRequestEntryList) throws ProduceException
      Throws:
      ProduceException

    • endpoint

      public String endpoint(AdaptrisMessage msg) throws ProduceException
      Specified by:
      endpoint in class AdaptrisMessageProducerImp
      Throws:
      ProduceException

    • getCreateStreamMaxWaitTimeMillis

      public long getCreateStreamMaxWaitTimeMillis()

    • setCreateStreamMaxWaitTimeMillis

      public void setCreateStreamMaxWaitTimeMillis(long createStreamMaxWaitTimeMillis)

    • getCreateStreamPollTimeMillis

      public long getCreateStreamPollTimeMillis()

    • setCreateStreamPollTimeMillis

      public void setCreateStreamPollTimeMillis(long createStreamPollTimeMillis)

    • getStream

      public String getStream()
      The kinesis stream name.

    • setStream

      public void setStream(String stream)
      The kinesis stream name.

    • getPartitionKey

      public String getPartitionKey()
      The kinesis partition key.

    • getRequestBuilder

      public RequestBuilder getRequestBuilder()
      Request Builder enables the control on how the records are put, this is used in conjunction with batchWindow.

    • setRequestBuilder

      public void setRequestBuilder(RequestBuilder requestBuilder)
      Request Builder enables the control on how the records are put, this is used in conjunction with batchWindow.

    • getBatchWindow

      public Integer getBatchWindow()
      Batch window controls on how the records are put to the stream, this used in conjunction with a splitting implementation of the request builder.

    • setBatchWindow

      public void setBatchWindow(Integer batchWindow)
      Batch window controls on how the records are put to the stream, this used in conjunction with a splitting implementation of the request builder.

    • getCreateIfNotExists

      public boolean getCreateIfNotExists()
      Toggles whether to create the stream if it does not exist at the time of sending

    • setCreateIfNotExists

      public void setCreateIfNotExists(boolean createIfNotExists)
      Toggles whether to create the stream if it does not exist at the time of sending

    • getShardCount

      public int getShardCount()
      If creating a stream, this specifies the number of shards for StreamMode.PROVISIONED. A value of 0 means StreamMode.ON_DEMAND.

    • setShardCount

      public void setShardCount(int shardCount)
      If creating a stream, this specifies the number of shards for StreamMode.PROVISIONED. A value of 0 means StreamMode.ON_DEMAND.