mindol1004/Java-Spring-CRQS-Eventsourcing-Microservice
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: mindol1004:feature/monitoring
Branches Tags
mindol1004:master mindol1004:feature/improvements mindol1004:feature/resilience mindol1004:feature/monitoring mindol1004:feature/swagger mindol1004:feature/mongo mindol1004:feature/kafka
..
compare: mindol1004:feature/swagger
Branches Tags
mindol1004:master mindol1004:feature/improvements mindol1004:feature/resilience mindol1004:feature/monitoring mindol1004:feature/swagger mindol1004:feature/mongo mindol1004:feature/kafka

1 Commits
feature/mo ... feature/sw

Author SHA1 Message Date
Alexander
845e2e691d feature: add swagger openapi 2022-04-14 14:52:06 +03:00
7 changed files with 23 additions and 76 deletions
Expand all files Collapse all files

15
pom.xml
View File

@@ -17,21 +17,6 @@
<java.version>17</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-sleuth-zipkin</artifactId>
<version>3.1.1</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-sleuth</artifactId>
<version>3.1.1</version>
</dependency>
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-registry-prometheus</artifactId>
<version>1.8.4</version>
</dependency>
<dependency>
<groupId>org.springdoc</groupId>
<artifactId>springdoc-openapi-ui</artifactId>

16
src/main/java/com/eventsourcing/bankAccount/commands/BankAccountCommandHandler.java
View File

