[WIP] Async package delivery sample

Signed-off-by: Tihomir Surdilovic <tihomir@temporal.io>

Author: tsurdilo

core/src/main/java/io/temporal/samples/packetdelivery/Packet.java

@@ -0,0 +1,21 @@
+package io.temporal.samples.packetdelivery;
+
+public class Packet {
+  private int id;
+  private String content;
+
+  public Packet() {}
+
+  public Packet(int id, String content) {
+    this.id = id;
+    this.content = content;
+  }
+
+  public int getId() {
+    return id;
+  }
+
+  public String getContent() {
+    return content;
+  }
+}

core/src/main/java/io/temporal/samples/packetdelivery/PacketDelivery.java

@@ -0,0 +1,71 @@
+package io.temporal.samples.packetdelivery;
+
+import io.temporal.activity.ActivityOptions;
+import io.temporal.workflow.Async;
+import io.temporal.workflow.CompletablePromise;
+import io.temporal.workflow.Promise;
+import io.temporal.workflow.Workflow;
+import java.time.Duration;
+import org.slf4j.Logger;
+
+public class PacketDelivery {
+  private Packet packet;
+  private boolean deliveryConfirmation = false;
+  private CompletablePromise delivered = Workflow.newPromise();
+  private String deliveryConfirmationCode = "";
+
+  private Logger logger = Workflow.getLogger(this.getClass().getName());
+
+  private final PacketDeliveryActivities activities =
+      Workflow.newActivityStub(
+          PacketDeliveryActivities.class,
+          ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(3)).build());
+
+  public PacketDelivery(Packet packet) {
+    this.packet = packet;
+    processDeliveryAsync();
+  }
+
+  public Promise<Void> getDelivered() {
+    System.out.println("*************** CHECKING GET DELIVERED!!! : " + delivered);
+    return delivered;
+  }
+
+  public void processDeliveryAsync() {
+    delivered.completeFrom(Async.procedure(this::processDelivery));
+    System.out.println("*************** DONE WITH PROCESS DELIVERY ASYNC");
+  }
+
+  public void processDelivery() {
+    while (!deliveryConfirmationCode.equals("Confirmed")) {
+      // Step 1 perform delivery
+      logger.info(
+          "** Performing delivery for packet: " + packet.getId() + " - " + packet.getContent());
+      activities.performDelivery(packet);
+      // Step 2 wait for delivery confirmation
+      logger.info(
+          "** Delivery for packet: "
+              + packet.getId()
+              + " - "
+              + packet.getContent()
+              + " awaiting delivery confirmation");
+      Workflow.await(() -> deliveryConfirmation);
+      logger.info(
+          "** Delivery for packet: "
+              + packet.getId()
+              + " - "
+              + packet.getContent()
+              + " received confirmation");
+      // Step 3 complete delivery processing
+      logger.info(
+          "** Completing delivery for packet: " + packet.getId() + " - " + packet.getContent());
+      deliveryConfirmationCode = activities.completeDelivery(packet);
+      // Reset deliveryConfirmation
+      deliveryConfirmation = false;
+    }
+  }
+
+  public void confirmDelivery() {
+    this.deliveryConfirmation = true;
+  }
+}

core/src/main/java/io/temporal/samples/packetdelivery/PacketDeliveryActivities.java

@@ -0,0 +1,13 @@
+package io.temporal.samples.packetdelivery;
+
+import io.temporal.activity.ActivityInterface;
+import java.util.List;
+
+@ActivityInterface
+public interface PacketDeliveryActivities {
+  List<Packet> generatePackets();
+
+  void performDelivery(Packet packet);
+
+  String completeDelivery(Packet packet);
+}

core/src/main/java/io/temporal/samples/packetdelivery/PacketDeliveryActivitiesImpl.java

@@ -0,0 +1,47 @@
+package io.temporal.samples.packetdelivery;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class PacketDeliveryActivitiesImpl implements PacketDeliveryActivities {
+  @Override
+  public List<Packet> generatePackets() {
+    List<Packet> result = new ArrayList<>();
+    result.add(new Packet(1, "books"));
+    result.add(new Packet(2, "jewelry"));
+    result.add(new Packet(3, "furniture"));
+    result.add(new Packet(4, "food"));
+    result.add(new Packet(5, "electronics"));
+    return result;
+  }
+
+  @Override
+  public void performDelivery(Packet packet) {
+    System.out.println(
+        "** Activity - Performing delivery for packet: "
+            + packet.getId()
+            + " with content: "
+            + packet.getContent());
+    sleep(2);
+  }
+
+  @Override
+  public String completeDelivery(Packet packet) {
+    System.out.println(
+        "** Activity - Completing delivery for package: "
+            + packet.getId()
+            + " with content: "
+            + packet.getContent());
+    sleep(1);
+    // for sample we just confirm
+    return "Confirmed";
+  }
+
+  private void sleep(int seconds) {
+    try {
+      Thread.sleep(seconds * 1000L);
+    } catch (Exception e) {
+      System.out.println(e.getMessage());
+    }
+  }
+}

