From 920ce9149622268b5895255c188ab76b66ca7875 Mon Sep 17 00:00:00 2001 From: Alexander Date: Thu, 14 Apr 2022 14:42:24 +0300 Subject: [PATCH] feature: add bank account mongo projection improvements update --- .../projection/BankAccountMongoProjection.java | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/src/main/java/com/eventsourcing/bankAccount/projection/BankAccountMongoProjection.java b/src/main/java/com/eventsourcing/bankAccount/projection/BankAccountMongoProjection.java index f01015f..6ed2f34 100644 --- a/src/main/java/com/eventsourcing/bankAccount/projection/BankAccountMongoProjection.java +++ b/src/main/java/com/eventsourcing/bankAccount/projection/BankAccountMongoProjection.java @@ -15,7 +15,6 @@ import com.eventsourcing.es.SerializerUtils; import com.eventsourcing.mappers.BankAccountMapper; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.annotation.Value; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.listener.adapter.ConsumerRecordMetadata; import org.springframework.kafka.support.Acknowledgment; @@ -35,9 +34,6 @@ public class BankAccountMongoProjection implements Projection { private final BankAccountMongoRepository mongoRepository; private final EventStoreDB eventStoreDB; - @Value(value = "${microservice.kafka.topics.bank-account-event-store:bank-account-event-store}") - private String bankAccountTopicName; - @KafkaListener(topics = {"${microservice.kafka.topics.bank-account-event-store}"}, groupId = "${microservice.kafka.groupId}", @@ -59,14 +55,15 @@ public class BankAccountMongoProjection implements Projection { } private void processEvents(List events) { + if (events.isEmpty()) return; + try { events.forEach(this::when); } catch (Exception ex) { - // TODO: delete from mongo, get from eventStore, upsert to mongo mongoRepository.deleteByAggregateId(events.get(0).getAggregateId()); final var aggregate = eventStoreDB.load(events.get(0).getAggregateId(), BankAccountAggregate.class); final var document = BankAccountMapper.bankAccountDocumentFromAggregate(aggregate); - BankAccountDocument result = mongoRepository.save(document); + final var result = mongoRepository.save(document); log.info("(processEvents) saved document: {}", result); } }