Merge branch 'wip-channel-mapping' of https://github.com/microservice-patterns/ftgo-application into wip-separate-cdc

Author: dartartem

build-and-test-all.sh

@@ -41,7 +41,7 @@ ${DOCKER_COMPOSE?} up -d --build ${DATABASE_SERVICES?}
 
 echo mysql is started
 
-${DOCKER_COMPOSE?} up -d --build tram-cdc-service
+${DOCKER_COMPOSE?} up -d --build cdc-service
 
 if [ -z "$ASSEMBLE_ONLY" ] ; then
 

docker-compose-mysql-schema-per-service.yml

@@ -82,7 +82,7 @@ services:
     depends_on:
       - mysql
       - kafka
-      - tram-cdc-service
+      - cdc-service
     environment:
       SPRING_DATASOURCE_URL: jdbc:mysql://mysql/ftgoconsumerservice
       SPRING_DATASOURCE_USERNAME: mysqluser
@@ -99,7 +99,7 @@ services:
     depends_on:
       - mysql
       - kafka
-      - tram-cdc-service
+      - cdc-service
       - zipkin
     environment:
       SPRING_DATASOURCE_URL: jdbc:mysql://mysql/ftgoorderservice
@@ -121,7 +121,7 @@ services:
         - mysql
         - kafka
         - zookeeper
-        - tram-cdc-service
+        - cdc-service
       environment:
         SPRING_DATASOURCE_URL: jdbc:mysql://mysql/ftgokitchenservice
         SPRING_DATASOURCE_USERNAME: mysqluser
@@ -139,7 +139,7 @@ services:
         - mysql
         - kafka
         - zookeeper
-        - tram-cdc-service
+        - cdc-service
       environment:
         SPRING_DATASOURCE_URL: jdbc:mysql://mysql/ftgorestaurantservice
         SPRING_DATASOURCE_USERNAME: mysqluser
@@ -173,7 +173,7 @@ services:
         depends_on:
           - kafka
           - zookeeper
-          - tram-cdc-service
+          - cdc-service
           - dynamodblocal
           - dynamodblocal-init
         environment:

docker-compose.yml

@@ -24,7 +24,7 @@ services:
       - MYSQL_ROOT_PASSWORD=rootpassword
       - MYSQL_USER=mysqluser
       - MYSQL_PASSWORD=mysqlpw
-  tram-cdc-service:
+  cdc-service:
     image: test-eventuate-cdc-service
     ports:
       - "8099:8080"
@@ -74,7 +74,7 @@ services:
     depends_on:
       - mysql
       - kafka
-      - tram-cdc-service
+      - cdc-service
     environment:
       SPRING_DATASOURCE_URL: jdbc:mysql://mysql/ftgoconsumerservice
       SPRING_DATASOURCE_USERNAME: mysqluser
@@ -93,7 +93,7 @@ services:
     depends_on:
       - mysql
       - kafka
-      - tram-cdc-service
+      - cdc-service
       - zipkin
     environment:
       SPRING_DATASOURCE_URL: jdbc:mysql://mysql/ftgoorderservice
@@ -114,7 +114,7 @@ services:
         - mysql
         - kafka
         - zookeeper
-        - tram-cdc-service
+        - cdc-service
       environment:
         SPRING_DATASOURCE_URL: jdbc:mysql://mysql/ftgokitchenservice
         SPRING_DATASOURCE_USERNAME: mysqluser
@@ -134,7 +134,7 @@ services:
         - mysql
         - kafka
         - zookeeper
-        - tram-cdc-service
+        - cdc-service
       environment:
         SPRING_DATASOURCE_URL: jdbc:mysql://mysql/ftgorestaurantservice
         SPRING_DATASOURCE_USERNAME: mysqluser
@@ -172,7 +172,7 @@ services:
         depends_on:
           - kafka
           - zookeeper
-          - tram-cdc-service
+          - cdc-service
           - dynamodblocal
           - dynamodblocal-init
         environment:

ftgo-accounting-service/src/main/java/net/chrisrichardson/ftgo/accountingservice/main/AccountingServiceMain.java

