billydh

package io.codebrews.kotlinkafkadynamodemo.kafka

import org.junit.jupiter.api.Assertions.*
import org.springframework.boot.test.context.SpringBootTest
import org.springframework.kafka.test.context.EmbeddedKafka
import org.springframework.test.annotation.DirtiesContext

@DirtiesContext
@EmbeddedKafka(partitions = 1)
@SpringBootTest

billydh

dependencies {
	implementation("org.springframework.boot:spring-boot-starter-webflux")
	implementation("com.fasterxml.jackson.module:jackson-module-kotlin")
	implementation("org.jetbrains.kotlin:kotlin-reflect")
	implementation("org.jetbrains.kotlin:kotlin-stdlib-jdk8")
	implementation("software.amazon.awssdk:dynamodb")
	implementation("io.projectreactor.kafka:reactor-kafka")
	implementation("io.confluent:kafka-avro-serializer:6.0.0")

	testImplementation("org.springframework.boot:spring-boot-starter-test") {

billydh

    fun updateCustomer(customer: CustomerPersist): Mono<Unit> {
        return Mono.fromFuture(table.updateItem { it.item(customer).ignoreNulls(true) })
            .map { Unit }
    }

billydh

  fun updateCustomer(customer: CustomerPersist): Mono<Unit> {
        return Mono.fromFuture(table.updateItem(customer))
            .map { Unit }
    }

billydh

import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.*

@DynamoDbBean
data class Account(
    @get:DynamoDbPartitionKey var accountId: String? = null,
    @get:DynamoDbSortKey var creationTimestampEpoch: Long? = null,
    @get:DynamoDbSecondaryPartitionKey(indexNames = ["accountByCustomerId"]) var customerId: String? = null,
    @get:DynamoDbSecondarySortKey(indexNames = ["accountByCustomerId"]) var amount: Double? = null,
    var accountType: String? = null
)

billydh

import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbBean
import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbPartitionKey
import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbSortKey

@DynamoDbBean
data class AccountActivity(
    @get:DynamoDbPartitionKey var accountId: String? = null,
    @get:DynamoDbSortKey var eventTimestampEpoch: Long? = null,
    var activityName: String? = null
)

billydh

import org.slf4j.Logger
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Value
import org.springframework.stereotype.Repository
import reactor.core.publisher.Mono
import software.amazon.awssdk.enhanced.dynamodb.DynamoDbEnhancedAsyncClient
import software.amazon.awssdk.enhanced.dynamodb.Key
import software.amazon.awssdk.enhanced.dynamodb.TableSchema

@Repository

billydh

import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider
import software.amazon.awssdk.enhanced.dynamodb.DynamoDbEnhancedAsyncClient
import software.amazon.awssdk.regions.Region
import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient
import java.net.URI

@Configuration
class DynamoClientProperties(private val dynamoConfigProperties: DynamoConfigProperties) {

billydh

import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbBean
import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbPartitionKey

@DynamoDbBean
data class CustomerPersist(
    @get:DynamoDbPartitionKey var customerId: String? = null,
    var emailAddress: String? = null,
    var firstName: String? = null,
    var lastName: String? = null
)

billydh

data class CustomerPersist(val customerId: String, val emailAddress: String, val firstName: String, val lastName: String)