@@ -5,20 +5,17 @@ import com.eventsourcing.bankAccount.domain.BankAccountAggregate;
import com.eventsourcing.es.EventStoreDB;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.sleuth.annotation.NewSpan;
import org.springframework.cloud.sleuth.annotation.SpanTag;
import org.springframework.stereotype.Service;
@RequiredArgsConstructor
@Slf4j
@Service
public class BankAccountCommandHandler implements BankAccountCommandService {
public class BankAccountCommandHandler implements BankAccountCommandService{
private final EventStoreDB eventStoreDB;
@Override
@NewSpan
public String handle(@SpanTag("command") CreateBankAccountCommand command) {
public String handle(CreateBankAccountCommand command) {
final var aggregate = new BankAccountAggregate(command.aggregateID());
aggregate.createBankAccount(command.email(), command.address(), command.userName());
eventStoreDB.save(aggregate);
@@ -28,8 +25,7 @@ public class BankAccountCommandHandler implements BankAccountCommandService {
}
@Override
@NewSpan
public void handle(@SpanTag("command") ChangeEmailCommand command) {
public void handle(ChangeEmailCommand command) {
final var aggregate = eventStoreDB.load(command.aggregateID(), BankAccountAggregate.class);
aggregate.changeEmail(command.newEmail());
eventStoreDB.save(aggregate);
@@ -37,8 +33,7 @@ public class BankAccountCommandHandler implements BankAccountCommandService {
}
@Override
@NewSpan
public void handle(@SpanTag("command") ChangeAddressCommand command) {
public void handle(ChangeAddressCommand command) {
final var aggregate = eventStoreDB.load(command.aggregateID(), BankAccountAggregate.class);
aggregate.changeAddress(command.newAddress());
eventStoreDB.save(aggregate);
@@ -46,8 +41,7 @@ public class BankAccountCommandHandler implements BankAccountCommandService {
}
@Override
@NewSpan
public void handle(@SpanTag("command") DepositAmountCommand command) {
public void handle(DepositAmountCommand command) {
final var aggregate = eventStoreDB.load(command.aggregateID(), BankAccountAggregate.class);
aggregate.depositBalance(command.amount());
eventStoreDB.save(aggregate);

20
src/main/java/com/eventsourcing/bankAccount/projection/BankAccountMongoProjection.java
View File

@@ -15,8 +15,6 @@ import com.eventsourcing.es.SerializerUtils;
import com.eventsourcing.mappers.BankAccountMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.sleuth.annotation.NewSpan;
import org.springframework.cloud.sleuth.annotation.SpanTag;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.adapter.ConsumerRecordMetadata;
import org.springframework.kafka.support.Acknowledgment;
@@ -56,8 +54,7 @@ public class BankAccountMongoProjection implements Projection {
}
}
@NewSpan
private void processEvents(@SpanTag("events") List<Event> events) {
private void processEvents(List<Event> events) {
if (events.isEmpty()) return;
try {
@@ -72,8 +69,7 @@ public class BankAccountMongoProjection implements Projection {
}
@Override
@NewSpan
public void when(@SpanTag("event") Event event) {
public void when(Event event) {
final var aggregateId = event.getAggregateId();
log.info("(when) >>>>> aggregateId: {}", aggregateId);
@@ -91,8 +87,7 @@ public class BankAccountMongoProjection implements Projection {
}
@NewSpan
private void handle(@SpanTag("event") BankAccountCreatedEvent event) {
private void handle(BankAccountCreatedEvent event) {
log.info("(when) BankAccountCreatedEvent: {}, aggregateID: {}", event, event.getAggregateId());
final var document = BankAccountDocument.builder()
@@ -107,8 +102,7 @@ public class BankAccountMongoProjection implements Projection {
log.info("(BankAccountCreatedEvent) insert: {}", insert);
}
@NewSpan
private void handle(@SpanTag("event") EmailChangedEvent event) {
private void handle(EmailChangedEvent event) {
log.info("(when) EmailChangedEvent: {}, aggregateID: {}", event, event.getAggregateId());
Optional<BankAccountDocument> documentOptional = mongoRepository.findByAggregateId(event.getAggregateId());
if (documentOptional.isEmpty())
@@ -119,8 +113,7 @@ public class BankAccountMongoProjection implements Projection {
mongoRepository.save(document);
}
@NewSpan
private void handle(@SpanTag("event") AddressUpdatedEvent event) {
private void handle(AddressUpdatedEvent event) {
log.info("(when) AddressUpdatedEvent: {}, aggregateID: {}", event, event.getAggregateId());
Optional<BankAccountDocument> documentOptional = mongoRepository.findByAggregateId(event.getAggregateId());
if (documentOptional.isEmpty())
@@ -131,8 +124,7 @@ public class BankAccountMongoProjection implements Projection {
mongoRepository.save(document);
}
@NewSpan
private void handle(@SpanTag("event") BalanceDepositedEvent event) {
private void handle(BalanceDepositedEvent event) {
log.info("(when) BalanceDepositedEvent: {}, aggregateID: {}", event, event.getAggregateId());
Optional<BankAccountDocument> documentOptional = mongoRepository.findByAggregateId(event.getAggregateId());
if (documentOptional.isEmpty())

5
src/main/java/com/eventsourcing/bankAccount/queries/BankAccountQueryHandler.java
View File

@@ -8,8 +8,6 @@ import com.eventsourcing.es.EventStoreDB;
import com.eventsourcing.mappers.BankAccountMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.sleuth.annotation.NewSpan;
import org.springframework.cloud.sleuth.annotation.SpanTag;
import org.springframework.stereotype.Service;
import java.util.Optional;
@@ -24,8 +22,7 @@ public class BankAccountQueryHandler implements BankAccountQueryService {
private final BankAccountMongoRepository mongoRepository;
@Override
@NewSpan
public BankAccountResponseDTO handle(@SpanTag("query") GetBankAccountByIDQuery query) {
public BankAccountResponseDTO handle(GetBankAccountByIDQuery query) {
Optional<BankAccountDocument> optionalDocument = mongoRepository.findByAggregateId(query.aggregateID());
if (optionalDocument.isPresent()) {
return BankAccountMapper.bankAccountResponseDTOFromDocument(optionalDocument.get());

32
src/main/java/com/eventsourcing/es/EventStore.java
View File

@@ -4,8 +4,6 @@ package com.eventsourcing.es;
import com.eventsourcing.es.exceptions.AggregateNotFoundException;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.sleuth.annotation.NewSpan;
import org.springframework.cloud.sleuth.annotation.SpanTag;
import org.springframework.dao.EmptyResultDataAccessException;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.stereotype.Repository;
@@ -30,8 +28,7 @@ public class EventStore implements EventStoreDB {
private final EventBus eventBus;
@Override
@NewSpan
public void saveEvents(@SpanTag("events") List<Event> events) {
public void saveEvents(List<Event> events) {
if (events.isEmpty()) return;
final List<Event> changes = new ArrayList<>(events);
@@ -49,8 +46,7 @@ public class EventStore implements EventStoreDB {
}
@Override
@NewSpan
public List<Event> loadEvents(@SpanTag("aggregateId") String aggregateId, @SpanTag("version") long version) {
public List<Event> loadEvents(String aggregateId, long version) {
final List<Event> events = jdbcTemplate.query(LOAD_EVENTS_QUERY, Map.of("aggregate_id", aggregateId, "version", version),
(rs, rowNum) -> Event.builder()
.aggregateId(rs.getString("aggregate_id"))
@@ -66,8 +62,7 @@ public class EventStore implements EventStoreDB {
return events;
}
@NewSpan
private <T extends AggregateRoot> void saveSnapshot(@SpanTag("aggregate") T aggregate) {
private <T extends AggregateRoot> void saveSnapshot(T aggregate) {
aggregate.toSnapshot();
final var snapshot = EventSourcingUtils.snapshotFromAggregate(aggregate);
@@ -83,8 +78,7 @@ public class EventStore implements EventStoreDB {
@Override
@Transactional
@NewSpan
public <T extends AggregateRoot> void save(@SpanTag("aggregate") T aggregate) {
public <T extends AggregateRoot> void save(T aggregate) {
final List<Event> aggregateEvents = new ArrayList<>(aggregate.getChanges());
if (aggregate.getVersion() > 1) {
@@ -101,8 +95,7 @@ public class EventStore implements EventStoreDB {
log.info("(save) saved aggregate: {}", aggregate);
}
@NewSpan
private void handleConcurrency(@SpanTag("aggregateId") String aggregateId) {
private void handleConcurrency(String aggregateId) {
try {
String aggregateID = jdbcTemplate.queryForObject(HANDLE_CONCURRENCY_QUERY, Map.of("aggregate_id", aggregateId), String.class);
log.info("(handleConcurrency) aggregateID for lock: {}", aggregateID);
@@ -112,8 +105,7 @@ public class EventStore implements EventStoreDB {
log.info("(handleConcurrency) aggregateID for lock: {}", aggregateId);
}
@NewSpan
private Optional<Snapshot> loadSnapshot(@SpanTag("aggregateId") String aggregateId) {
private Optional<Snapshot> loadSnapshot(String aggregateId) {
final Optional<Snapshot> snapshot = jdbcTemplate.query(LOAD_SNAPSHOT_QUERY, Map.of("aggregate_id", aggregateId), (rs, rowNum) -> Snapshot.builder()
.aggregateId(rs.getString("aggregate_id"))
.aggregateType(rs.getString("aggregate_type"))
@@ -127,8 +119,7 @@ public class EventStore implements EventStoreDB {
return snapshot;
}
@NewSpan
private <T extends AggregateRoot> T getAggregate(@SpanTag("aggregateId") final String aggregateId, @SpanTag("aggregateType") final Class<T> aggregateType) {
private <T extends AggregateRoot> T getAggregate(final String aggregateId, final Class<T> aggregateType) {
try {
return aggregateType.getConstructor(String.class).newInstance(aggregateId);
} catch (Exception ex) {
@@ -136,8 +127,7 @@ public class EventStore implements EventStoreDB {
}
}
@NewSpan
private <T extends AggregateRoot> T getSnapshotFromClass(@SpanTag("snapshot") Optional<Snapshot> snapshot, @SpanTag("aggregateId") String aggregateId, @SpanTag("aggregateType") Class<T> aggregateType) {
private <T extends AggregateRoot> T getSnapshotFromClass(Optional<Snapshot> snapshot, String aggregateId, Class<T> aggregateType) {
if (snapshot.isEmpty()) {
final var defaultSnapshot = EventSourcingUtils.snapshotFromAggregate(getAggregate(aggregateId, aggregateType));
return EventSourcingUtils.aggregateFromSnapshot(defaultSnapshot, aggregateType);
@@ -147,8 +137,7 @@ public class EventStore implements EventStoreDB {
@Override
@Transactional(readOnly = true)
@NewSpan
public <T extends AggregateRoot> T load(@SpanTag("aggregateId") String aggregateId, @SpanTag("aggregateType") Class<T> aggregateType) {
public <T extends AggregateRoot> T load(String aggregateId, Class<T> aggregateType) {
final Optional<Snapshot> snapshot = this.loadSnapshot(aggregateId);
@@ -167,8 +156,7 @@ public class EventStore implements EventStoreDB {
}
@Override
@NewSpan
public Boolean exists(@SpanTag("aggregateId") String aggregateId) {
public Boolean exists(String aggregateId) {
try {
final var id = jdbcTemplate.queryForObject(EXISTS_QUERY, Map.of("aggregate_id", aggregateId), String.class);
log.info("aggregate exists id: {}", id);

5
src/main/java/com/eventsourcing/es/KafkaEventBus.java
View File

@@ -5,8 +5,6 @@ import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.sleuth.annotation.NewSpan;
import org.springframework.cloud.sleuth.annotation.SpanTag;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
@@ -23,8 +21,7 @@ public class KafkaEventBus implements EventBus {
private String bankAccountTopicName;
@Override
@NewSpan
public void publish(@SpanTag("events") List<Event> events) {
public void publish(List<Event> events) {
final byte[] eventsBytes = SerializerUtils.serializeToJsonBytes(events.toArray(new Event[]{}));
final ProducerRecord<String, byte[]> record = new ProducerRecord<>(bankAccountTopicName, eventsBytes);

6
src/main/resources/application.properties
View File

@@ -38,9 +38,3 @@ spring.data.mongodb.password=admin
spring.data.mongodb.database=microservices
springdoc.swagger-ui.path=/swagger-ui.html
management.endpoints.web.exposure.include=health,prometheus,info
spring.sleuth.propagation.type=w3c,b3
spring.sleuth.opentracing.enabled=true
spring.zipkin.base-url=http://localhost:9411