@@ -21,11 +21,6 @@
         TramJdbcKafkaConfiguration.class})
 public class AccountingServiceMain {
 
-  @Bean
-  public ChannelMapping channelMapping() {
-    return new DefaultChannelMapping.DefaultChannelMappingBuilder().build();
-  }
-
   public static void main(String[] args) {
     SpringApplication.run(AccountingServiceMain.class, args);
   }

ftgo-accounting-service/src/main/java/net/chrisrichardson/ftgo/accountingservice/messaging/AccountingMessagingConfiguration.java

@@ -2,11 +2,13 @@
 
 import io.eventuate.javaclient.spring.EnableEventHandlers;
 import io.eventuate.tram.commands.consumer.CommandDispatcher;
+import io.eventuate.tram.commands.consumer.CommandDispatcherFactory;
+import io.eventuate.tram.commands.consumer.TramCommandConsumerConfiguration;
 import io.eventuate.tram.consumer.common.DuplicateMessageDetector;
-import io.eventuate.tram.events.common.DefaultDomainEventNameMapping;
 import io.eventuate.tram.events.subscriber.DomainEventDispatcher;
+import io.eventuate.tram.events.subscriber.DomainEventDispatcherFactory;
+import io.eventuate.tram.events.subscriber.TramEventSubscriberConfiguration;
 import io.eventuate.tram.messaging.consumer.MessageConsumer;
-import io.eventuate.tram.messaging.producer.MessageProducer;
 import io.eventuate.tram.sagas.eventsourcingsupport.SagaReplyRequestedEventSubscriber;
 import net.chrisrichardson.ftgo.accountingservice.domain.Account;
 import net.chrisrichardson.ftgo.accountingservice.domain.AccountServiceConfiguration;
@@ -19,7 +21,7 @@
 
 @Configuration
 @EnableEventHandlers
-@Import({AccountServiceConfiguration.class, CommonConfiguration.class})
+@Import({AccountServiceConfiguration.class, CommonConfiguration.class, TramEventSubscriberConfiguration.class, TramCommandConsumerConfiguration.class})
 public class AccountingMessagingConfiguration {
 
   @Bean
@@ -28,11 +30,8 @@ public AccountingEventConsumer accountingEventConsumer() {
   }
 
   @Bean
-  public DomainEventDispatcher domainEventDispatcher(AccountingEventConsumer accountingEventConsumer, MessageConsumer messageConsumer) {
-    return new DomainEventDispatcher("accountingServiceDomainEventDispatcher",
-            accountingEventConsumer.domainEventHandlers(),
-            messageConsumer,
-            new DefaultDomainEventNameMapping());
+  public DomainEventDispatcher domainEventDispatcher(AccountingEventConsumer accountingEventConsumer, DomainEventDispatcherFactory domainEventDispatcherFactory) {
+    return domainEventDispatcherFactory.make("accountingServiceDomainEventDispatcher", accountingEventConsumer.domainEventHandlers());
   }
 
   @Bean
@@ -43,10 +42,8 @@ public AccountingServiceCommandHandler accountCommandHandler() {
 
   @Bean
   public CommandDispatcher commandDispatcher(AccountingServiceCommandHandler target,
-                                             AccountServiceChannelConfiguration data,
-                                             MessageConsumer messageConsumer,
-                                             MessageProducer messageProducer) {
-    return new CommandDispatcher(data.getCommandDispatcherId(), target.commandHandlers(), messageConsumer, messageProducer);
+                                             AccountServiceChannelConfiguration data, CommandDispatcherFactory commandDispatcherFactory) {
+    return commandDispatcherFactory.make(data.getCommandDispatcherId(), target.commandHandlers());
   }
 
   @Bean

ftgo-accounting-service/src/test/java/net/chrisrichardson/ftgo/accountingservice/messaging/AccountingServiceCommandHandlerTest.java

@@ -57,11 +57,6 @@ public TestMessageConsumerFactory testMessageConsumerFactory() {
       return new TestMessageConsumerFactory();
     }
 
-    @Bean
-    public ChannelMapping channelMapping() {
-      return new DefaultChannelMapping.DefaultChannelMappingBuilder().build();
-    }
-
     @Bean
     public DataSource dataSource() {
       EmbeddedDatabaseBuilder builder = new EmbeddedDatabaseBuilder();

ftgo-consumer-service/src/main/java/net/chrisrichardson/ftgo/consumerservice/domain/ConsumerServiceConfiguration.java

@@ -2,12 +2,7 @@
 
 import io.eventuate.tram.commands.consumer.CommandDispatcher;
 import io.eventuate.tram.events.publisher.TramEventsPublisherConfiguration;
-import io.eventuate.tram.messaging.common.ChannelMapping;
-import io.eventuate.tram.messaging.common.DefaultChannelMapping;
-import io.eventuate.tram.messaging.consumer.MessageConsumer;
-import io.eventuate.tram.messaging.producer.MessageProducer;
-import io.eventuate.tram.sagas.common.SagaLockManager;
-import io.eventuate.tram.sagas.participant.SagaCommandDispatcher;
+import io.eventuate.tram.sagas.participant.SagaCommandDispatcherFactory;
 import io.eventuate.tram.sagas.participant.SagaParticipantConfiguration;
 import net.chrisrichardson.ftgo.common.CommonConfiguration;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
@@ -21,7 +16,7 @@
 @Configuration
 @EnableJpaRepositories
 @EnableAutoConfiguration
-@Import({SagaParticipantConfiguration.class, TramEventsPublisherConfiguration.class, CommonConfiguration.class})
+@Import({SagaParticipantConfiguration.class, TramEventsPublisherConfiguration.class, CommonConfiguration.class, SagaParticipantConfiguration.class})
 @EnableTransactionManagement
 @ComponentScan
 public class ConsumerServiceConfiguration {
@@ -37,20 +32,9 @@ public ConsumerService consumerService() {
   }
 
   @Bean
-  public CommandDispatcher commandDispatcher(ConsumerServiceCommandHandlers consumerServiceCommandHandlers,
-                                             MessageConsumer messageConsumer,
-                                             MessageProducer messageProducer,
-                                             SagaLockManager sagaLockManager) {
-    return new SagaCommandDispatcher("consumerServiceDispatcher",
-            consumerServiceCommandHandlers.commandHandlers(),
-            messageConsumer,
-            messageProducer,
-            sagaLockManager);
+  public CommandDispatcher commandDispatcher(ConsumerServiceCommandHandlers consumerServiceCommandHandlers, SagaCommandDispatcherFactory sagaCommandDispatcherFactory) {
+    return sagaCommandDispatcherFactory.make("consumerServiceDispatcher", consumerServiceCommandHandlers.commandHandlers());
   }
 
-  @Bean
-  public ChannelMapping channelMapping() {
-    return new DefaultChannelMapping.DefaultChannelMappingBuilder().build();
-  }
 
 }

ftgo-kitchen-service/src/main/java/net/chrisrichardson/ftgo/kitchenservice/main/KitchenServiceMain.java

@@ -1,8 +1,6 @@
 package net.chrisrichardson.ftgo.kitchenservice.main;
 
 import io.eventuate.jdbckafka.TramJdbcKafkaConfiguration;
-import io.eventuate.tram.messaging.common.ChannelMapping;
-import io.eventuate.tram.messaging.common.DefaultChannelMapping;
 import net.chrisrichardson.eventstore.examples.customersandorders.commonswagger.CommonSwaggerConfiguration;
 import net.chrisrichardson.ftgo.kitchenservice.messagehandlers.KitchenServiceMessageHandlersConfiguration;
 import net.chrisrichardson.ftgo.kitchenservice.web.KitchenServiceWebConfiguration;
@@ -17,11 +15,6 @@
         CommonSwaggerConfiguration.class})
 public class KitchenServiceMain {
 
-  @Bean
-  public ChannelMapping channelMapping() {
-    return new DefaultChannelMapping.DefaultChannelMappingBuilder().build();
-  }
-
   public static void main(String[] args) {
     SpringApplication.run(KitchenServiceMain.class, args);
   }

ftgo-kitchen-service/src/main/java/net/chrisrichardson/ftgo/kitchenservice/messagehandlers/KitchenServiceMessageHandlersConfiguration.java

@@ -1,11 +1,11 @@
 package net.chrisrichardson.ftgo.kitchenservice.messagehandlers;
 
-import io.eventuate.tram.events.common.DefaultDomainEventNameMapping;
 import io.eventuate.tram.events.subscriber.DomainEventDispatcher;
+import io.eventuate.tram.events.subscriber.DomainEventDispatcherFactory;
+import io.eventuate.tram.events.subscriber.TramEventSubscriberConfiguration;
 import io.eventuate.tram.messaging.consumer.MessageConsumer;
-import io.eventuate.tram.messaging.producer.MessageProducer;
-import io.eventuate.tram.sagas.common.SagaLockManager;
 import io.eventuate.tram.sagas.participant.SagaCommandDispatcher;
+import io.eventuate.tram.sagas.participant.SagaCommandDispatcherFactory;
 import io.eventuate.tram.sagas.participant.SagaParticipantConfiguration;
 import net.chrisrichardson.ftgo.common.CommonConfiguration;
 import net.chrisrichardson.ftgo.kitchenservice.domain.KitchenDomainConfiguration;
@@ -14,7 +14,7 @@
 import org.springframework.context.annotation.Import;
 
 @Configuration
-@Import({KitchenDomainConfiguration.class, SagaParticipantConfiguration.class, CommonConfiguration.class})
+@Import({KitchenDomainConfiguration.class, SagaParticipantConfiguration.class, CommonConfiguration.class, TramEventSubscriberConfiguration.class, SagaParticipantConfiguration.class})
 public class KitchenServiceMessageHandlersConfiguration {
 
   @Bean
@@ -28,22 +28,12 @@ public KitchenServiceCommandHandler kitchenServiceCommandHandler() {
   }
 
   @Bean
-  public SagaCommandDispatcher kitchenServiceSagaCommandDispatcher(KitchenServiceCommandHandler kitchenServiceCommandHandler,
-                                                                   MessageConsumer messageConsumer,
-                                                                   MessageProducer messageProducer,
-                                                                   SagaLockManager sagaLockManager) {
-    return new SagaCommandDispatcher("kitchenServiceCommands",
-            kitchenServiceCommandHandler.commandHandlers(),
-            messageConsumer,
-            messageProducer,
-            sagaLockManager);
+  public SagaCommandDispatcher kitchenServiceSagaCommandDispatcher(KitchenServiceCommandHandler kitchenServiceCommandHandler, SagaCommandDispatcherFactory sagaCommandDispatcherFactory) {
+    return sagaCommandDispatcherFactory.make("kitchenServiceCommands", kitchenServiceCommandHandler.commandHandlers());
   }
 
   @Bean
-  public DomainEventDispatcher domainEventDispatcher(KitchenServiceEventConsumer kitchenServiceEventConsumer, MessageConsumer messageConsumer) {
-    return new DomainEventDispatcher("kitchenServiceEvents",
-            kitchenServiceEventConsumer.domainEventHandlers(),
-            messageConsumer,
-            new DefaultDomainEventNameMapping()); // @Autowire
+  public DomainEventDispatcher domainEventDispatcher(KitchenServiceEventConsumer kitchenServiceEventConsumer, DomainEventDispatcherFactory domainEventDispatcherFactory) {
+    return domainEventDispatcherFactory.make("kitchenServiceEvents", kitchenServiceEventConsumer.domainEventHandlers());
   }
 }

ftgo-kitchen-service/src/test/java/net/chrisrichardson/ftgo/kitchenservice/domain/KitchenServiceInMemoryIntegrationTest.java

@@ -51,11 +51,6 @@ public class KitchenServiceInMemoryIntegrationTest {
           TramInMemoryConfiguration.class})
   public static class TestConfiguration {
 
-    @Bean
-    public ChannelMapping channelMapping() {
-      return new DefaultChannelMapping.DefaultChannelMappingBuilder().build();
-    }
-
     @Bean
     public TestMessageConsumerFactory testMessageConsumerFactory() {
       return new TestMessageConsumerFactory();

ftgo-order-history-service/src/main/java/net/chrisrichardson/ftgo/cqrs/orderhistory/messaging/OrderHistoryServiceMessagingConfiguration.java

@@ -4,17 +4,17 @@
 import io.eventuate.tram.consumer.common.MessageConsumerImplementation;
 import io.eventuate.tram.consumer.common.TramNoopDuplicateMessageDetectorConfiguration;
 import io.eventuate.tram.consumer.wrappers.EventuateKafkaMessageConsumerWrapper;
-import io.eventuate.tram.events.common.DefaultDomainEventNameMapping;
 import io.eventuate.tram.events.subscriber.DomainEventDispatcher;
-import io.eventuate.tram.messaging.consumer.MessageConsumer;
+import io.eventuate.tram.events.subscriber.DomainEventDispatcherFactory;
+import io.eventuate.tram.events.subscriber.TramEventSubscriberConfiguration;
 import net.chrisrichardson.ftgo.common.CommonConfiguration;
 import net.chrisrichardson.ftgo.cqrs.orderhistory.OrderHistoryDao;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 
 @Configuration
-@Import({CommonConfiguration.class, TramNoopDuplicateMessageDetectorConfiguration.class})
+@Import({CommonConfiguration.class, TramNoopDuplicateMessageDetectorConfiguration.class, TramEventSubscriberConfiguration.class})
 public class OrderHistoryServiceMessagingConfiguration {
 
   @Bean
@@ -28,11 +28,8 @@ public OrderHistoryEventHandlers orderHistoryEventHandlers(OrderHistoryDao order
   }
 
   @Bean
-  public DomainEventDispatcher orderHistoryDomainEventDispatcher(OrderHistoryEventHandlers orderHistoryEventHandlers, MessageConsumer messageConsumer) {
-    return new DomainEventDispatcher("orderHistoryDomainEventDispatcher",
-            orderHistoryEventHandlers.domainEventHandlers(),
-            messageConsumer,
-            new DefaultDomainEventNameMapping());
+  public DomainEventDispatcher orderHistoryDomainEventDispatcher(OrderHistoryEventHandlers orderHistoryEventHandlers, DomainEventDispatcherFactory domainEventDispatcherFactory) {
+    return domainEventDispatcherFactory.make("orderHistoryDomainEventDispatcher", orderHistoryEventHandlers.domainEventHandlers());
   }
 
 }