core/src/main/java/io/temporal/samples/packetdelivery/PacketDeliveryWorkflow.java

@@ -0,0 +1,14 @@
+package io.temporal.samples.packetdelivery;
+
+import io.temporal.workflow.SignalMethod;
+import io.temporal.workflow.WorkflowInterface;
+import io.temporal.workflow.WorkflowMethod;
+
+@WorkflowInterface
+public interface PacketDeliveryWorkflow {
+  @WorkflowMethod
+  String execute();
+
+  @SignalMethod
+  void confirmDelivery(int deliveryId);
+}

core/src/main/java/io/temporal/samples/packetdelivery/PacketDeliveryWorkflowImpl.java

@@ -0,0 +1,44 @@
+package io.temporal.samples.packetdelivery;
+
+import io.temporal.activity.ActivityOptions;
+import io.temporal.workflow.Promise;
+import io.temporal.workflow.Workflow;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class PacketDeliveryWorkflowImpl implements PacketDeliveryWorkflow {
+
+  private final Map<Integer, PacketDelivery> packetDeliveries = new HashMap<>();
+  //  private Logger logger = Workflow.getLogger(this.getClass().getName());
+
+  private final PacketDeliveryActivities activities =
+      Workflow.newActivityStub(
+          PacketDeliveryActivities.class,
+          ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(3)).build());
+
+  @Override
+  public String execute() {
+    List<Promise<Void>> packetsDelivered = new ArrayList<>();
+    // Step 1 - upload initial packets to deliver
+    List<Packet> initialPackets = activities.generatePackets();
+    // Step 2 - set up delivery processing
+    for (Packet packet : initialPackets) {
+      PacketDelivery delivery = new PacketDelivery(packet);
+      packetDeliveries.put(packet.getId(), delivery);
+      packetsDelivered.add(delivery.getDelivered());
+    }
+
+    Promise.allOf(packetsDelivered).get();
+    return "completed";
+  }
+
+  @Override
+  public void confirmDelivery(int deliveryId) {
+    if (packetDeliveries.containsKey(deliveryId)) {
+      packetDeliveries.get(deliveryId).confirmDelivery();
+    }
+  }
+}

core/src/main/java/io/temporal/samples/packetdelivery/Starter.java

@@ -0,0 +1,56 @@
+package io.temporal.samples.packetdelivery;
+
+import io.temporal.client.WorkflowClient;
+import io.temporal.client.WorkflowOptions;
+import io.temporal.client.WorkflowStub;
+import io.temporal.serviceclient.WorkflowServiceStubs;
+import io.temporal.worker.Worker;
+import io.temporal.worker.WorkerFactory;
+
+public class Starter {
+  public static void main(String[] args) {
+    WorkflowServiceStubs service = WorkflowServiceStubs.newLocalServiceStubs();
+    WorkflowClient client = WorkflowClient.newInstance(service);
+    WorkerFactory factory = WorkerFactory.newInstance(client);
+    Worker worker = factory.newWorker("packet-delivery-taskqueue");
+
+    worker.registerWorkflowImplementationTypes(PacketDeliveryWorkflowImpl.class);
+    worker.registerActivitiesImplementations(new PacketDeliveryActivitiesImpl());
+
+    factory.start();
+
+    PacketDeliveryWorkflow workflow =
+        client.newWorkflowStub(
+            PacketDeliveryWorkflow.class,
+            WorkflowOptions.newBuilder()
+                .setWorkflowId("packet-delivery-workflow")
+                .setTaskQueue("packet-delivery-taskqueue")
+                .build());
+
+    WorkflowClient.start(workflow::execute);
+
+    // start completing package deliveries (send confirmations)
+    sleep(3);
+    workflow.confirmDelivery(3); // furniture
+    sleep(1);
+    workflow.confirmDelivery(5); // electronics
+    sleep(1);
+    workflow.confirmDelivery(1); // books
+    sleep(1);
+    workflow.confirmDelivery(2); // jewelry
+    sleep(1);
+    workflow.confirmDelivery(4); // food
+
+    // wait for workflow to complete
+    String result = WorkflowStub.fromTyped(workflow).getResult(String.class);
+    System.out.println("** Workflow Result: " + result);
+  }
+
+  private static void sleep(int seconds) {
+    try {
+      Thread.sleep(seconds * 1000L);
+    } catch (Exception e) {
+      System.out.println(e.getMessage());
+    }
+  }
+}