マイクロサービスでイベントソーシングを利用したくなる理由とその動作原理を学ぼう / Why Event Sourcing

イベントソーシングがなぜ必要になるのかの解説と、その動作原理を解説する際に利用した資料です。
この資料を使ったトークが YouTube にアップロードされています。
トークURL: https://youtu.be/Jtcp9ry8ZcE

◆ URL
トークURL: https://youtu.be/Jtcp9ry8ZcE
チャンネル登録: https://www.youtube.com/c/narusemi?sub_confirmation=1
Twitter: https://twitter.com/nrslib
SampleCode: https://github.com/nrslib/microservice-with-event-sourcing-sample-kotlin

Transcript

24. @Transactional public void order(String itemId) { var itemDao = new ItemDao(); var item = itemDao.findById(itemId); ... var accountDao = new AccountDao(); var account = accountDao.find(Session.Id); var paymentDao = new PaymentDao(); var paymentInfo = paymentDao.find(account.id); paymentInfo.validate(); creditService.approval(paymentInfo.cardInfo, order.total); ... orderDao.save(order); }
25. @Transactional public void order(String itemId) { var itemDao = new ItemDao(); var item = itemDao.findById(itemId); ... var accountDao = new AccountDao(); var account = accountDao.find(Session.Id); var paymentDao = new PaymentDao(); var paymentInfo = paymentDao.find(account.id); paymentInfo.validate(); creditService.approval(paymentInfo.cardInfo, order.total); ... orderDao.save(order); }
26. @Transactional public void order(String itemId) { var itemDao = new ItemDao(); var item = itemDao.findById(itemId); ... var accountDao = new AccountDao(); var account = accountDao.find(Session.Id); var paymentDao = new PaymentDao(); var paymentInfo = paymentDao.find(account.id); paymentInfo.validate(); creditService.approval(paymentInfo.cardInfo, order.total); ... orderDao.save(order); }
166. class Order : CommandProcessingAggregate<Order, OrderCommand>() { lateinit var accountId: String var total: Money = Money.zero() lateinit var items: OrderItems lateinit var orderState: OrderState ... }
167. class OrderApplicationService( private val itemRepository: ItemRepository, private val orderRepository: AggregateRepository<Order, OrderCommand> ) { @Transactional fun createOrder(accountId: String, itemAndQuantities: List<ItemIdAndQuantity>): String { val items = itemRepository.find(itemAndQuantities.map { ItemId(it.itemId) }) val orderItems = makeOrderItems(itemAndQuantities, items) val command = CreateOrderCommand(UUID.randomUUID().toString(), accountId, orderItems) val result = orderRepository.save(command) return result.idAndVersion.id } ... private fun makeOrderItems(itemAndQuantities: List<ItemIdAndQuantity>, items: List<Item>): OrderItems { val orderItems = itemAndQuantities.map { val item = items.first { item -> item.id.value == it.itemId } OrderItem( it.quantity, item.price ) } return OrderItems(orderItems) } }
168. class OrderApplicationService( private val itemRepository: ItemRepository, private val orderRepository: AggregateRepository<Order, OrderCommand> ) { @Transactional fun createOrder(accountId: String, itemAndQuantities: List<ItemIdAndQuantity>): String { val items = itemRepository.find(itemAndQuantities.map { ItemId(it.itemId) }) val orderItems = makeOrderItems(itemAndQuantities, items) val command = CreateOrderCommand(UUID.randomUUID().toString(), accountId, orderItems) val result = orderRepository.save(command) return result.idAndVersion.id } ... private fun makeOrderItems(itemAndQuantities: List<ItemIdAndQuantity>, items: List<Item>): OrderItems { val orderItems = itemAndQuantities.map { val item = items.first { item -> item.id.value == it.itemId } OrderItem( it.quantity, item.price ) } return OrderItems(orderItems) } }
169. class OrderApplicationService( private val itemRepository: ItemRepository, private val orderRepository: AggregateRepository<Order, OrderCommand> ) { @Transactional fun createOrder(accountId: String, itemAndQuantities: List<ItemIdAndQuantity>): String { val items = itemRepository.find(itemAndQuantities.map { ItemId(it.itemId) }) val orderItems = makeOrderItems(itemAndQuantities, items) val command = CreateOrderCommand(UUID.randomUUID().toString(), accountId, orderItems) val result = orderRepository.save(command) return result.idAndVersion.id } ... private fun makeOrderItems(itemAndQuantities: List<ItemIdAndQuantity>, items: List<Item>): OrderItems { val orderItems = itemAndQuantities.map { val item = items.first { item -> item.id.value == it.itemId } OrderItem( it.quantity, item.price ) } return OrderItems(orderItems) } }
170. class OrderApplicationService( private val itemRepository: ItemRepository, private val orderRepository: AggregateRepository<Order, OrderCommand> ) { @Transactional fun createOrder(accountId: String, itemAndQuantities: List<ItemIdAndQuantity>): String { val items = itemRepository.find(itemAndQuantities.map { ItemId(it.itemId) }) val orderItems = makeOrderItems(itemAndQuantities, items) val command = CreateOrderCommand(UUID.randomUUID().toString(), accountId, orderItems) val result = orderRepository.save(command) return result.idAndVersion.id } ... private fun makeOrderItems(itemAndQuantities: List<ItemIdAndQuantity>, items: List<Item>): OrderItems { val orderItems = itemAndQuantities.map { val item = items.first { item -> item.id.value == it.itemId } OrderItem( it.quantity, item.price ) } return OrderItems(orderItems) } }
171. class OrderApplicationService( private val itemRepository: ItemRepository, private val orderRepository: AggregateRepository<Order, OrderCommand> ) { @Transactional fun createOrder(accountId: String, itemAndQuantities: List<ItemIdAndQuantity>): String { val items = itemRepository.find(itemAndQuantities.map { ItemId(it.itemId) }) val orderItems = makeOrderItems(itemAndQuantities, items) val command = CreateOrderCommand(UUID.randomUUID().toString(), accountId, orderItems) val result = orderRepository.save(command) return result.idAndVersion.id } ... private fun makeOrderItems(itemAndQuantities: List<ItemIdAndQuantity>, items: List<Item>): OrderItems { val orderItems = itemAndQuantities.map { val item = items.first { item -> item.id.value == it.itemId } OrderItem( it.quantity, item.price ) } return OrderItems(orderItems) } }
172. class OrderApplicationService( private val itemRepository: ItemRepository, private val orderRepository: AggregateRepository<Order, OrderCommand> ) { @Transactional fun createOrder(accountId: String, itemAndQuantities: List<ItemIdAndQuantity>): String { val items = itemRepository.find(itemAndQuantities.map { ItemId(it.itemId) }) val orderItems = makeOrderItems(itemAndQuantities, items) val command = CreateOrderCommand(UUID.randomUUID().toString(), accountId, orderItems) val result = orderRepository.save(command) return result.idAndVersion.id } ... private fun makeOrderItems(itemAndQuantities: List<ItemIdAndQuantity>, items: List<Item>): OrderItems { val orderItems = itemAndQuantities.map { val item = items.first { item -> item.id.value == it.itemId } OrderItem( it.quantity, item.price ) } return OrderItems(orderItems) } }
173. class OrderApplicationService( private val itemRepository: ItemRepository, private val orderRepository: AggregateRepository<Order, OrderCommand> ) { @Transactional fun createOrder(accountId: String, itemAndQuantities: List<ItemIdAndQuantity>): String { val items = itemRepository.find(itemAndQuantities.map { ItemId(it.itemId) }) val orderItems = makeOrderItems(itemAndQuantities, items) val command = CreateOrderCommand(UUID.randomUUID().toString(), accountId, orderItems) val result = orderRepository.save(command) return result.idAndVersion.id } ... private fun makeOrderItems(itemAndQuantities: List<ItemIdAndQuantity>, items: List<Item>): OrderItems { val orderItems = itemAndQuantities.map { val item = items.first { item -> item.id.value == it.itemId } OrderItem( it.quantity, item.price ) } return OrderItems(orderItems) } }
174. class Order : CommandProcessingAggregate<Order, OrderCommand>() { lateinit var accountId: String var total: Money = Money.zero() lateinit var items: OrderItems lateinit var orderState: OrderState fun process(command: CreateOrderCommand): List<Event> { val total = command.orderItems.total() val event = OrderCreatedEvent(command.id, command.accountId, total, command.orderItems) return listOf(event) } fun apply(event: OrderCreatedEvent) { accountId = event.accountId total = event.total items = event.orderItems orderState = OrderState.APPROVAL_PENDING } }
175. class Order : CommandProcessingAggregate<Order, OrderCommand>() { lateinit var accountId: String var total: Money = Money.zero() lateinit var items: OrderItems lateinit var orderState: OrderState fun process(command: CreateOrderCommand): List<Event> { val total = command.orderItems.total() val event = OrderCreatedEvent(command.id, command.accountId, total, command.orderItems) return listOf(event) } fun apply(event: OrderCreatedEvent) { accountId = event.accountId total = event.total items = event.orderItems orderState = OrderState.APPROVAL_PENDING } }
176. class Order : CommandProcessingAggregate<Order, OrderCommand>() { lateinit var accountId: String var total: Money = Money.zero() lateinit var items: OrderItems lateinit var orderState: OrderState fun process(command: CreateOrderCommand): List<Event> { val total = command.orderItems.total() val event = OrderCreatedEvent(command.id, command.accountId, total, command.orderItems) return listOf(event) } fun apply(event: OrderCreatedEvent) { accountId = event.accountId total = event.total items = event.orderItems orderState = OrderState.APPROVAL_PENDING } }
177. class Order : CommandProcessingAggregate<Order, OrderCommand>() { lateinit var accountId: String var total: Money = Money.zero() lateinit var items: OrderItems lateinit var orderState: OrderState fun process(command: CreateOrderCommand): List<Event> { val total = command.orderItems.total() val event = OrderCreatedEvent(command.id, command.accountId, total, command.orderItems) return listOf(event) } fun apply(event: OrderCreatedEvent) { accountId = event.accountId total = event.total items = event.orderItems orderState = OrderState.APPROVAL_PENDING } }
178. class Order : CommandProcessingAggregate<Order, OrderCommand>() { lateinit var accountId: String var total: Money = Money.zero() lateinit var items: OrderItems lateinit var orderState: OrderState fun process(command: CreateOrderCommand): List<Event> { val total = command.orderItems.total() val event = OrderCreatedEvent(command.id, command.accountId, total, command.orderItems) return listOf(event) } fun apply(event: OrderCreatedEvent) { accountId = event.accountId total = event.total items = event.orderItems orderState = OrderState.APPROVAL_PENDING } }
179. class Order : CommandProcessingAggregate<Order, OrderCommand>() { lateinit var accountId: String var total: Money = Money.zero() lateinit var items: OrderItems lateinit var orderState: OrderState fun process(command: CreateOrderCommand): List<Event> { val total = command.orderItems.total() val event = OrderCreatedEvent(command.id, command.accountId, total, command.orderItems) return listOf(event) } fun apply(event: OrderCreatedEvent) { accountId = event.accountId total = event.total items = event.orderItems orderState = OrderState.APPROVAL_PENDING } }
185. class Order : CommandProcessingAggregate<Order, OrderCommand>() { lateinit var accountId: String var total: Money = Money.zero() lateinit var items: OrderItems lateinit var orderState: OrderState fun process(command: CreateOrderCommand): List<Event> { val total = command.orderItems.total() val event = OrderCreatedEvent(command.id, command.accountId, total, command.orderItems) return listOf(event) } ... }
189. class Order : CommandProcessingAggregate<Order, OrderCommand>() { lateinit var accountId: String var total: Money = Money.zero() lateinit var items: OrderItems lateinit var orderState: OrderState ... fun apply(event: OrderCreatedEvent) { accountId = event.accountId total = event.total items = event.orderItems orderState = OrderState.APPROVAL_PENDING } ... }
191. class AggregateRepository<T, CT>( private val clazz: Class<T>, private val aggregateStore: AggregateStore ) where T : CommandProcessingAggregate<T, CT>, CT : Command { fun save(command: CT): EntityWithIdAndVersion<T> { val aggregate = clazz.getDeclaredConstructor().newInstance() val events = aggregate.processCommand(command) Aggregate.applyEvents(aggregate, events) val serializedEvents = events.map { EventTypeAndData(it.javaClass.name, Gson().toJson(it)) } val entityIdVersionAndEventIds = aggregateStore.save(clazz, serializedEvents) return EntityWithIdAndVersion( entityIdVersionAndEventIds, aggregate ) } ... }
193. class AggregateRepository<T, CT>( private val clazz: Class<T>, private val aggregateStore: AggregateStore ) where T : CommandProcessingAggregate<T, CT>, CT : Command { fun save(command: CT): EntityWithIdAndVersion<T> { val aggregate = clazz.getDeclaredConstructor().newInstance() val events = aggregate.processCommand(command) Aggregate.applyEvents(aggregate, events) val serializedEvents = events.map { EventTypeAndData(it.javaClass.name, Gson().toJson(it)) } val entityIdVersionAndEventIds = aggregateStore.save(clazz, serializedEvents) return EntityWithIdAndVersion( entityIdVersionAndEventIds, aggregate ) } ... }
195. class AggregateRepository<T, CT>( private val clazz: Class<T>, private val aggregateStore: AggregateStore ) where T : CommandProcessingAggregate<T, CT>, CT : Command { fun save(command: CT): EntityWithIdAndVersion<T> { val aggregate = clazz.getDeclaredConstructor().newInstance() val events = aggregate.processCommand(command) Aggregate.applyEvents(aggregate, events) val serializedEvents = events.map { EventTypeAndData(it.javaClass.name, Gson().toJson(it)) } val entityIdVersionAndEventIds = aggregateStore.save(clazz, serializedEvents) return EntityWithIdAndVersion( entityIdVersionAndEventIds, aggregate ) } ... }
197. class AggregateRepository<T, CT>( private val clazz: Class<T>, private val aggregateStore: AggregateStore ) where T : CommandProcessingAggregate<T, CT>, CT : Command { fun save(command: CT): EntityWithIdAndVersion<T> { val aggregate = clazz.getDeclaredConstructor().newInstance() val events = aggregate.processCommand(command) Aggregate.applyEvents(aggregate, events) val serializedEvents = events.map { EventTypeAndData(it.javaClass.name, Gson().toJson(it)) } val entityIdVersionAndEventIds = aggregateStore.save(clazz, serializedEvents) return EntityWithIdAndVersion( entityIdVersionAndEventIds, aggregate ) } ... }
199. class AggregateRepository<T, CT>( private val clazz: Class<T>, private val aggregateStore: AggregateStore ) where T : CommandProcessingAggregate<T, CT>, CT : Command { fun save(command: CT): EntityWithIdAndVersion<T> { val aggregate = clazz.getDeclaredConstructor().newInstance() val events = aggregate.processCommand(command) Aggregate.applyEvents(aggregate, events) val serializedEvents = events.map { EventTypeAndData(it.javaClass.name, Gson().toJson(it)) } val entityIdVersionAndEventIds = aggregateStore.save(clazz, serializedEvents) return EntityWithIdAndVersion( entityIdVersionAndEventIds, aggregate ) } ... }
200. class AggregateRepository<T, CT>( private val clazz: Class<T>, private val aggregateStore: AggregateStore ) where T : CommandProcessingAggregate<T, CT>, CT : Command { fun save(command: CT): EntityWithIdAndVersion<T> { val aggregate = clazz.getDeclaredConstructor().newInstance() val events = aggregate.processCommand(command) Aggregate.applyEvents(aggregate, events) val serializedEvents = events.map { EventTypeAndData(it.javaClass.name, Gson().toJson(it)) } val entityIdVersionAndEventIds = aggregateStore.save(clazz, serializedEvents) return EntityWithIdAndVersion( entityIdVersionAndEventIds, aggregate ) } ... }
206. class Order : CommandProcessingAggregate<Order, OrderCommand>() { lateinit var accountId: String var total: Money = Money.zero() lateinit var items: OrderItems lateinit var orderState: OrderState ... fun apply(event: OrderCreatedEvent) { accountId = event.accountId total = event.total items = event.orderItems orderState = OrderState.APPROVAL_PENDING } fun apply(event: OrderApprovedEvent) { this.orderState = OrderState.APPROVED } fun apply(event: OrderRejectedEvent) { this.orderState = OrderState.APPROVAL_REJECTED } }
207. class Order : CommandProcessingAggregate<Order, OrderCommand>() { lateinit var accountId: String var total: Money = Money.zero() lateinit var items: OrderItems lateinit var orderState: OrderState ... fun apply(event: OrderCreatedEvent) { accountId = event.accountId total = event.total items = event.orderItems orderState = OrderState.APPROVAL_PENDING } fun apply(event: OrderApprovedEvent) { this.orderState = OrderState.APPROVED } fun apply(event: OrderRejectedEvent) { this.orderState = OrderState.APPROVAL_REJECTED } }
210. class Order : CommandProcessingAggregate<Order, OrderCommand>() { lateinit var accountId: String var total: Money = Money.zero() lateinit var items: OrderItems lateinit var orderState: OrderState ... fun process(command: ApproveOrderCommand): List<Event> { return listOf(OrderApprovedEvent()) } }
213. class AggregateRepository<T, CT>( private val clazz: Class<T>, private val aggregateStore: AggregateStore ) where T : CommandProcessingAggregate<T, CT>, CT : Command { ... fun update(id: String, cmd: CT, updateOption: UpdateOptions = UpdateOptions()) : EntityWithIdAndVersion<T> { val loadedData = aggregateStore.find(clazz, id) ?: throw IllegalArgumentException("$id is notfound") val aggregate = loadedData.entity val events = aggregate.processCommand(cmd) Aggregate.applyEvents(aggregate, events) val idAndVersion = aggregateStore.update( clazz, loadedData.idAndVersion, events, withPossibleSnapshot( updateOption, aggregate, loadedData.snapshotVersion, loadedData.events, events) ) return EntityWithIdAndVersion(idAndVersion, aggregate) } }
215. class AggregateRepository<T, CT>( private val clazz: Class<T>, private val aggregateStore: AggregateStore ) where T : CommandProcessingAggregate<T, CT>, CT : Command { ... fun update(id: String, cmd: CT, updateOption: UpdateOptions = UpdateOptions()) : EntityWithIdAndVersion<T> { val loadedData = aggregateStore.find(clazz, id) ?: throw IllegalArgumentException("$id is notfound") val aggregate = loadedData.entity val events = aggregate.processCommand(cmd) Aggregate.applyEvents(aggregate, events) val idAndVersion = aggregateStore.update( clazz, loadedData.idAndVersion, events, withPossibleSnapshot( updateOption, aggregate, loadedData.snapshotVersion, loadedData.events, events) ) return EntityWithIdAndVersion(idAndVersion, aggregate) } }
217. class AggregateRepository<T, CT>( private val clazz: Class<T>, private val aggregateStore: AggregateStore ) where T : CommandProcessingAggregate<T, CT>, CT : Command { ... fun update(id: String, cmd: CT, updateOption: UpdateOptions = UpdateOptions()) : EntityWithIdAndVersion<T> { val loadedData = aggregateStore.find(clazz, id) ?: throw IllegalArgumentException("$id is notfound") val aggregate = loadedData.entity val events = aggregate.processCommand(cmd) Aggregate.applyEvents(aggregate, events) val idAndVersion = aggregateStore.update( clazz, loadedData.idAndVersion, events, withPossibleSnapshot( updateOption, aggregate, loadedData.snapshotVersion, loadedData.events, events) ) return EntityWithIdAndVersion(idAndVersion, aggregate) } }
219. class AggregateRepository<T, CT>( private val clazz: Class<T>, private val aggregateStore: AggregateStore ) where T : CommandProcessingAggregate<T, CT>, CT : Command { ... fun update(id: String, cmd: CT, updateOption: UpdateOptions = UpdateOptions()) : EntityWithIdAndVersion<T> { val loadedData = aggregateStore.find(clazz, id) ?: throw IllegalArgumentException("$id is notfound") val aggregate = loadedData.entity val events = aggregate.processCommand(cmd) Aggregate.applyEvents(aggregate, events) val idAndVersion = aggregateStore.update( clazz, loadedData.idAndVersion, events, withPossibleSnapshot( updateOption, aggregate, loadedData.snapshotVersion, loadedData.events, events) ) return EntityWithIdAndVersion(idAndVersion, aggregate) } }
221. class AggregateRepository<T, CT>( private val clazz: Class<T>, private val aggregateStore: AggregateStore ) where T : CommandProcessingAggregate<T, CT>, CT : Command { ... fun update(id: String, cmd: CT, updateOption: UpdateOptions = UpdateOptions()) : EntityWithIdAndVersion<T> { val loadedData = aggregateStore.find(clazz, id) ?: throw IllegalArgumentException("$id is notfound") val aggregate = loadedData.entity val events = aggregate.processCommand(cmd) Aggregate.applyEvents(aggregate, events) val idAndVersion = aggregateStore.update( clazz, loadedData.idAndVersion, events, withPossibleSnapshot( updateOption, aggregate, loadedData.snapshotVersion, loadedData.events, events) ) return EntityWithIdAndVersion(idAndVersion, aggregate) } }
223. class AggregateRepository<T, CT>( private val clazz: Class<T>, private val aggregateStore: AggregateStore ) where T : CommandProcessingAggregate<T, CT>, CT : Command { ... fun update(id: String, cmd: CT, updateOption: UpdateOptions = UpdateOptions()) : EntityWithIdAndVersion<T> { val loadedData = aggregateStore.find(clazz, id) ?: throw IllegalArgumentException("$id is notfound") val aggregate = loadedData.entity val events = aggregate.processCommand(cmd) Aggregate.applyEvents(aggregate, events) val idAndVersion = aggregateStore.update( clazz, loadedData.idAndVersion, events, withPossibleSnapshot( updateOption, aggregate, loadedData.snapshotVersion, loadedData.events, events) ) return EntityWithIdAndVersion(idAndVersion, aggregate) } }
224. class AggregateRepository<T, CT>( private val clazz: Class<T>, private val aggregateStore: AggregateStore ) where T : CommandProcessingAggregate<T, CT>, CT : Command { ... fun update(id: String, cmd: CT, updateOption: UpdateOptions = UpdateOptions()) : EntityWithIdAndVersion<T> { val loadedData = aggregateStore.find(clazz, id) ?: throw IllegalArgumentException("$id is notfound") val aggregate = loadedData.entity val events = aggregate.processCommand(cmd) Aggregate.applyEvents(aggregate, events) val idAndVersion = aggregateStore.update( clazz, loadedData.idAndVersion, events, withPossibleSnapshot( updateOption, aggregate, loadedData.snapshotVersion, loadedData.events, events) ) return EntityWithIdAndVersion(idAndVersion, aggregate) } }
226. enum class DocumentType { Entity, Event, Snapshot } class EventSourcingDocument { @Id lateinit var id: String lateinit var type: DocumentType lateinit var entityId: String lateinit var body: String var version: Long = -1 @JsonProperty("_etag") var etag: String = "" }
227. enum class DocumentType { Entity, Event, Snapshot } class EventSourcingDocument { @Id lateinit var id: String lateinit var type: DocumentType lateinit var entityId: String lateinit var body: String var version: Long = -1 @JsonProperty("_etag") var etag: String = "" }
228. enum class DocumentType { Entity, Event, Snapshot } class EventSourcingDocument { @Id lateinit var id: String lateinit var type: DocumentType lateinit var entityId: String lateinit var body: String var version: Long = -1 @JsonProperty("_etag") var etag: String = "" }
229. class EventDataModel { lateinit var id: String lateinit var eventType: String lateinit var entityType: String lateinit var entityId: String lateinit var payload: String var version: Long = -1 ... }
230. class EntityDataModel { lateinit var id: String lateinit var type: String var version: Long = -1 ... }
231. class SnapshotDataModel { lateinit var id: String lateinit var entityType: String lateinit var entityId: String var entityVersion: Long = -1 lateinit var snapshotType: String lateinit var snapshotData: String ... }
233. class CosmosAggregateStore(...) : AggregateStore { override fun <T : Aggregate<T>> save(clazz: Class<T>, events: List<EventTypeAndData>, options: AggregateCrudSaveOptions val container = getContainer() val entityId = UUID.randomUUID().toString() val version = events.count().toLong() val batch = TransactionalBatch.createTransactionalBatch(PartitionKey(entityId)) val entityDataModel = EntityDataModel(entityId, clazz.name, version) val entityDocument = entityDataModel.toDocument() batch.createItemOperation(entityDocument) events.mapIndexed { index, it -> EventDataModel( UUID.randomUUID().toString(), it.eventType, clazz.name, entityId, index.toLong() + 1, it.eventData ) }.map(EventDataModel::toDocument) .forEach { batch.createItemOperation(it) } val response = container.executeTransactionalBatch(batch) if (!response.isSuccessStatusCode) { throw UnsupportedOperationException() } return EntityIdAndVersion(entityId, version) }
234. class CosmosAggregateStore(...) : AggregateStore { override fun <T : Aggregate<T>> save(clazz: Class<T>, events: List<EventTypeAndData>, options: AggregateCrudSaveOptions? val container = getContainer() val entityId = UUID.randomUUID().toString() val version = events.count().toLong() val batch = TransactionalBatch.createTransactionalBatch(PartitionKey(entityId)) val entityDataModel = EntityDataModel(entityId, clazz.name, version) val entityDocument = entityDataModel.toDocument() batch.createItemOperation(entityDocument) events.mapIndexed { index, it -> EventDataModel( UUID.randomUUID().toString(), it.eventType, clazz.name, entityId, index.toLong() + 1, it.eventData ) }.map(EventDataModel::toDocument) .forEach { batch.createItemOperation(it) } val response = container.executeTransactionalBatch(batch) if (!response.isSuccessStatusCode) { throw UnsupportedOperationException() } return EntityIdAndVersion(entityId, version) }
235. class CosmosAggregateStore(...) : AggregateStore { override fun <T : Aggregate<T>> save(clazz: Class<T>, events: List<EventTypeAndData>, options: AggregateCrudSaveOptions? val container = getContainer() val entityId = UUID.randomUUID().toString() val version = events.count().toLong() val batch = TransactionalBatch.createTransactionalBatch(PartitionKey(entityId)) val entityDataModel = EntityDataModel(entityId, clazz.name, version) val entityDocument = entityDataModel.toDocument() batch.createItemOperation(entityDocument) events.mapIndexed { index, it -> EventDataModel( UUID.randomUUID().toString(), it.eventType, clazz.name, entityId, index.toLong() + 1, it.eventData ) }.map(EventDataModel::toDocument) .forEach { batch.createItemOperation(it) } val response = container.executeTransactionalBatch(batch) if (!response.isSuccessStatusCode) { throw UnsupportedOperationException() } return EntityIdAndVersion(entityId, version) }
236. class CosmosAggregateStore(...) : AggregateStore { override fun <T : Aggregate<T>> save(clazz: Class<T>, events: List<EventTypeAndData>, options: AggregateCrudSaveOptions? val container = getContainer() val entityId = UUID.randomUUID().toString() val version = events.count().toLong() val batch = TransactionalBatch.createTransactionalBatch(PartitionKey(entityId)) val entityDataModel = EntityDataModel(entityId, clazz.name, version) val entityDocument = entityDataModel.toDocument() batch.createItemOperation(entityDocument) events.mapIndexed { index, it -> EventDataModel( UUID.randomUUID().toString(), it.eventType, clazz.name, entityId, index.toLong() + 1, it.eventData ) }.map(EventDataModel::toDocument) .forEach { batch.createItemOperation(it) } val response = container.executeTransactionalBatch(batch) if (!response.isSuccessStatusCode) { throw UnsupportedOperationException() } return EntityIdAndVersion(entityId, version) }
237. class CosmosAggregateStore(...) : AggregateStore { override fun <T : Aggregate<T>> save(clazz: Class<T>, events: List<EventTypeAndData>, options: AggregateCrudSaveOptions? val container = getContainer() val entityId = UUID.randomUUID().toString() val version = events.count().toLong() val batch = TransactionalBatch.createTransactionalBatch(PartitionKey(entityId)) val entityDataModel = EntityDataModel(entityId, clazz.name, version) val entityDocument = entityDataModel.toDocument() batch.createItemOperation(entityDocument) events.mapIndexed { index, it -> EventDataModel( UUID.randomUUID().toString(), it.eventType, clazz.name, entityId, index.toLong() + 1, it.eventData ) }.map(EventDataModel::toDocument) .forEach { batch.createItemOperation(it) } val response = container.executeTransactionalBatch(batch) if (!response.isSuccessStatusCode) { throw UnsupportedOperationException() } return EntityIdAndVersion(entityId, version) }
238. @Beta(value = Beta.SinceVersion.V4_7_0, warningText = Beta.PREVIEW_SUBJECT_TO_CHANGE_WARNING) @Deprecated() //forRemoval = true, since = "4.19" public final class TransactionalBatch {
239. @Beta(value = Beta.SinceVersion.V4_7_0, warningText = Beta.PREVIEW_SUBJECT_TO_CHANGE_WARNING) @Deprecated() //forRemoval = true, since = "4.19" public final class TransactionalBatch {
240. class CosmosAggregateStore(...) : AggregateStore { override fun <T : Aggregate<T>> save(clazz: Class<T>, events: List<EventTypeAndData>, options: AggregateCrudSaveOptions? val container = getContainer() val entityId = UUID.randomUUID().toString() val version = events.count().toLong() val batch = TransactionalBatch.createTransactionalBatch(PartitionKey(entityId)) val entityDataModel = EntityDataModel(entityId, clazz.name, version) val entityDocument = entityDataModel.toDocument() batch.createItemOperation(entityDocument) events.mapIndexed { index, it -> EventDataModel( UUID.randomUUID().toString(), it.eventType, clazz.name, entityId, index.toLong() + 1, it.eventData ) }.map(EventDataModel::toDocument) .forEach { batch.createItemOperation(it) } val response = container.executeTransactionalBatch(batch) if (!response.isSuccessStatusCode) { throw UnsupportedOperationException() } return EntityIdAndVersion(entityId, version) }
241. class CosmosAggregateStore(...) : AggregateStore { override fun <T : Aggregate<T>> save(clazz: Class<T>, events: List<EventTypeAndData>, options: AggregateCrudSaveOptions? val container = getContainer() val entityId = UUID.randomUUID().toString() val version = events.count().toLong() val batch = TransactionalBatch.createTransactionalBatch(PartitionKey(entityId)) val entityDataModel = EntityDataModel(entityId, clazz.name, version) val entityDocument = entityDataModel.toDocument() batch.createItemOperation(entityDocument) events.mapIndexed { index, it -> EventDataModel( UUID.randomUUID().toString(), it.eventType, clazz.name, entityId, index.toLong() + 1, it.eventData ) }.map(EventDataModel::toDocument) .forEach { batch.createItemOperation(it) } val response = container.executeTransactionalBatch(batch) if (!response.isSuccessStatusCode) { throw UnsupportedOperationException() } return EntityIdAndVersion(entityId, version) }
242. class CosmosAggregateStore(...) : AggregateStore { override fun <T : Aggregate<T>> save(clazz: Class<T>, events: List<EventTypeAndData>, options: AggregateCrudSaveOptions? val container = getContainer() val entityId = UUID.randomUUID().toString() val version = events.count().toLong() val batch = TransactionalBatch.createTransactionalBatch(PartitionKey(entityId)) val entityDataModel = EntityDataModel(entityId, clazz.name, version) val entityDocument = entityDataModel.toDocument() batch.createItemOperation(entityDocument) events.mapIndexed { index, it -> EventDataModel( UUID.randomUUID().toString(), it.eventType, clazz.name, entityId, index.toLong() + 1, it.eventData ) }.map(EventDataModel::toDocument) .forEach { batch.createItemOperation(it) } val response = container.executeTransactionalBatch(batch) if (!response.isSuccessStatusCode) { throw UnsupportedOperationException() } return EntityIdAndVersion(entityId, version) }
244. } val eventDatamodels = events.mapIndexed { index, it -> EventDataModel( UUID.randomUUID().toString(), it.javaClass.name, clazz.name, entityId, currentVersion + index + 1, Gson().toJson(it) ) } eventDatamodels .map(EventDataModel::toDocument) .forEach { batch.createItemOperation(it) } val latestVersion = eventDatamodels.last().version val entity = EntityDataModel.from(entityDocument) entity.version = latestVersion val replaceEntityDocument = entity.toDocument() batch.replaceItemOperation(entityId, replaceEntityDocument, TransactionalBatchItemRequestOptions().setIfMatchETag(entit if (options?.snapshot != null) { val snapshot = options.snapshot val snapshotDataModel = SnapshotDataModel( UUID.randomUUID().toString(), clazz.name, entityIdAndVersion.id, latestVersion, snapshot!!.javaClass.name,
246. override fun <T : Aggregate<T>> update(clazz: Class<T>, entityIdAndVersion: EntityIdAndVersion, events: List<Event>, option val entityId = entityIdAndVersion.id val currentVersion = entityIdAndVersion.version val batch = TransactionalBatch.createTransactionalBatch(PartitionKey(entityId)) val container = getContainer() val entityDocumentReadResponse = container.readItem(entityId, PartitionKey(entityId), EventSourcingDocument::class.java val entityDocument = entityDocumentReadResponse.item if (currentVersion != entityDocument.version) { throw OptimisticConcurrencyException(entityId, currentVersion, entityDocument.version) } val eventDatamodels = events.mapIndexed { index, it -> EventDataModel( UUID.randomUUID().toString(), it.javaClass.name, clazz.name, entityId, currentVersion + index + 1, Gson().toJson(it) ) } eventDatamodels.map(EventDataModel::toDocument).forEach {batch.createItemOperation(it)} val latestVersion = eventDatamodels.last().version val entity = EntityDataModel.from(entityDocument) entity.version = latestVersion val replaceEntityDocument = entity.toDocument() batch.replaceItemOperation(entityId, replaceEntityDocument, TransactionalBatchItemRequestOptions().setIfMatchETag(entit
247. override fun <T : Aggregate<T>> update(clazz: Class<T>, entityIdAndVersion: EntityIdAndVersion, events: List<Event>, option val entityId = entityIdAndVersion.id val currentVersion = entityIdAndVersion.version val batch = TransactionalBatch.createTransactionalBatch(PartitionKey(entityId)) val container = getContainer() val entityDocumentReadResponse = container.readItem(entityId, PartitionKey(entityId), EventSourcingDocument::class.java val entityDocument = entityDocumentReadResponse.item if (currentVersion != entityDocument.version) { throw OptimisticConcurrencyException(entityId, currentVersion, entityDocument.version) } val eventDatamodels = events.mapIndexed { index, it -> EventDataModel( UUID.randomUUID().toString(), it.javaClass.name, clazz.name, entityId, currentVersion + index + 1, Gson().toJson(it) ) } eventDatamodels.map(EventDataModel::toDocument).forEach {batch.createItemOperation(it)} val latestVersion = eventDatamodels.last().version val entity = EntityDataModel.from(entityDocument) entity.version = latestVersion val replaceEntityDocument = entity.toDocument() batch.replaceItemOperation(entityId, replaceEntityDocument, TransactionalBatchItemRequestOptions().setIfMatchETag(entit
248. override fun <T : Aggregate<T>> update(clazz: Class<T>, entityIdAndVersion: EntityIdAndVersion, events: List<Event>, option val entityId = entityIdAndVersion.id val currentVersion = entityIdAndVersion.version val batch = TransactionalBatch.createTransactionalBatch(PartitionKey(entityId)) val container = getContainer() val entityDocumentReadResponse = container.readItem(entityId, PartitionKey(entityId), EventSourcingDocument::class.java val entityDocument = entityDocumentReadResponse.item if (currentVersion != entityDocument.version) { throw OptimisticConcurrencyException(entityId, currentVersion, entityDocument.version) } val eventDatamodels = events.mapIndexed { index, it -> EventDataModel( UUID.randomUUID().toString(), it.javaClass.name, clazz.name, entityId, currentVersion + index + 1, Gson().toJson(it) ) } eventDatamodels.map(EventDataModel::toDocument).forEach {batch.createItemOperation(it)} val latestVersion = eventDatamodels.last().version val entity = EntityDataModel.from(entityDocument) entity.version = latestVersion val replaceEntityDocument = entity.toDocument() batch.replaceItemOperation(entityId, replaceEntityDocument, TransactionalBatchItemRequestOptions().setIfMatchETag(entit
249. override fun <T : Aggregate<T>> update(clazz: Class<T>, entityIdAndVersion: EntityIdAndVersion, events: List<Event>, option val entityId = entityIdAndVersion.id val currentVersion = entityIdAndVersion.version val batch = TransactionalBatch.createTransactionalBatch(PartitionKey(entityId)) val container = getContainer() val entityDocumentReadResponse = container.readItem(entityId, PartitionKey(entityId), EventSourcingDocument::class.java val entityDocument = entityDocumentReadResponse.item if (currentVersion != entityDocument.version) { throw OptimisticConcurrencyException(entityId, currentVersion, entityDocument.version) } val eventDatamodels = events.mapIndexed { index, it -> EventDataModel( UUID.randomUUID().toString(), it.javaClass.name, clazz.name, entityId, currentVersion + index + 1, Gson().toJson(it) ) } eventDatamodels.map(EventDataModel::toDocument).forEach {batch.createItemOperation(it)} val latestVersion = eventDatamodels.last().version val entity = EntityDataModel.from(entityDocument) entity.version = latestVersion val replaceEntityDocument = entity.toDocument() batch.replaceItemOperation(entityId, replaceEntityDocument, TransactionalBatchItemRequestOptions().setIfMatchETag(entit
250. override fun <T : Aggregate<T>> update(clazz: Class<T>, entityIdAndVersion: EntityIdAndVersion, events: List<Event>, option val entityId = entityIdAndVersion.id val currentVersion = entityIdAndVersion.version val batch = TransactionalBatch.createTransactionalBatch(PartitionKey(entityId)) val container = getContainer() val entityDocumentReadResponse = container.readItem(entityId, PartitionKey(entityId), EventSourcingDocument::class.java val entityDocument = entityDocumentReadResponse.item if (currentVersion != entityDocument.version) { throw OptimisticConcurrencyException(entityId, currentVersion, entityDocument.version) } val eventDatamodels = events.mapIndexed { index, it -> EventDataModel( UUID.randomUUID().toString(), it.javaClass.name, clazz.name, entityId, currentVersion + index + 1, Gson().toJson(it) ) } eventDatamodels.map(EventDataModel::toDocument).forEach {batch.createItemOperation(it)} val latestVersion = eventDatamodels.last().version val entity = EntityDataModel.from(entityDocument) entity.version = latestVersion val replaceEntityDocument = entity.toDocument() batch.replaceItemOperation(entityId, replaceEntityDocument, TransactionalBatchItemRequestOptions().setIfMatchETag(entit
251. if (options?.snapshot != null) { val snapshot = options.snapshot val snapshotDataModel = SnapshotDataModel( UUID.randomUUID().toString(), clazz.name, entityIdAndVersion.id, latestVersion, snapshot!!.javaClass.name, Gson().toJson(snapshot) ) val snapshotDocument = snapshotDataModel.toDocument() batch.createItemOperation(snapshotDocument) } val result = container.executeTransactionalBatch(batch) if (!result.isSuccessStatusCode) { throw UnsupportedOperationException() } return EntityIdAndVersion(entity.id, latestVersion) }
252. if (options?.snapshot != null) { val snapshot = options.snapshot val snapshotDataModel = SnapshotDataModel( UUID.randomUUID().toString(), clazz.name, entityIdAndVersion.id, latestVersion, snapshot!!.javaClass.name, Gson().toJson(snapshot) ) val snapshotDocument = snapshotDataModel.toDocument() batch.createItemOperation(snapshotDocument) } val result = container.executeTransactionalBatch(batch) if (!result.isSuccessStatusCode) { throw UnsupportedOperationException() } return EntityIdAndVersion(entity.id, latestVersion) }
253. class AggregateRepository<T, CT>( private val clazz: Class<T>, private val aggregateStore: AggregateStore ) where T : CommandProcessingAggregate<T, CT>, CT : Command { ... fun update(id: String, cmd: CT, updateOption: UpdateOptions = UpdateOptions()) : EntityWithIdAndVersion<T> { val loadedData = aggregateStore.find(clazz, id) ?: throw IllegalArgumentException("$id is notfound") val aggregate = loadedData.entity val events = aggregate.processCommand(cmd) Aggregate.applyEvents(aggregate, events) val idAndVersion = aggregateStore.update( clazz, loadedData.idAndVersion, events, withPossibleSnapshot( updateOption, aggregate, loadedData.snapshotVersion, loadedData.events, events) ) return EntityWithIdAndVersion(idAndVersion, aggregate) } }
254. class AggregateRepository<T, CT>( private val clazz: Class<T>, private val aggregateStore: AggregateStore ) where T : CommandProcessingAggregate<T, CT>, CT : Command { ... fun update(id: String, cmd: CT, updateOption: UpdateOptions = UpdateOptions()) : EntityWithIdAndVersion<T> { val loadedData = aggregateStore.find(clazz, id) ?: throw IllegalArgumentException("$id is notfound") val aggregate = loadedData.entity val events = aggregate.processCommand(cmd) Aggregate.applyEvents(aggregate, events) val idAndVersion = aggregateStore.update( clazz, loadedData.idAndVersion, events, withPossibleSnapshot( updateOption, aggregate, loadedData.snapshotVersion, loadedData.events, events) ) return EntityWithIdAndVersion(idAndVersion, aggregate) } }
255. class AggregateRepository<T, CT>( private val clazz: Class<T>, private val aggregateStore: AggregateStore ) where T : CommandProcessingAggregate<T, CT>, CT : Command { ... fun update(id: String, cmd: CT, updateOption: UpdateOptions = UpdateOptions()) : EntityWithIdAndVersion<T> { val loadedData = aggregateStore.find(clazz, id) ?: throw IllegalArgumentException("$id is notfound") val aggregate = loadedData.entity val events = aggregate.processCommand(cmd) Aggregate.applyEvents(aggregate, events) val idAndVersion = aggregateStore.update( clazz, loadedData.idAndVersion, events, withPossibleSnapshot( updateOption, aggregate, loadedData.snapshotVersion, loadedData.events, events) ) return EntityWithIdAndVersion(idAndVersion, aggregate) } }
256. class OrderSnapshotStrategy : SnapshotStrategy { override fun getAggregateClass(): Class<*> = Order::class.java override fun possiblySnapshot( aggregate: Aggregate<*>, oldEvents: List<EventWithMetaData>, newEvents: List<Event>, snapshotVersion: String? ): Snapshot? { val order = aggregate as Order val snapshot = OrderSnapshot() snapshot.accountId = order.accountId snapshot.orderState = order.orderState snapshot.total = order.total snapshot.items = order.items return snapshot } override fun <T> recreateAggregate(clasz: Class<T>, snapshot: Snapshot): T where T : Aggregate<T> { val orderSnapshot = snapshot as OrderSnapshot val aggregate = Order() aggregate.accountId = orderSnapshot.accountId aggregate.orderState = orderSnapshot.orderState aggregate.total = orderSnapshot.total aggregate.items = orderSnapshot.items return aggregate as T } }
257. if (options?.snapshot != null) { val snapshot = options.snapshot val snapshotDataModel = SnapshotDataModel( UUID.randomUUID().toString(), clazz.name, entityIdAndVersion.id, latestVersion, snapshot!!.javaClass.name, Gson().toJson(snapshot) ) val snapshotDocument = snapshotDataModel.toDocument() batch.createItemOperation(snapshotDocument) } val result = container.executeTransactionalBatch(batch) if (!result.isSuccessStatusCode) { throw UnsupportedOperationException() } return EntityIdAndVersion(entity.id, latestVersion) }
259. val entity = EntityDataModel.from(entityDocument) val getEventsQuery = if (snapshotDataModel != null) { "SELECT * FROM x WHERE x.type = ¥"${DocumentType.Event}¥" AND x.version > ${entity.version} AND x.entityId = ¥"${en } else { "SELECT * FROM x WHERE x.type = ¥"${DocumentType.Event}¥" AND x.entityId = ¥"${entity.id}¥" ORDER BY x.version" } val eventDocuments = container.queryItems( getEventsQuery, CosmosQueryRequestOptions().setPartitionKey(PartitionKey(entity.id)), EventSourcingDocument::class.java ) val events = eventDocuments.map(EventDataModel.Companion::from).map { val eventClazz = Class.forName(it.eventType) val event = Gson().fromJson(it.payload, eventClazz) as Event EventWithMetaData(event, it.version) } if (snapshotDataModel == null && events.isEmpty()) { return null } val version = if (events.isEmpty()) { entity.version } else { events.last().version } val idAndVersion = EntityIdAndVersion(id, version) val recreatedAggregate = if (snapshotDataModel != null) {
261. override fun <T : Aggregate<T>> find(clazz: Class<T>, id: String): EntityWithMetadata<T>? { val container = getContainer() val readSnapshotDocumentResponse = container.queryItems( "SELECT * FROM x WHERE x.type = ¥"${DocumentType.Snapshot}¥" AND x.entityId = ¥"${id}¥" ORDER BY x.version DESC", CosmosQueryRequestOptions().setPartitionKey(PartitionKey(id)).setMaxBufferedItemCount(1), EventSourcingDocument::class.java ) val snapshotDataModel = readSnapshotDocumentResponse.map(SnapshotDataModel.Companion::from).firstOrNull() val readEntityDocumentResponse = container.readItem(id, PartitionKey(id), EventSourcingDocument::class.java) val entityDocument = readEntityDocumentResponse.item val entity = EntityDataModel.from(entityDocument) val getEventsQuery = if (snapshotDataModel != null) { "SELECT * FROM x WHERE x.type = ¥"${DocumentType.Event}¥" AND x.version > ${entity.version} AND x.entityId = ¥"${en } else { "SELECT * FROM x WHERE x.type = ¥"${DocumentType.Event}¥" AND x.entityId = ¥"${entity.id}¥" ORDER BY x.version" } val eventDocuments = container.queryItems( getEventsQuery, CosmosQueryRequestOptions().setPartitionKey(PartitionKey(entity.id)), EventSourcingDocument::class.java ) val events = eventDocuments.map(EventDataModel.Companion::from).map { val eventClazz = Class.forName(it.eventType) val event = Gson().fromJson(it.payload, eventClazz) as Event EventWithMetaData(event, it.version) }
262. override fun <T : Aggregate<T>> find(clazz: Class<T>, id: String): EntityWithMetadata<T>? { val container = getContainer() val readSnapshotDocumentResponse = container.queryItems( "SELECT * FROM x WHERE x.type = ¥"${DocumentType.Snapshot}¥" AND x.entityId = ¥"${id}¥" ORDER BY x.version DESC", CosmosQueryRequestOptions().setPartitionKey(PartitionKey(id)).setMaxBufferedItemCount(1), EventSourcingDocument::class.java ) val snapshotDataModel = readSnapshotDocumentResponse.map(SnapshotDataModel.Companion::from).firstOrNull() val readEntityDocumentResponse = container.readItem(id, PartitionKey(id), EventSourcingDocument::class.java) val entityDocument = readEntityDocumentResponse.item val entity = EntityDataModel.from(entityDocument) val getEventsQuery = if (snapshotDataModel != null) { "SELECT * FROM x WHERE x.type = ¥"${DocumentType.Event}¥" AND x.version > ${entity.version} AND x.entityId = ¥"${en } else { "SELECT * FROM x WHERE x.type = ¥"${DocumentType.Event}¥" AND x.entityId = ¥"${entity.id}¥" ORDER BY x.version" } val eventDocuments = container.queryItems( getEventsQuery, CosmosQueryRequestOptions().setPartitionKey(PartitionKey(entity.id)), EventSourcingDocument::class.java ) val events = eventDocuments.map(EventDataModel.Companion::from).map { val eventClazz = Class.forName(it.eventType) val event = Gson().fromJson(it.payload, eventClazz) as Event EventWithMetaData(event, it.version) }
263. override fun <T : Aggregate<T>> find(clazz: Class<T>, id: String): EntityWithMetadata<T>? { val container = getContainer() val readSnapshotDocumentResponse = container.queryItems( "SELECT * FROM x WHERE x.type = ¥"${DocumentType.Snapshot}¥" AND x.entityId = ¥"${id}¥" ORDER BY x.version DESC", CosmosQueryRequestOptions().setPartitionKey(PartitionKey(id)).setMaxBufferedItemCount(1), EventSourcingDocument::class.java ) val snapshotDataModel = readSnapshotDocumentResponse.map(SnapshotDataModel.Companion::from).firstOrNull() val readEntityDocumentResponse = container.readItem(id, PartitionKey(id), EventSourcingDocument::class.java) val entityDocument = readEntityDocumentResponse.item val entity = EntityDataModel.from(entityDocument) val getEventsQuery = if (snapshotDataModel != null) { "SELECT * FROM x WHERE x.type = ¥"${DocumentType.Event}¥" AND x.version > ${entity.version} AND x.entityId = ¥"${en } else { "SELECT * FROM x WHERE x.type = ¥"${DocumentType.Event}¥" AND x.entityId = ¥"${entity.id}¥" ORDER BY x.version" } val eventDocuments = container.queryItems( getEventsQuery, CosmosQueryRequestOptions().setPartitionKey(PartitionKey(entity.id)), EventSourcingDocument::class.java ) val events = eventDocuments.map(EventDataModel.Companion::from).map { val eventClazz = Class.forName(it.eventType) val event = Gson().fromJson(it.payload, eventClazz) as Event EventWithMetaData(event, it.version) }
264. override fun <T : Aggregate<T>> find(clazz: Class<T>, id: String): EntityWithMetadata<T>? { val container = getContainer() val readSnapshotDocumentResponse = container.queryItems( "SELECT * FROM x WHERE x.type = ¥"${DocumentType.Snapshot}¥" AND x.entityId = ¥"${id}¥" ORDER BY x.version DESC", CosmosQueryRequestOptions().setPartitionKey(PartitionKey(id)).setMaxBufferedItemCount(1), EventSourcingDocument::class.java ) val snapshotDataModel = readSnapshotDocumentResponse.map(SnapshotDataModel.Companion::from).firstOrNull() val readEntityDocumentResponse = container.readItem(id, PartitionKey(id), EventSourcingDocument::class.java) val entityDocument = readEntityDocumentResponse.item val entity = EntityDataModel.from(entityDocument) val getEventsQuery = if (snapshotDataModel != null) { "SELECT * FROM x WHERE x.type = ¥"${DocumentType.Event}¥" AND x.version > ${entity.version} AND x.entityId = ¥"${en } else { "SELECT * FROM x WHERE x.type = ¥"${DocumentType.Event}¥" AND x.entityId = ¥"${entity.id}¥" ORDER BY x.version" } val eventDocuments = container.queryItems( getEventsQuery, CosmosQueryRequestOptions().setPartitionKey(PartitionKey(entity.id)), EventSourcingDocument::class.java ) val events = eventDocuments.map(EventDataModel.Companion::from).map { val eventClazz = Class.forName(it.eventType) val event = Gson().fromJson(it.payload, eventClazz) as Event EventWithMetaData(event, it.version) }
265. override fun <T : Aggregate<T>> find(clazz: Class<T>, id: String): EntityWithMetadata<T>? { val container = getContainer() val readSnapshotDocumentResponse = container.queryItems( "SELECT * FROM x WHERE x.type = ¥"${DocumentType.Snapshot}¥" AND x.entityId = ¥"${id}¥" ORDER BY x.version DESC", CosmosQueryRequestOptions().setPartitionKey(PartitionKey(id)).setMaxBufferedItemCount(1), EventSourcingDocument::class.java ) val snapshotDataModel = readSnapshotDocumentResponse.map(SnapshotDataModel.Companion::from).firstOrNull() val readEntityDocumentResponse = container.readItem(id, PartitionKey(id), EventSourcingDocument::class.java) val entityDocument = readEntityDocumentResponse.item val entity = EntityDataModel.from(entityDocument) val getEventsQuery = if (snapshotDataModel != null) { "SELECT * FROM x WHERE x.type = ¥"${DocumentType.Event}¥" AND x.version > ${entity.version} AND x.entityId = ¥"${en } else { "SELECT * FROM x WHERE x.type = ¥"${DocumentType.Event}¥" AND x.entityId = ¥"${entity.id}¥" ORDER BY x.version" } val eventDocuments = container.queryItems( getEventsQuery, CosmosQueryRequestOptions().setPartitionKey(PartitionKey(entity.id)), EventSourcingDocument::class.java ) val events = eventDocuments.map(EventDataModel.Companion::from).map { val eventClazz = Class.forName(it.eventType) val event = Gson().fromJson(it.payload, eventClazz) as Event EventWithMetaData(event, it.version) }
266. if (snapshotDataModel == null && events.isEmpty()) { return null } val version = if (events.isEmpty()) { entity.version } else { events.last().version } val idAndVersion = EntityIdAndVersion(id, version) val recreatedAggregate = if (snapshotDataModel != null) { val snapshotClazz = Class.forName(snapshotDataModel.snapshotType) val snapshot = Gson().fromJson(snapshotDataModel.snapshotData, snapshotClazz) as Snapshot val aggregate = snapshotManager.recreateFromSnapshot(clazz, snapshot) Aggregate.applyEvents(aggregate, events.map { it.event }) } else { Aggregate.recreate(clazz, events.map { it.event }) } return EntityWithMetadata( idAndVersion, events, recreatedAggregate, snapshotDataModel?.entityVersion.toString() ) }
267. if (snapshotDataModel == null && events.isEmpty()) { return null } val version = if (events.isEmpty()) { entity.version } else { events.last().version } val idAndVersion = EntityIdAndVersion(id, version) val recreatedAggregate = if (snapshotDataModel != null) { val snapshotClazz = Class.forName(snapshotDataModel.snapshotType) val snapshot = Gson().fromJson(snapshotDataModel.snapshotData, snapshotClazz) as Snapshot val aggregate = snapshotManager.recreateFromSnapshot(clazz, snapshot) Aggregate.applyEvents(aggregate, events.map { it.event }) } else { Aggregate.recreate(clazz, events.map { it.event }) } return EntityWithMetadata( idAndVersion, events, recreatedAggregate, snapshotDataModel?.entityVersion.toString() ) }
268. if (snapshotDataModel == null && events.isEmpty()) { return null } val version = if (events.isEmpty()) { entity.version } else { events.last().version } val idAndVersion = EntityIdAndVersion(id, version) val recreatedAggregate = if (snapshotDataModel != null) { val snapshotClazz = Class.forName(snapshotDataModel.snapshotType) val snapshot = Gson().fromJson(snapshotDataModel.snapshotData, snapshotClazz) as Snapshot val aggregate = snapshotManager.recreateFromSnapshot(clazz, snapshot) Aggregate.applyEvents(aggregate, events.map { it.event }) } else { Aggregate.recreate(clazz, events.map { it.event }) } return EntityWithMetadata( idAndVersion, events, recreatedAggregate, snapshotDataModel?.entityVersion.toString() ) }
269. if (snapshotDataModel == null && events.isEmpty()) { return null } val version = if (events.isEmpty()) { entity.version } else { events.last().version } val idAndVersion = EntityIdAndVersion(id, version) val recreatedAggregate = if (snapshotDataModel != null) { val snapshotClazz = Class.forName(snapshotDataModel.snapshotType) val snapshot = Gson().fromJson(snapshotDataModel.snapshotData, snapshotClazz) as Snapshot val aggregate = snapshotManager.recreateFromSnapshot(clazz, snapshot) Aggregate.applyEvents(aggregate, events.map { it.event }) } else { Aggregate.recreate(clazz, events.map { it.event }) } return EntityWithMetadata( idAndVersion, events, recreatedAggregate, snapshotDataModel?.entityVersion.toString() ) }