Adding support for instances object (#1028)

Some callable tasks might require to share internal implementation
resources per workflow instance (for example a Python embedded
interpreter or a l4cj agenticscope instance)

Signed-off-by: fjtirado <ftirados@redhat.com>

Author: fjtirado

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java

@@ -34,6 +34,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Supplier;
 
 public class WorkflowMutableInstance implements WorkflowInstance {
 
@@ -47,6 +48,8 @@ public class WorkflowMutableInstance implements WorkflowInstance {
   protected AtomicReference<CompletableFuture<WorkflowModel>> futureRef = new AtomicReference<>();
   protected Instant completedAt;
 
+  protected final Map<String, Object> additionalObjects = new ConcurrentHashMap<String, Object>();
+
   private Lock statusLock = new ReentrantLock();
   private Map<CompletableFuture<TaskContext>, TaskContext> suspended;
 
@@ -84,14 +87,18 @@ protected final CompletableFuture<WorkflowModel> startExecution(Runnable runnabl
                     .inputFilter()
                     .map(f -> f.apply(workflowContext, null, input))
                     .orElse(input))
-            .whenComplete(this::whenFailed)
+            .whenComplete(this::whenCompleted)
             .thenApply(this::whenSuccess);
     futureRef.set(future);
     return future;
   }
 
-  private void whenFailed(WorkflowModel result, Throwable ex) {
+  private void whenCompleted(WorkflowModel result, Throwable ex) {
     completedAt = Instant.now();
+    additionalObjects.values().stream()
+        .filter(AutoCloseable.class::isInstance)
+        .map(AutoCloseable.class::cast)
+        .forEach(WorkflowUtils::safeClose);
     if (ex != null) {
       handleException(ex instanceof CompletionException ? ex = ex.getCause() : ex);
     }
@@ -278,5 +285,9 @@ public boolean cancel() {
     }
   }
 
+  public <T> T additionalObject(String key, Supplier<T> supplier) {
+    return (T) additionalObjects.computeIfAbsent(key, k -> supplier.get());
+  }
+
   public void restoreContext(WorkflowContext workflow, TaskContext context) {}
 }