feat: FEEL function for invoking outbound connectors

connector-runtime/connector-runtime-core/src/main/java/io/camunda/connector/runtime/core/external/ExternalOutboundConnectorContext.java

@@ -28,7 +28,7 @@ public ExternalOutboundConnectorContext(
       SecretProvider secretProvider,
       ValidationProvider validationProvider,
       ObjectMapper objectMapper,
-      String jsonVariables) {
+      Object jsonVariables) {
     super(secretProvider, validationProvider, objectMapper, jsonVariables);
   }
 

connector-runtime/connector-runtime-core/src/main/java/io/camunda/connector/runtime/core/outbound/AbstractOutboundConnectorContext.java

@@ -16,51 +16,91 @@
  */
 package io.camunda.connector.runtime.core.outbound;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.fasterxml.jackson.databind.node.TextNode;
 import io.camunda.connector.api.error.ConnectorException;
 import io.camunda.connector.api.outbound.OutboundConnectorContext;
 import io.camunda.connector.api.secret.SecretProvider;
 import io.camunda.connector.api.validation.ValidationProvider;
 import io.camunda.connector.runtime.core.AbstractConnectorContext;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
 
 public abstract class AbstractOutboundConnectorContext extends AbstractConnectorContext
     implements OutboundConnectorContext {
 
   protected final ObjectMapper objectMapper;
 
-  private final String jsonVariables;
+  private final Object variables;
+  private JsonNode variablesWithReplacedSecrets = null;
+
   private String jsonWithSecrets = null;
 
   public AbstractOutboundConnectorContext(
       final SecretProvider secretProvider,
       final ValidationProvider validationProvider,
       final ObjectMapper objectMapper,
-      String jsonVariables) {
+      Object variables) {
     super(secretProvider, validationProvider);
-    this.jsonVariables = jsonVariables;
+    this.variables = variables;
     this.objectMapper = objectMapper;
   }
 
   @Override
   public <T> T bindVariables(Class<T> cls) {
-    var mappedObject = mapJson(cls);
+    T mappedObject = getVariablesWithReplacedSecrets(cls);
     getValidationProvider().validate(mappedObject);
     return mappedObject;
   }
 
-  protected String getJsonReplacedWithSecrets() {
-    if (jsonWithSecrets == null) {
-      jsonWithSecrets = getSecretHandler().replaceSecrets(jsonVariables);
+  @SuppressWarnings("unchecked")
+  protected <T> T getVariablesWithReplacedSecrets(Class<T> cls) {
+    try {
+      if (variablesWithReplacedSecrets == null) {
+        variablesWithReplacedSecrets = getVariablesWithReplacedSecretsInternal();
+      }
+      if (cls == String.class) {
+        return (T) objectMapper.writeValueAsString(variablesWithReplacedSecrets);
+      }
+      return objectMapper.convertValue(variablesWithReplacedSecrets, cls);
+    } catch (JsonProcessingException e) {
+      throw new ConnectorException("JSON_MAPPING", "Error during json mapping.");
     }
-    return jsonWithSecrets;
   }
 
-  private <T> T mapJson(Class<T> cls) {
-    var jsonWithSecrets = getJsonReplacedWithSecrets();
-    try {
-      return objectMapper.readValue(jsonWithSecrets, cls);
-    } catch (Exception e) {
-      throw new ConnectorException("JSON_MAPPING", "Error during json mapping.");
+  protected JsonNode getVariablesWithReplacedSecretsInternal() throws JsonProcessingException {
+    if (variables instanceof String stringVars) {
+      String stringVarsWithSecrets = getSecretHandler().replaceSecrets(stringVars);
+      return objectMapper.readTree(stringVarsWithSecrets);
+    }
+    JsonNode convertedVars = objectMapper.valueToTree(variables);
+    return replaceSecretsViaReflection(convertedVars);
+  }
+
+  private JsonNode replaceSecretsViaReflection(JsonNode node) {
+    if (node.isTextual()) {
+      return new TextNode(getSecretHandler().replaceSecrets(node.asText()));
+    } else if (node.isObject()) {
+      Map<String, JsonNode> fields =
+          node.properties().stream()
+              .collect(
+                  Collectors.toMap(
+                      Map.Entry::getKey, e -> replaceSecretsViaReflection(e.getValue())));
+      return new ObjectNode(objectMapper.getNodeFactory(), fields);
+    } else if (node.isArray()) {
+      List<JsonNode> elements = new ArrayList<>();
+      for (JsonNode element : node) {
+        elements.add(replaceSecretsViaReflection(element));
+      }
+      return new ArrayNode(objectMapper.getNodeFactory(), elements);
+    } else {
+      return node;
     }
   }
 }

connector-runtime/connector-runtime-core/src/main/java/io/camunda/connector/runtime/core/outbound/JobHandlerContext.java

@@ -41,7 +41,8 @@ public JobHandlerContext(
       final ObjectMapper objectMapper) {
     super(secretProvider, validationProvider, objectMapper, job.getVariables());
     this.job = job;
-    this.jobContext = new ActivatedJobContext(job, this::getJsonReplacedWithSecrets);
+    this.jobContext =
+        new ActivatedJobContext(job, () -> getVariablesWithReplacedSecrets(String.class));
   }
 
   @Override

connector-runtime/connector-runtime-core/src/test/java/io/camunda/connector/runtime/core/external/ExternalOutboundConnectorContextTest.java

@@ -0,0 +1,101 @@
+/*
+ * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH
+ * under one or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information regarding copyright
+ * ownership. Camunda licenses this file to you under the Apache License,
+ * Version 2.0; you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.camunda.connector.runtime.core.external;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.when;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.camunda.connector.api.json.ConnectorsObjectMapperSupplier;
+import io.camunda.connector.api.secret.SecretProvider;
+import io.camunda.connector.api.validation.ValidationProvider;
+import java.util.List;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+public class ExternalOutboundConnectorContextTest {
+
+  @Mock private SecretProvider secretProvider;
+  private final ObjectMapper objectMapper = ConnectorsObjectMapperSupplier.DEFAULT_MAPPER;
+
+  @Mock private ValidationProvider validationProvider;
+
+  private record TestRecord(Integer key) {}
+
+  private record NestedTestRecord(TestRecord nested) {}
+
+  private record ArrayTestRecord(List<Integer> key) {}
+
+  @Test
+  void getVariablesAsTypeFromString() {
+    var context =
+        new ExternalOutboundConnectorContext(
+            secretProvider, validationProvider, objectMapper, "{\"key\": 3}");
+    TestRecord result = context.bindVariables(TestRecord.class);
+    assertThat(result.key()).isEqualTo(3);
+  }
+
+  @Test
+  void getVariablesAsTypeFromObject() {
+    var context =
+        new ExternalOutboundConnectorContext(
+            secretProvider, validationProvider, objectMapper, Map.of("key", 3));
+    TestRecord result = context.bindVariables(TestRecord.class);
+    assertThat(result.key()).isEqualTo(3);
+  }
+
+  @Test
+  void secretsAreReplaced() {
+    var context =
+        new ExternalOutboundConnectorContext(
+            secretProvider, validationProvider, objectMapper, Map.of("key", "{{secrets.KEY}}"));
+    when(secretProvider.getSecret("KEY")).thenReturn("5");
+    TestRecord result = context.bindVariables(TestRecord.class);
+    assertThat(result.key()).isEqualTo(5);
+  }
+
+  @Test
+  void nestedSecretsAreReplaced() {
+    var context =
+        new ExternalOutboundConnectorContext(
+            secretProvider,
+            validationProvider,
+            objectMapper,
+            Map.of("nested", Map.of("key", "{{secrets.KEY}}")));
+    when(secretProvider.getSecret("KEY")).thenReturn("5");
+    NestedTestRecord result = context.bindVariables(NestedTestRecord.class);
+    assertThat(result.nested().key()).isEqualTo(5);
+  }
+
+  @Test
+  void arraySecretsAreReplaced() {
+    var context =
+        new ExternalOutboundConnectorContext(
+            secretProvider,
+            validationProvider,
+            objectMapper,
+            Map.of("key", List.of("{{secrets.KEY}}")));
+
+    when(secretProvider.getSecret("KEY")).thenReturn("5");
+    ArrayTestRecord result = context.bindVariables(ArrayTestRecord.class);
+    assertThat(result.key()).containsExactly(5);
+  }
+}

connector-runtime/connector-runtime-core/src/test/java/io/camunda/connector/runtime/core/outbound/JobHandlerContextTest.java

@@ -16,11 +16,13 @@
  */
 package io.camunda.connector.runtime.core.outbound;
 
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import io.camunda.connector.api.json.ConnectorsObjectMapperSupplier;
 import io.camunda.connector.api.secret.SecretProvider;
 import io.camunda.connector.api.validation.ValidationProvider;
 import io.camunda.zeebe.client.api.response.ActivatedJob;
@@ -34,18 +36,21 @@ class JobHandlerContextTest {
 
   @Mock private ActivatedJob activatedJob;
   @Mock private SecretProvider secretProvider;
-  @Mock private ObjectMapper objectMapper;
+  private final ObjectMapper objectMapper = ConnectorsObjectMapperSupplier.DEFAULT_MAPPER;
 
   @Mock private ValidationProvider validationProvider;
 
+  private record TestRecord(Integer key) {}
+
+  private record NestedTestRecord(TestRecord nested) {}
+
   @Test
   void getVariablesAsType() throws JsonProcessingException {
-    when(activatedJob.getVariables()).thenReturn("");
+    when(activatedJob.getVariables()).thenReturn("{\"key\": 3}");
     JobHandlerContext jobHandlerContext =
         new JobHandlerContext(activatedJob, secretProvider, validationProvider, objectMapper);
-    Class<Integer> integerClass = Integer.class;
-    jobHandlerContext.bindVariables(integerClass);
-    verify(objectMapper).readValue("", Integer.class);
+    TestRecord result = jobHandlerContext.bindVariables(TestRecord.class);
+    assertThat(result.key()).isEqualTo(3);
   }
 
   @Test
@@ -56,4 +61,24 @@ void getVariables() {
     jobHandlerContext.getJobContext().getVariables();
     verify(activatedJob).getVariables();
   }
+
+  @Test
+  void secretsAreReplaced() {
+    when(activatedJob.getVariables()).thenReturn("{\"key\": \"{{secrets.KEY}}\"}");
+    when(secretProvider.getSecret("KEY")).thenReturn("5");
+    JobHandlerContext jobHandlerContext =
+        new JobHandlerContext(activatedJob, secretProvider, validationProvider, objectMapper);
+    TestRecord result = jobHandlerContext.bindVariables(TestRecord.class);
+    assertThat(result.key()).isEqualTo(5);
+  }
+
+  @Test
+  void nestedSecretsAreReplaced() {
+    when(activatedJob.getVariables()).thenReturn("{\"nested\": {\"key\": \"{{secrets.KEY}}\"}}");
+    when(secretProvider.getSecret("KEY")).thenReturn("5");
+    JobHandlerContext jobHandlerContext =
+        new JobHandlerContext(activatedJob, secretProvider, validationProvider, objectMapper);
+    NestedTestRecord result = jobHandlerContext.bindVariables(NestedTestRecord.class);
+    assertThat(result.nested().key()).isEqualTo(5);
+  }
 }

connector-runtime/connector-runtime-spring/src/main/java/io/camunda/connector/runtime/feel/ConnectorInvocationFeelFunctionProvider.java

@@ -16,13 +16,8 @@
  */
 package io.camunda.connector.runtime.feel;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import io.camunda.connector.api.secret.SecretProvider;
-import io.camunda.connector.api.validation.ValidationProvider;
 import io.camunda.connector.feel.FeelEngineWrapperUtil;
-import io.camunda.connector.runtime.core.external.ExternalOutboundConnectorContext;
-import io.camunda.connector.runtime.core.outbound.OutboundConnectorFactory;
+import io.camunda.connector.runtime.core.external.ExternalConnectorExecutor;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
@@ -35,24 +30,14 @@
 
 public class ConnectorInvocationFeelFunctionProvider extends JavaFunctionProvider {
 
-  private final ObjectMapper objectMapper;
-  private final OutboundConnectorFactory connectorFactory;
-  private final SecretProvider secretProvider;
-  private final ValidationProvider validationProvider;
+  private final ExternalConnectorExecutor connectorExecutor;
 
   private static final ValueMapper feelValueMapper = ValueMapper.defaultValueMapper();
 
   private JavaFunction function;
 
-  public ConnectorInvocationFeelFunctionProvider(
-      ObjectMapper objectMapper,
-      OutboundConnectorFactory connectorFactory,
-      SecretProvider secretProvider,
-      ValidationProvider validationProvider) {
-    this.objectMapper = objectMapper;
-    this.connectorFactory = connectorFactory;
-    this.secretProvider = secretProvider;
-    this.validationProvider = validationProvider;
+  public ConnectorInvocationFeelFunctionProvider(ExternalConnectorExecutor connectorExecutor) {
+    this.connectorExecutor = connectorExecutor;
     initFunction();
   }
 
@@ -67,23 +52,12 @@ private void initFunction() {
               final ValContext variables = (ValContext) args.get(1);
               Object unpackedVariables =
                   FeelEngineWrapperUtil.sanitizeScalaOutput(feelValueMapper.unpackVal(variables));
-              String variablesAsString;
-              try {
-                variablesAsString = objectMapper.writeValueAsString(unpackedVariables);
-              } catch (JsonProcessingException e) {
-                throw new RuntimeException(e);
-              }
-
-              var connector = connectorFactory.getInstance(typeValue);
-              var context =
-                  new ExternalOutboundConnectorContext(
-                      secretProvider, validationProvider, objectMapper, variablesAsString);
 
               Object result;
               try {
-                result = connector.execute(context);
+                result = connectorExecutor.execute(typeValue, unpackedVariables);
               } catch (Exception e) {
-                throw new RuntimeException("Failed to execute connector: " + e.getMessage(), e);
+                throw new RuntimeException(e);
               }
               return feelValueMapper.toVal(result);
             });

connector-runtime/connector-runtime-spring/src/test/java/io/camunda/connector/runtime/feel/ConnectorFeelFunctionTest.java

@@ -23,6 +23,7 @@
 import io.camunda.connector.feel.FeelEngineWrapper;
 import io.camunda.connector.http.base.model.HttpCommonResult;
 import io.camunda.connector.runtime.core.discovery.SPIConnectorDiscovery;
+import io.camunda.connector.runtime.core.external.ExternalConnectorExecutor;
 import io.camunda.connector.runtime.core.outbound.DefaultOutboundConnectorFactory;
 import io.camunda.connector.runtime.core.secret.SecretProviderAggregator;
 import io.camunda.connector.runtime.core.secret.SecretProviderDiscovery;
@@ -35,12 +36,14 @@ public class ConnectorFeelFunctionTest {
 
   private final ObjectMapper objectMapper = ConnectorsObjectMapperSupplier.DEFAULT_MAPPER;
 
-  private final FunctionProvider functionProvider =
-      new ConnectorInvocationFeelFunctionProvider(
-          objectMapper,
+  private final ExternalConnectorExecutor externalConnectorExecutor =
+      new ExternalConnectorExecutor(
           new DefaultOutboundConnectorFactory(SPIConnectorDiscovery.discoverOutbound()),
           new SecretProviderAggregator(SecretProviderDiscovery.discoverSecretProviders()),
-          new DefaultValidationProvider());
+          new DefaultValidationProvider(),
+          objectMapper);
+  private final FunctionProvider functionProvider =
+      new ConnectorInvocationFeelFunctionProvider(externalConnectorExecutor);
 
   private final FeelEngineWrapper feelEngineWrapper =
       new FeelEngineWrapper(List.of(functionProvider));