update

Signed-off-by: Tihomir Surdilovic <[email protected]>

Author: tsurdilo

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

@@ -10,6 +10,7 @@
 public class PacketDelivery {
   private Packet packet;
   private boolean deliveryConfirmation = false;
+  private boolean needDeliveryConfirmation = false;
   private CompletablePromise delivered = Workflow.newPromise();
   private CancellationScope cancellationScope;
 
@@ -61,6 +62,7 @@ public void processDelivery() {
                         + " - "
                         + packet.getContent()
                         + " awaiting delivery confirmation");
+                needDeliveryConfirmation = true;
                 Workflow.await(() -> deliveryConfirmation);
                 logger.info(
                     "** Delivery for packet: "
@@ -75,8 +77,9 @@ public void processDelivery() {
                         + " - "
                         + packet.getContent());
                 deliveryConfirmationResult = activities.completeDelivery(packet);
-                // Reset deliveryConfirmation
+                // Reset deliveryConfirmation and needDeliveryConfirmation
                 deliveryConfirmation = false;
+                needDeliveryConfirmation = false;
               }
             });
 
@@ -93,6 +96,7 @@ public void processDelivery() {
       // Just for show for example that cancel could come in while we are waiting on approval signal
       // too
       else if (e instanceof CanceledFailure) {
+        needDeliveryConfirmation = false;
         // Run compensation activity and complete
         compensationActivities.compensateDelivery(packet);
       }
@@ -109,4 +113,12 @@ public void cancelDelivery(String reason) {
       cancellationScope.cancel(reason);
     }
   }
+
+  public boolean isNeedDeliveryConfirmation() {
+    return needDeliveryConfirmation;
+  }
+
+  public Packet getPacket() {
+    return packet;
+  }
 }

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

@@ -1,8 +1,10 @@
 package io.temporal.samples.packetdelivery;
 
+import io.temporal.workflow.QueryMethod;
 import io.temporal.workflow.SignalMethod;
 import io.temporal.workflow.WorkflowInterface;
 import io.temporal.workflow.WorkflowMethod;
+import java.util.List;
 
 @WorkflowInterface
 public interface PacketDeliveryWorkflow {
@@ -14,4 +16,7 @@ public interface PacketDeliveryWorkflow {
 
   @SignalMethod
   void cancelDelivery(int deliveryId, String reason);
+
+  @QueryMethod
+  List<Packet> deliveryConfirmationPackets();
 }

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

@@ -8,10 +8,11 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import org.slf4j.Logger;
 
 public class PacketDeliveryWorkflowImpl implements PacketDeliveryWorkflow {
-
   private final Map<Integer, PacketDelivery> packetDeliveries = new HashMap<>();
+  private final Logger logger = Workflow.getLogger(this.getClass().getName());
 
   private final PacketDeliveryActivities activities =
       Workflow.newActivityStub(
@@ -47,7 +48,31 @@ public void confirmDelivery(int deliveryId) {
   @Override
   public void cancelDelivery(int deliveryId, String reason) {
     if (packetDeliveries.containsKey(deliveryId)) {
-      packetDeliveries.get(deliveryId).cancelDelivery(reason);
+      // Only makes sense to cancel if delivery is not done yet
+      if (!packetDeliveries.get(deliveryId).getDelivered().isCompleted()) {
+        logger.info("Sending cancellation for delivery : " + deliveryId + " and reason: " + reason);
+        packetDeliveries.get(deliveryId).cancelDelivery(reason);
+      }
+      logger.info(
+          "Bypassing sending cancellation for delivery : "
+              + deliveryId
+              + " and reason: "
+              + reason
+              + " because delivery already completed");
     }
   }
+
+  @Override
+  public List<Packet> deliveryConfirmationPackets() {
+    List<Packet> confirmationPackets = new ArrayList<>();
+    packetDeliveries
+        .values()
+        .forEach(
+            p -> {
+              if (p.isNeedDeliveryConfirmation()) {
+                confirmationPackets.add(p.getPacket());
+              }
+            });
+    return confirmationPackets;
+  }
 }

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

@@ -1,11 +1,13 @@
 package io.temporal.samples.packetdelivery;
 
 import io.temporal.client.WorkflowClient;
+import io.temporal.client.WorkflowNotFoundException;
 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;
+import java.util.List;
 
 public class Starter {
   public static void main(String[] args) {
@@ -30,16 +32,25 @@ public static void main(String[] args) {
     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
+    // Query workflow for packets that need confirmation, confirm until none need confirmation any
+    // more
+    while (true) {
+      sleep(3);
+      List<Packet> packets = workflow.deliveryConfirmationPackets();
+      if (packets.isEmpty()) {
+        break;
+      }
+
+      for (Packet p : packets) {
+        try {
+          workflow.confirmDelivery(p.getId());
+        } catch (WorkflowNotFoundException e) {
+          // In some cases with cancellations happening, workflow could be completed by now
+          // We just ignore and exit out of loop
+          break;
+        }
+      }
+    }
 
     // wait for workflow to complete
     String result = WorkflowStub.fromTyped(workflow).getResult(String.class);