Update apiVersion in K8s objects, fix path

build.gradle

@@ -30,10 +30,11 @@ subprojects {
     group = "net.chrisrichardson.ftgo"
 
     repositories {
-        eventuateMavenRepoUrl.split('[ ,]').each { repoUrl -> maven { url repoUrl.trim() } }
-
         mavenCentral()
         jcenter()
+        eventuateMavenRepoUrl.split('[ ,]').each { repoUrl -> maven { url repoUrl.trim() } }
+
+
 
         maven {
             url 'https://jitpack.io'

deployment/kubernetes/cdc-service/ftgo-cdc-service.yml

@@ -9,13 +9,16 @@ spec:
   selector:
     svc: ftgo-cdc-service
 ---
-apiVersion: extensions/v1beta1
+apiVersion: apps/v1
 kind: Deployment
 metadata:
   name: ftgo-cdc-service
   labels:
     application: ftgo
 spec:
+  selector:
+    matchLabels:
+      svc: ftgo-cdc-service
   replicas: 1
   strategy:
     rollingUpdate:
@@ -27,7 +30,7 @@ spec:
     spec:
       containers:
       - name: ftgo-cdc-service
-        image: eventuateio/eventuate-cdc-service:0.4.0.RELEASE
+        image: eventuateio/eventuate-cdc-service:0.6.0.RC3
         imagePullPolicy: Always
         ports:
         - containerPort: 8080
@@ -37,7 +40,7 @@ spec:
           - name: JAVA_OPTS
             value: "-Dsun.net.inetaddr.ttl=30"
           - name: EVENTUATELOCAL_KAFKA_BOOTSTRAP_SERVERS
-            value: ftgo-kafka:29092
+            value: ftgo-kafka:9092
           - name: EVENTUATELOCAL_ZOOKEEPER_CONNECTION_STRING
             value: ftgo-zookeeper:2181
           - name: EVENTUATE_CDC_PIPELINE_PIPELINE1_TYPE

deployment/kubernetes/scripts/kubernetes-deploy-all.sh

@@ -1,9 +1,11 @@
 #! /bin/bash -e
 
+set -vx
+
 kubectl apply -f <(cat deployment/kubernetes/stateful-services/*.yml)
 
 ./deployment/kubernetes/scripts/kubernetes-wait-for-ready-pods.sh ftgo-mysql-0 ftgo-kafka-0 ftgo-dynamodb-local-0 ftgo-zookeeper-0
 
-kubectl apply -f <(cat deployment/kubernetes/cdc-services/*.yml)
+kubectl apply -f <(cat deployment/kubernetes/cdc-service/*.yml)
 
 kubectl apply -f <(cat */src/deployment/kubernetes/*.yml)

deployment/kubernetes/scripts/port-forwards.sh

@@ -15,7 +15,7 @@ doforward() {
 
 
 doforward 'ftgo-accounting-service' 8085 8080
-
+doforward 'ftgo-order-history-service' 8086 8080
 doforward 'ftgo-consumer-service' 8081 8080
 doforward 'ftgo-api-gateway' 8087 8080
 doforward 'ftgo-order-service' 8082 8080

deployment/kubernetes/stateful-services/ftgo-dynamodb-local.yml

@@ -9,13 +9,16 @@ spec:
   selector:
     svc: ftgo-dynamodb-local
 ---
-apiVersion: apps/v1beta1
+apiVersion: apps/v1
 kind: StatefulSet
 metadata:
   name: ftgo-dynamodb-local
   labels:
     application: ftgo
 spec:
+  selector:
+    matchLabels:
+      svc: ftgo-dynamodb-local
   serviceName: "ftgo-dynamodb"
   replicas: 1
   template:

deployment/kubernetes/stateful-services/ftgo-kafka-deployment.yml

@@ -11,11 +11,14 @@ spec:
   selector:
     role: ftgo-kafka
 ---
-apiVersion: apps/v1beta1
+apiVersion: apps/v1
 kind: StatefulSet
 metadata:
   name: ftgo-kafka
 spec:
+  selector:
+    matchLabels:
+      role: ftgo-kafka
   serviceName: "kafka"
   replicas: 1
   template:

deployment/kubernetes/stateful-services/ftgo-mysql-deployment.yml

@@ -12,11 +12,14 @@ spec:
   selector:
     role: ftgo-mysql
 ---
-apiVersion: apps/v1beta1
+apiVersion: apps/v1
 kind: StatefulSet
 metadata:
   name: ftgo-mysql
 spec:
+  selector:
+    matchLabels:
+      role: ftgo-mysql
   serviceName: "mysql"
   replicas: 1
   template:

deployment/kubernetes/stateful-services/ftgo-zookeeper-deployment.yml

@@ -12,11 +12,14 @@ spec:
   selector:
     role: ftgo-zookeeper
 ---
-apiVersion: apps/v1beta1
+apiVersion: apps/v1
 kind: StatefulSet
 metadata:
   name: ftgo-zookeeper
 spec:
+  selector:
+    matchLabels:
+      role: ftgo-zookeeper
   serviceName: "zookeeper"
   replicas: 1
   template:

docker-compose.yml

@@ -161,6 +161,7 @@ services:
           - kafka
           - zookeeper
           - cdc-service
+          - zipkin
         environment:
           SPRING_DATASOURCE_URL: jdbc:mysql://mysql/ftgo_accounting_service
           SPRING_DATASOURCE_USERNAME: ftgo_accounting_service_user
@@ -169,6 +170,9 @@ services:
           EVENTUATELOCAL_KAFKA_BOOTSTRAP_SERVERS: kafka:29092
           EVENTUATELOCAL_ZOOKEEPER_CONNECTION_STRING: zookeeper:2181
           EVENTUATE_DATABASE_SCHEMA: ftgo_accounting_service
+          SPRING_SLEUTH_ENABLED: "true"
+          SPRING_SLEUTH_SAMPLER_PROBABILITY: 1
+          SPRING_ZIPKIN_BASE_URL: http://zipkin:9411/
   ftgo-delivery-service:
     build: ./ftgo-delivery-service
     ports:

ftgo-accounting-service-api/src/main/java/net/chrisrichardson/ftgo/accountservice/api/AccountLimitExceededReply.java

@@ -0,0 +1,4 @@
+package net.chrisrichardson.ftgo.accountservice.api;
+
+public class AccountLimitExceededReply {
+}

ftgo-accounting-service-api/src/main/java/net/chrisrichardson/ftgo/accountservice/api/CheckAccountLimitCommand.java

@@ -0,0 +1,32 @@
+package net.chrisrichardson.ftgo.accountservice.api;
+
+import io.eventuate.tram.commands.common.Command;
+import net.chrisrichardson.ftgo.common.Money;
+
+public class CheckAccountLimitCommand implements Command {
+
+    private Money money;
+    private Long consumerId;
+    private Long orderId;
+
+    private CheckAccountLimitCommand() {
+    }
+
+    public CheckAccountLimitCommand(Long consumerId, Long orderId, Money money) {
+        this.money = money;
+        this.consumerId = consumerId;
+        this.orderId = orderId;
+    }
+
+    public Money getMoney() {
+        return money;
+    }
+
+    public Long getConsumerId() {
+        return consumerId;
+    }
+
+    public Long getOrderId() {
+        return orderId;
+    }
+}

ftgo-accounting-service/src/deployment/kubernetes/ftgo-accounting-service.yml

@@ -9,13 +9,16 @@ spec:
   selector:
     svc: ftgo-accounting-service
 ---
-apiVersion: extensions/v1beta1
+apiVersion: apps/v1
 kind: Deployment
 metadata:
   name: ftgo-accounting-service
   labels:
     application: ftgo
 spec:
+  selector:
+    matchLabels:
+        svc: ftgo-accounting-service
   replicas: 1
   strategy:
     rollingUpdate:
@@ -37,23 +40,19 @@ spec:
           - name: JAVA_OPTS
             value: "-Dsun.net.inetaddr.ttl=30"
           - name: SPRING_DATASOURCE_URL
-            value: jdbc:mysql://ftgo-mysql/eventuate
+            value: jdbc:mysql://ftgo-mysql/ftgo_accounting_service
           - name: SPRING_DATASOURCE_USERNAME
-            valueFrom:
-              secretKeyRef:
-                name: ftgo-db-secret
-                key: username
+            value: ftgo_accounting_service_user
           - name: SPRING_DATASOURCE_PASSWORD
-            valueFrom:
-              secretKeyRef:
-                name: ftgo-db-secret
-                key: password
+            value: ftgo_accounting_service_password
           - name: SPRING_DATASOURCE_DRIVER_CLASS_NAME
             value: com.mysql.jdbc.Driver
           - name: EVENTUATELOCAL_KAFKA_BOOTSTRAP_SERVERS
             value: ftgo-kafka:9092
           - name: EVENTUATELOCAL_ZOOKEEPER_CONNECTION_STRING
             value: ftgo-zookeeper:2181
+          - name: EVENTUATE_DATABASE_SCHEMA
+            value: ftgo_accounting_service
         livenessProbe:
           httpGet:
             path: /actuator/health

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

@@ -3,25 +3,54 @@
 import io.eventuate.Event;
 import io.eventuate.ReflectiveMutableCommandProcessingAggregate;
 import io.eventuate.tram.sagas.eventsourcingsupport.SagaReplyRequestedEvent;
+import net.chrisrichardson.ftgo.common.Money;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
+import java.math.BigDecimal;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 
 import static io.eventuate.EventUtil.events;
 
 public class Account extends ReflectiveMutableCommandProcessingAggregate<Account, AccountCommand> {
 
+  public Money getBalance() {
+    return balance;
+  }
+  public boolean accountLimitSufficient(Money money){
+    return balance.isGreaterThanOrEqual(money);
+  }
+
+  public void setBalance(Money balance) {
+    this.balance = balance;
+  }
+
+  private final Logger logger = LoggerFactory.getLogger(getClass());
+
+  private Money balance;
+
   public List<Event> process(CreateAccountCommand command) {
-    return events(new AccountCreatedEvent());
+    logger.info(command.getCustomerId().toString());
+    return events(new AccountCreatedEvent(command.getCustomerId(), command.getInitialBalance()));
   }
 
   public void apply(AccountCreatedEvent event) {
-
+    this.balance = event.getInitialBalance();
+    logger.info(this.balance.asString());
+    logger.info(event.getCustomerId().toString());
   }
 
+  public List<Event> process(CheckAccountLimitCommandInternal command) {
+    if(balance.isGreaterThanOrEqual(command.getMoney())){
+      return events(new AccountLimitSufficientEvent());
+    }
+    return events(new AccountLimitExceededEvent());
+  }
 
   public List<Event> process(AuthorizeCommandInternal command) {
-    return events(new AccountAuthorizedEvent());
+    return events(new AccountAuthorizedEvent(command.getOrderTotal()));
   }
 
   public List<Event> process(ReverseAuthorizationCommandInternal command) {
@@ -32,7 +61,15 @@ public List<Event> process(ReviseAuthorizationCommandInternal command) {
   }
 
   public void apply(AccountAuthorizedEvent event) {
+    if(balance.isGreaterThanOrEqual(event.getMoney())) {
+      setBalance(getBalance().subtract(event.getMoney()));
+    }
+  }
+
+  public void apply(AccountLimitSufficientEvent event) {
+  }
 
+  public void apply(AccountLimitExceededEvent event) {
   }
 
   public void apply(SagaReplyRequestedEvent event) {

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

@@ -1,6 +1,19 @@
 package net.chrisrichardson.ftgo.accountingservice.domain;
 
 import io.eventuate.Event;
+import net.chrisrichardson.ftgo.common.Money;
 
 public class AccountAuthorizedEvent implements Event {
+    private Money money;
+
+    public AccountAuthorizedEvent() {
+    }
+
+    public AccountAuthorizedEvent(Money money) {
+        this.money = money;
+    }
+
+    public Money getMoney() {
+        return money;
+    }
 }

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

@@ -1,6 +1,31 @@
 package net.chrisrichardson.ftgo.accountingservice.domain;
 
 import io.eventuate.Event;
+import net.chrisrichardson.ftgo.common.Money;
+
+import java.math.BigDecimal;
 
 public class AccountCreatedEvent implements Event {
+    private Money initialBalance;
+
+    private Long customerId;
+
+    private AccountCreatedEvent() {}
+
+    public AccountCreatedEvent(Long customerId, Money initialBalance){
+        this.customerId = customerId;
+        this.initialBalance = initialBalance;
+    }
+
+    public Money getInitialBalance() {
+        return initialBalance;
+    }
+
+    public Long getCustomerId() {
+        return customerId;
+    }
+
+    public void setCustomerId(Long customerId) {
+        this.customerId = customerId;
+    }
 }

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

@@ -0,0 +1,6 @@
+package net.chrisrichardson.ftgo.accountingservice.domain;
+
+import io.eventuate.Event;
+
+public class AccountLimitExceededEvent implements Event {
+}

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

@@ -0,0 +1,4 @@
+package net.chrisrichardson.ftgo.accountingservice.domain;
+
+public class AccountLimitExceededException extends RuntimeException {
+}

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

@@ -0,0 +1,6 @@
+package net.chrisrichardson.ftgo.accountingservice.domain;
+
+import io.eventuate.Event;
+
+public class AccountLimitSufficientEvent implements Event {
+}