Added AccountEventProducer to publish events on Kafka

bank-acct/acct.cmd/src/main/java/co/marcuss/acct/cmd/infrastructure/AccountEventProducer.java

@@ -0,0 +1,19 @@
+package co.marcuss.acct.cmd.infrastructure;
+
+import co.marcuss.cqrs.core.events.BaseEvent;
+import co.marcuss.cqrs.core.producers.EventProducer;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Service;
+
+@Service
+public class AccountEventProducer implements EventProducer {
+
+    @Autowired
+    private KafkaTemplate<String, Object> kafkaTemplate;
+
+    @Override
+    public void produce(String topic, BaseEvent event) {
+        this.kafkaTemplate.send(topic, event);
+    }
+}

bank-acct/acct.cmd/src/main/java/co/marcuss/acct/cmd/infrastructure/AccountEventStore.java

@@ -7,18 +7,20 @@
 import co.marcuss.cqrs.core.exceptions.AggregateNotFoundException;
 import co.marcuss.cqrs.core.exceptions.ConcurrencyException;
 import co.marcuss.cqrs.core.infrastructure.EventStore;
+import co.marcuss.cqrs.core.producers.EventProducer;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 import org.springframework.util.CollectionUtils;
 
-import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
 @Service
 public class AccountEventStore implements EventStore {
+    @Autowired
+    private EventProducer eventProducer;
     @Autowired
     private EventStoreRepository eventStoreRepository;
 
@@ -42,8 +44,8 @@ public void save(String aggregateId, Iterable<BaseEvent> events, int expectedVer
                             .eventData(e)
                             .build();
                     var persistedEvent = eventStoreRepository.save(eventModel);
-                    if (persistedEvent != null) {
-                        // TODO: produce event to kafka
+                    if (!persistedEvent.getId().isEmpty()) {
+                       eventProducer.produce(e.getClass().getSimpleName(), e);
                     }
                 }
         );

cqrs-es/cqrs.core/src/main/java/co/marcuss/cqrs/core/producers/EventProducer.java

@@ -0,0 +1,8 @@
+package co.marcuss.cqrs.core.producers;
+
+import co.marcuss.cqrs.core.events.BaseEvent;
+
+public interface EventProducer {
+
+    void produce(String topic, BaseEvent event);
+}