Added payment microservice alternative using Zeebe (related to #73)

kafka/java/payment-zeebe/Dockerfile

@@ -0,0 +1,5 @@
+FROM openjdk:8-jdk-alpine
+VOLUME /tmp
+ADD target/*.jar app.jar
+ENV JAVA_OPTS=""
+ENTRYPOINT exec java $JAVA_OPTS -jar /app.jar

kafka/java/payment-zeebe/pom.xml

@@ -0,0 +1,58 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<groupId>io.flowing.retail</groupId>
+	<artifactId>flowing-retail-kafka-payment-zeebe</artifactId>
+	<version>0.0.1-SNAPSHOT</version>
+	<properties>
+		<maven.compiler.release>8</maven.compiler.release>
+		<spring.boot.version>2.6.7</spring.boot.version>
+		<spring.kafka.version>2.8.5</spring.kafka.version>
+		<zeebe.version>8.0.4</zeebe.version>
+	</properties>
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-web</artifactId>
+			<version>${spring.boot.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>io.camunda</groupId>
+			<artifactId>spring-zeebe-starter</artifactId>
+			<version>${zeebe.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+			<version>${spring.kafka.version}</version>
+		</dependency>
+	</dependencies>
+
+    <build>
+    	<plugins>
+			 <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>exec-maven-plugin</artifactId>              
+                <configuration>                         
+    				  <mainClass>io.flowing.retail.payment.PaymentApplication</mainClass>
+                </configuration>
+             </plugin>      
+			 <plugin>
+	            <groupId>org.springframework.boot</groupId>
+	            <artifactId>spring-boot-maven-plugin</artifactId>
+	            <executions>
+			        <execution>
+			          <goals>
+			            <goal>repackage</goal>
+			          </goals>
+			        </execution>
+			    </executions>
+	        </plugin>
+			<plugin>
+			    <groupId>org.apache.maven.plugins</groupId>
+			    <artifactId>maven-compiler-plugin</artifactId>
+			    <version>3.8.1</version>
+			</plugin>
+    	</plugins>
+    </build>
+</project>

kafka/java/payment-zeebe/src/main/java/io/flowing/retail/payment/PaymentApplication.java

@@ -0,0 +1,24 @@
+package io.flowing.retail.payment;
+
+import io.camunda.zeebe.spring.client.EnableZeebeClient;
+import io.camunda.zeebe.spring.client.annotation.ZeebeDeployment;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
+import org.springframework.web.client.RestTemplate;
+
+@SpringBootApplication
+@EnableZeebeClient
+@ZeebeDeployment(resources = "classpath:payment.bpmn")
+public class PaymentApplication {
+
+  public static void main(String[] args) throws Exception {
+    SpringApplication.run(PaymentApplication.class, args);
+  }
+
+  @Bean
+  public RestTemplate createRestTemplate() {
+    RestTemplate restTemplate = new RestTemplate();
+    return restTemplate;
+  }
+}

kafka/java/payment-zeebe/src/main/java/io/flowing/retail/payment/flow/ChargeCreditCardAdapter.java

@@ -0,0 +1,69 @@
+package io.flowing.retail.payment.flow;
+
+import io.camunda.zeebe.spring.client.annotation.ZeebeVariable;
+import io.camunda.zeebe.spring.client.annotation.ZeebeWorker;
+import io.camunda.zeebe.spring.client.exception.ZeebeBpmnError;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+import org.springframework.web.client.RestTemplate;
+
+import java.util.Collections;
+import java.util.Map;
+
+@Component
+@ConfigurationProperties
+public class ChargeCreditCardAdapter {
+
+  @Autowired
+  private RestTemplate rest;
+
+  /**
+   * Of course you could use Eureka for this
+   */
+  private String stripeChargeUrl = "http://localhost:8099/charge";
+
+  @ZeebeWorker(type = "chargeCreditCard", autoComplete = true)
+  public Map<String, String> chargeCreditCard(@ZeebeVariable Integer remaimingAmount) throws Exception {
+    CreateChargeRequest request = new CreateChargeRequest();
+    request.amount = remaimingAmount;
+
+    CreateChargeResponse response = rest.postForObject( //
+        stripeChargeUrl, //
+        request, //
+        CreateChargeResponse.class);   
+
+    // TODO Add error scenarios to StripeFake and then raise "Error_CreditCardError" here
+    if (response.errorCode!=null) {
+      throw new ZeebeBpmnError("Error_PaymentError", response.errorCode);
+    }
+
+    return Collections.singletonMap("paymentTransactionId", response.transactionId);
+  }
+
+  public static class CreateChargeRequest {
+    public int amount;
+  }
+
+  public static class CreateChargeResponse {
+    public String transactionId;
+    public String errorCode;
+  }
+
+  public RestTemplate getRest() {
+    return rest;
+  }
+
+  public void setRest(RestTemplate rest) {
+    this.rest = rest;
+  }
+
+  public String getStripeChargeUrl() {
+    return stripeChargeUrl;
+  }
+
+  public void setStripeChargeUrl(String stripeChargeUrl) {
+    this.stripeChargeUrl = stripeChargeUrl;
+  }
+
+}

kafka/java/payment-zeebe/src/main/java/io/flowing/retail/payment/flow/EmitEventAdapter.java

