-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
sample: Shopping cart sample from Akka guide (#14)
* feat: CreateTables utility * wrong location of logback-test.xml * sample: Shopping cart sample from Akka guide * remove wrong grpc-api dependency * incompatibility: NoSuchFieldError PickFirstLeafLoadBalancer HEALTH_CONSUMER_LISTENER_ARG_KEY * should be transitive from akka-grpc * remove obsolete fixme * update docker compose in readme * updated snapshot * r2dbc reference Co-authored-by: Peter Vlugter <[email protected]> * snapshot plugin * region and credentials environment variables * kubernetes for analytics-service * atLeastOnce --------- Co-authored-by: Peter Vlugter <[email protected]>
- Loading branch information
Showing
68 changed files
with
2,908 additions
and
265 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
25 changes: 25 additions & 0 deletions
25
core/src/main/scala/akka/persistence/dynamodb/util/javadsl/CreateTables.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,25 @@ | ||
| /* | ||
| * Copyright (C) 2024 Lightbend Inc. <https://www.lightbend.com> | ||
| */ | ||
|
|
||
| package akka.persistence.dynamodb.util.javadsl | ||
|
|
||
| import java.util.concurrent.CompletionStage | ||
|
|
||
| import scala.jdk.FutureConverters._ | ||
|
|
||
| import akka.Done | ||
| import akka.actor.typed.ActorSystem | ||
| import akka.persistence.dynamodb.DynamoDBSettings | ||
| import akka.persistence.dynamodb.util.scaladsl | ||
| import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient | ||
|
|
||
| object CreateTables { | ||
| def createJournalTable( | ||
| system: ActorSystem[_], | ||
| settings: DynamoDBSettings, | ||
| client: DynamoDbAsyncClient, | ||
| deleteIfExists: Boolean): CompletionStage[Done] = | ||
| scaladsl.CreateTables.createJournalTable(system, settings, client, deleteIfExists).asJava | ||
|
|
||
| } |
143 changes: 143 additions & 0 deletions
143
core/src/main/scala/akka/persistence/dynamodb/util/scaladsl/CreateTables.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,143 @@ | ||
| /* | ||
| * Copyright (C) 2024 Lightbend Inc. <https://www.lightbend.com> | ||
| */ | ||
|
|
||
| package akka.persistence.dynamodb.util.scaladsl | ||
|
|
||
| import java.util.concurrent.CompletionException | ||
|
|
||
| import scala.concurrent.ExecutionContext | ||
| import scala.concurrent.Future | ||
| import scala.jdk.FutureConverters._ | ||
| import scala.util.Failure | ||
| import scala.util.Success | ||
|
|
||
| import akka.Done | ||
| import akka.actor.typed.ActorSystem | ||
| import akka.persistence.dynamodb.DynamoDBSettings | ||
| import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient | ||
| import software.amazon.awssdk.services.dynamodb.model.AttributeDefinition | ||
| import software.amazon.awssdk.services.dynamodb.model.CreateTableRequest | ||
| import software.amazon.awssdk.services.dynamodb.model.DeleteTableRequest | ||
| import software.amazon.awssdk.services.dynamodb.model.DescribeTableRequest | ||
| import software.amazon.awssdk.services.dynamodb.model.GlobalSecondaryIndex | ||
| import software.amazon.awssdk.services.dynamodb.model.KeySchemaElement | ||
| import software.amazon.awssdk.services.dynamodb.model.KeyType | ||
| import software.amazon.awssdk.services.dynamodb.model.Projection | ||
| import software.amazon.awssdk.services.dynamodb.model.ProjectionType | ||
| import software.amazon.awssdk.services.dynamodb.model.ProvisionedThroughput | ||
| import software.amazon.awssdk.services.dynamodb.model.ResourceNotFoundException | ||
| import software.amazon.awssdk.services.dynamodb.model.ScalarAttributeType | ||
|
|
||
| object CreateTables { | ||
|
|
||
| def createJournalTable( | ||
| system: ActorSystem[_], | ||
| settings: DynamoDBSettings, | ||
| client: DynamoDbAsyncClient, | ||
| deleteIfExists: Boolean): Future[Done] = { | ||
| import akka.persistence.dynamodb.internal.JournalAttributes._ | ||
| implicit val ec: ExecutionContext = system.executionContext | ||
|
|
||
| val existingTable = | ||
| client.describeTable(DescribeTableRequest.builder().tableName(settings.journalTable).build()).asScala | ||
|
|
||
| def create(): Future[Done] = { | ||
| val sliceIndex = GlobalSecondaryIndex | ||
| .builder() | ||
| .indexName(settings.journalBySliceGsi) | ||
| .keySchema( | ||
| KeySchemaElement.builder().attributeName(EntityTypeSlice).keyType(KeyType.HASH).build(), | ||
| KeySchemaElement.builder().attributeName(Timestamp).keyType(KeyType.RANGE).build()) | ||
| .projection( | ||
| // FIXME we could skip a few attributes | ||
| Projection.builder().projectionType(ProjectionType.ALL).build()) | ||
| // FIXME config | ||
| .provisionedThroughput(ProvisionedThroughput.builder().readCapacityUnits(10L).writeCapacityUnits(10L).build()) | ||
| .build() | ||
|
|
||
| val req = CreateTableRequest | ||
| .builder() | ||
| .tableName(settings.journalTable) | ||
| .keySchema( | ||
| KeySchemaElement.builder().attributeName(Pid).keyType(KeyType.HASH).build(), | ||
| KeySchemaElement.builder().attributeName(SeqNr).keyType(KeyType.RANGE).build()) | ||
| .attributeDefinitions( | ||
| AttributeDefinition.builder().attributeName(Pid).attributeType(ScalarAttributeType.S).build(), | ||
| AttributeDefinition.builder().attributeName(SeqNr).attributeType(ScalarAttributeType.N).build(), | ||
| AttributeDefinition.builder().attributeName(EntityTypeSlice).attributeType(ScalarAttributeType.S).build(), | ||
| AttributeDefinition.builder().attributeName(Timestamp).attributeType(ScalarAttributeType.N).build()) | ||
| // FIXME config | ||
| .provisionedThroughput(ProvisionedThroughput.builder().readCapacityUnits(5L).writeCapacityUnits(5L).build()) | ||
| .globalSecondaryIndexes(sliceIndex) | ||
| .build() | ||
|
|
||
| client.createTable(req).asScala.map(_ => Done) | ||
| } | ||
|
|
||
| def delete(): Future[Done] = { | ||
| val req = DeleteTableRequest.builder().tableName(settings.journalTable).build() | ||
| client.deleteTable(req).asScala.map(_ => Done) | ||
| } | ||
|
|
||
| existingTable.transformWith { | ||
| case Success(_) => | ||
| if (deleteIfExists) delete().flatMap(_ => create()) | ||
| else Future.successful(Done) | ||
| case Failure(_: ResourceNotFoundException) => create() | ||
| case Failure(exception: CompletionException) => | ||
| exception.getCause match { | ||
| case _: ResourceNotFoundException => create() | ||
| case failure => Future.failed[Done](failure) | ||
| } | ||
| case Failure(exc) => | ||
| Future.failed[Done](exc) | ||
| } | ||
| } | ||
|
|
||
| def createSnapshotsTable( | ||
| system: ActorSystem[_], | ||
| settings: DynamoDBSettings, | ||
| client: DynamoDbAsyncClient, | ||
| deleteIfExists: Boolean): Future[Done] = { | ||
| import akka.persistence.dynamodb.internal.SnapshotAttributes._ | ||
|
|
||
| implicit val ec: ExecutionContext = system.executionContext | ||
|
|
||
| val existingTable = | ||
| client.describeTable(DescribeTableRequest.builder().tableName(settings.snapshotTable).build()).asScala | ||
|
|
||
| def create(): Future[Done] = { | ||
| val request = CreateTableRequest | ||
| .builder() | ||
| .tableName(settings.snapshotTable) | ||
| .keySchema(KeySchemaElement.builder().attributeName(Pid).keyType(KeyType.HASH).build()) | ||
| .attributeDefinitions( | ||
| AttributeDefinition.builder().attributeName(Pid).attributeType(ScalarAttributeType.S).build()) | ||
| // FIXME config | ||
| .provisionedThroughput(ProvisionedThroughput.builder().readCapacityUnits(5L).writeCapacityUnits(5L).build()) | ||
| .build() | ||
|
|
||
| client.createTable(request).asScala.map(_ => Done) | ||
| } | ||
|
|
||
| def delete(): Future[Done] = { | ||
| val req = DeleteTableRequest.builder().tableName(settings.snapshotTable).build() | ||
| client.deleteTable(req).asScala.map(_ => Done) | ||
| } | ||
|
|
||
| existingTable.transformWith { | ||
| case Success(_) => | ||
| if (deleteIfExists) delete().flatMap(_ => create()) | ||
| else Future.successful(Done) | ||
| case Failure(_: ResourceNotFoundException) => create() | ||
| case Failure(exception: CompletionException) => | ||
| exception.getCause match { | ||
| case _: ResourceNotFoundException => create() | ||
| case failure => Future.failed[Done](failure) | ||
| } | ||
| case Failure(exception) => Future.failed[Done](exception) | ||
| } | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.