dynamodb

README

Streams Driver for Amazon DynamoDB

The stream driver for Amazon DynamoDB which offers a deduplication storage implementation to ensure idempotency for message processing.

It is planned to offer in a near future a Writer implementation to be used by systems implementing the transactional outbox messaging pattern.

Moreover, the Message Egress Proxy (aka. log trailing) component could be used along this driver to publish the messages to the message broker / stream.

Furthermore, Amazon DynamoDB has a Change-Data-Capture stream feature ready to write changes into multiple services such as Lambda and Kinesis. Thus, this feature could be combined along the Message Egress Proxy component in order to stream messages into desired infrastructure non-supported by Amazon DynamoDB Stream feature.

Deduplication Storage Requirements

In order for this driver to work, the database MUST have a deduplication table with the following schema.

{
  "TableName": "deduplication-table",
  "KeySchema": [
    {
      "KeyType": "HASH",
      "AttributeName": "worker_id"
    },
    {
      "KeyType": "RANGE",
      "AttributeName": "message_id"
    }
  ],
  "AttributeDefinitions": [
    {
      "AttributeName": "worker_id",
      "AttributeType": "S"
    },
    {
      "AttributeName": "message_id",
      "AttributeType": "S"
    },
    {
      "AttributeName": "expiration_time",
      "AttributeType": "N"
    }
  ],
  "BillingMode": "PAY_PER_REQUEST"
}

Transactional Outbox Requirements

In order for this driver to work, the database MUST have an outbox table with the following schema.

{
  "TableName": "deduplication-table",
  "KeySchema": [
    {
      "KeyType": "HASH",
      "AttributeName": "worker_id"
    },
    {
      "KeyType": "RANGE",
      "AttributeName": "message_id"
    }
  ],
  "AttributeDefinitions": [
    {
      "AttributeName": "worker_id",
      "AttributeType": "S"
    },
    {
      "AttributeName": "message_id",
      "AttributeType": "S"
    }
  ],
  "BillingMode": "PAY_PER_REQUEST"
}

Documentation

Index

• type DeduplicationStorage
  • func NewDeduplicationStorage(cfg DeduplicationStorageConfig, client *dynamodb.Client) DeduplicationStorage
  • func (d DeduplicationStorage) Commit(ctx context.Context, workerID, messageID string)
  • func (d DeduplicationStorage) IsDuplicated(ctx context.Context, workerID, messageID string) (bool, error)
• type DeduplicationStorageConfig

Types

type DeduplicationStorage

type DeduplicationStorage struct {
	// contains filtered or unexported fields
}

DeduplicationStorage is the Amazon DynamoDB streams.DeduplicationStorage

func NewDeduplicationStorage

func NewDeduplicationStorage(cfg DeduplicationStorageConfig, client *dynamodb.Client) DeduplicationStorage

func (DeduplicationStorage) Commit

func (d DeduplicationStorage) Commit(ctx context.Context, workerID, messageID string)

func (DeduplicationStorage) IsDuplicated

func (d DeduplicationStorage) IsDuplicated(ctx context.Context, workerID, messageID string) (bool, error)

type DeduplicationStorageConfig

type DeduplicationStorageConfig struct {
	TableName      string
	Logger         *log.Logger
	ErrorLogger    *log.Logger
	RowTTLDuration time.Duration // Total duration for a row to be available; DynamoDB will remove the row automatically if configured.
}

DeduplicationStorageConfig is the configuration schema for Amazon DynamoDB streams.DeduplicationStorage implementation.