@@ -0,0 +1,28 @@
+package io.flowing.retail.payment.flow;
+
+import io.camunda.zeebe.client.api.response.ActivatedJob;
+import io.camunda.zeebe.spring.client.annotation.ZeebeWorker;
+import io.flowing.retail.payment.messages.Message;
+import io.flowing.retail.payment.messages.MessageSender;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+@Component
+public class EmitEventAdapter {
+
+    @Autowired
+    private MessageSender messageSender;
+
+    @ZeebeWorker(type = "emitEvent", autoComplete = true)
+    public void emitEvent(ActivatedJob job) throws Exception {
+        String traceId = (String) job.getVariablesAsMap().get("traceId");
+        String eventName = (String) job.getCustomHeaders().get("eventName");
+
+        messageSender.send( //
+                new Message<String>( //
+                        eventName, //
+                        traceId, //
+                        null)); // no payload at the moment
+    }
+
+}

kafka/java/payment-zeebe/src/main/java/io/flowing/retail/payment/flow/NoopAdapter.java

@@ -0,0 +1,12 @@
+package io.flowing.retail.payment.flow;
+
+import io.camunda.zeebe.spring.client.annotation.ZeebeWorker;
+import org.springframework.stereotype.Component;
+
+@Component
+public class NoopAdapter {
+
+  @ZeebeWorker(type = "noop", autoComplete = true)
+  public void doNothing() {
+  }
+}

kafka/java/payment-zeebe/src/main/java/io/flowing/retail/payment/flow/PaymentReceivedAdapter.java

@@ -0,0 +1,31 @@
+package io.flowing.retail.payment.flow;
+
+import io.camunda.zeebe.client.api.response.ActivatedJob;
+import io.camunda.zeebe.spring.client.annotation.ZeebeWorker;
+import io.flowing.retail.payment.messages.Message;
+import io.flowing.retail.payment.messages.MessageSender;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+@Component
+public class PaymentReceivedAdapter {
+
+  @Autowired
+  private MessageSender messageSender;
+
+  @ZeebeWorker(type = "paymentReceived")
+  public void paymentReceived(ActivatedJob job) throws Exception {
+    String refId = (String) job.getVariablesAsMap().get("refId");
+    String correlationId = (String) job.getVariablesAsMap().get("correlationId");
+    String traceId = (String) job.getVariablesAsMap().get("traceId");
+
+    messageSender.send( //
+        new Message<PaymentReceivedEventPayload>( //
+            "PaymentReceivedEvent", //
+            traceId, //
+            new PaymentReceivedEventPayload() //
+                .setRefId(refId))
+    		.setCorrelationid(correlationId));
+  }
+
+}

kafka/java/payment-zeebe/src/main/java/io/flowing/retail/payment/flow/PaymentReceivedEventPayload.java

@@ -0,0 +1,14 @@
+package io.flowing.retail.payment.flow;
+
+public class PaymentReceivedEventPayload {
+  private String refId;
+
+  public String getRefId() {
+    return refId;
+  }
+
+  public PaymentReceivedEventPayload setRefId(String refId) {
+    this.refId = refId;
+    return this;
+  }
+}

kafka/java/payment-zeebe/src/main/java/io/flowing/retail/payment/messages/Message.java

@@ -0,0 +1,108 @@
+package io.flowing.retail.payment.messages;
+
+import java.util.Date;
+import java.util.UUID;
+
+import org.springframework.beans.factory.annotation.Value;
+
+public class Message<T> {
+
+  // Cloud Events compliant 
+  private String type;
+  private String id = UUID.randomUUID().toString(); // unique id of this message
+  private String source="Payment Zeebe";
+  private Date time = new Date();
+  private T data;
+  private String datacontenttype="application/json";
+  private String specversion="1.0";
+
+  // Extension attributes
+  private String traceid = UUID.randomUUID().toString(); // trace id, default: new unique
+  private String correlationid; // id which can be used for correlation later if required
+  private String group = "flowing-retail";
+
+  public Message() {    
+  }
+
+  public Message(String type, T payload) {
+    this.type = type;
+    this.data = payload;
+  }
+
+  public Message(String type, String traceid, T payload) {
+    this.type = type;
+    this.traceid = traceid;
+    this.data = payload;
+  }
+
+  @Override
+  public String toString() {
+    return "Message [type=" + type + ", id=" + id + ", time=" + time + ", data=" + data + ", correlationid=" + correlationid + ", traceid=" + traceid + "]";
+  }
+
+  public String getType() {
+    return type;
+  }
+
+  public void setType(String type) {
+    this.type = type;
+  }
+
+  public String getId() {
+    return id;
+  }
+
+  public void setId(String id) {
+    this.id = id;
+  }
+
+  public Date getTime() {
+    return time;
+  }
+
+  public void setTime(Date time) {
+    this.time = time;
+  }
+
+  public T getData() {
+    return data;
+  }
+
+  public void setData(T data) {
+    this.data = data;
+  }
+
+  public String getTraceid() {
+    return traceid;
+  }
+
+  public void setTraceid(String traceid) {
+    this.traceid = traceid;
+  }
+
+  public String getCorrelationid() {
+    return correlationid;
+  }
+
+  public Message<T> setCorrelationid(String correlationid) {
+    this.correlationid = correlationid;
+    return this;
+  }
+
+  public String getSource() {
+    return source;
+  }
+
+  public String getDatacontenttype() {
+    return datacontenttype;
+  }
+
+  public String getSpecversion() {
+    return specversion;
+  }
+
+  public String getGroup() {
+    return group;
+  }
+
+}