Created Azure functions node

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/credentials/ClientCredentials.java

@@ -21,6 +21,7 @@
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
 import org.thingsboard.rule.engine.mqtt.azure.AzureIotHubSasCredentials;
+import org.thingsboard.rule.engine.rest.AzureFunctionsCredentials;
 
 import javax.net.ssl.SSLException;
 
@@ -29,7 +30,8 @@
         @JsonSubTypes.Type(value = AnonymousCredentials.class, name = "anonymous"),
         @JsonSubTypes.Type(value = BasicCredentials.class, name = "basic"),
         @JsonSubTypes.Type(value = AzureIotHubSasCredentials.class, name = "sas"),
-        @JsonSubTypes.Type(value = CertPemCredentials.class, name = "cert.PEM")})
+        @JsonSubTypes.Type(value = CertPemCredentials.class, name = "cert.PEM"),
+        @JsonSubTypes.Type(value = AzureFunctionsCredentials.class, name = "access.key")})
 public interface ClientCredentials {
     @JsonIgnore
     CredentialsType getType();

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/credentials/CredentialsType.java

@@ -19,7 +19,8 @@ public enum CredentialsType {
     ANONYMOUS("anonymous"),
     BASIC("basic"),
     SAS("sas"),
-    CERT_PEM("cert.PEM");
+    CERT_PEM("cert.PEM"),
+    ACCESS_KEY("access.key"),;
 
     private final String label;
 

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/AzureFunctionsCredentials.java

@@ -0,0 +1,34 @@
+/**
+ * Copyright © 2016-2024 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * 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 org.thingsboard.rule.engine.rest;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.rule.engine.credentials.ClientCredentials;
+import org.thingsboard.rule.engine.credentials.CredentialsType;
+
+@Data
+@Slf4j
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class AzureFunctionsCredentials implements ClientCredentials {
+    private String accessKey;
+
+    @Override
+    public CredentialsType getType() {
+        return CredentialsType.ACCESS_KEY;
+    }
+}

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbAzureFunctionsNode.java

@@ -0,0 +1,123 @@
+/**
+ * Copyright © 2016-2024 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * 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 org.thingsboard.rule.engine.rest;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.web.util.UriBuilder;
+import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.rule.engine.api.RuleNode;
+import org.thingsboard.rule.engine.api.TbContext;
+import org.thingsboard.rule.engine.api.TbNodeConfiguration;
+import org.thingsboard.rule.engine.api.TbNodeException;
+import org.thingsboard.rule.engine.api.util.TbNodeUtils;
+import org.thingsboard.rule.engine.credentials.CredentialsType;
+import org.thingsboard.rule.engine.external.TbAbstractExternalNode;
+import org.thingsboard.server.common.data.plugin.ComponentClusteringMode;
+import org.thingsboard.server.common.data.plugin.ComponentType;
+import org.thingsboard.server.common.msg.TbMsg;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+
+@Slf4j
+@RuleNode(
+        type = ComponentType.EXTERNAL,
+        name = "azure functions",
+        configClazz = TbAzureFunctionsNodeConfiguration.class,
+        clusteringMode = ComponentClusteringMode.SINGLETON,
+        nodeDescription = "Pushes message data to the Azure Functions",
+        nodeDetails = "Will invoke REST API call GET | POST | PUT | DELETE to Azure Functions. " +
+                "Message payload added into Request body. Query parameters can be added to the url. " +
+                "Configured attributes can be added into Headers from Message Metadata. " +
+                "Outbound message will contain response fields (status, statusCode, statusReason and response headers) in the Message Metadata. " +
+                "Response body saved in outbound Message payload. For example statusCode field can be accessed with metadata.statusCode.</b>.",
+        uiResources = {""},
+        configDirective = ""
+)
+public class TbAzureFunctionsNode extends TbAbstractExternalNode {
+
+    private TbHttpClient httpClient;
+    private TbAzureFunctionsNodeConfiguration config;
+
+    @Override
+    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
+        super.init(ctx);
+        config = TbNodeUtils.convert(configuration, TbAzureFunctionsNodeConfiguration.class);
+        httpClient = new TbHttpClient(config, ctx.getSharedEventLoop());
+    }
+
+    @Override
+    public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException {
+        var tbMsg = ackIfNeeded(ctx, msg);
+        tbMsg = TbMsg.transformMsgData(tbMsg, getRequestBody(tbMsg));
+        config.setRestEndpointUrlPattern(buildUrl(msg));
+        httpClient.processMessage(ctx, tbMsg,
+                m -> tellSuccess(ctx, m),
+                (m, t) -> tellFailure(ctx, m, t));
+    }
+
+    @Override
+    public void destroy() {
+        if (httpClient != null) {
+            httpClient.destroy();
+        }
+    }
+
+    private String buildUrl(TbMsg msg) {
+        StringBuilder urlBuilder = new StringBuilder(config.getRestEndpointUrlPattern());
+        if (CredentialsType.ACCESS_KEY == config.getCredentials().getType()
+                && !config.getRestEndpointUrlPattern().contains("?code=")) {
+            urlBuilder.append("?code=").append(((AzureFunctionsCredentials) config.getCredentials()).getAccessKey());
+        }
+        Map<String, String> queryParams = processMappings(msg, config.getQueryParams());
+        queryParams.forEach((param, value) -> {
+            if (urlBuilder.toString().contains("?")) {
+                urlBuilder.append("&");
+            } else {
+                urlBuilder.append("?");
+            }
+            urlBuilder.append(param).append("=").append(value);
+        });
+        return urlBuilder.toString();
+    }
+
+    private String getRequestBody(TbMsg msg) {
+        ObjectNode requestBodyJson = JacksonUtil.newObjectNode();
+        Map<String, String> inputKeys = processMappings(msg, config.getInputKeys());
+        inputKeys.forEach(requestBodyJson::put);
+        return JacksonUtil.toString(requestBodyJson);
+    }
+
+    private Map<String, String> processMappings(TbMsg msg, Map<String, String> mappings) {
+        Map<String, String> processedMappings = new HashMap<>();
+        JsonNode msgData = JacksonUtil.toJsonNode(msg.getData());
+        mappings.forEach((funcKey, msgKey) -> {
+            String patternProcessedFuncKey = TbNodeUtils.processPattern(funcKey, msg);
+            String patternProcessedMsgValue;
+            try {
+                String patternProcessedMsgKey = TbNodeUtils.processPattern(msgKey, msg);
+                patternProcessedMsgValue = msgData.get(patternProcessedMsgKey).asText();
+            } catch (Exception e) {
+                patternProcessedMsgValue = msgKey;
+            }
+            processedMappings.put(patternProcessedFuncKey, patternProcessedMsgValue);
+        });
+        return processedMappings;
+    }
+}

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbAzureFunctionsNodeConfiguration.java

@@ -0,0 +1,43 @@
+/**
+ * Copyright © 2016-2024 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * 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 org.thingsboard.rule.engine.rest;
+
+import lombok.Data;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.thingsboard.rule.engine.credentials.AnonymousCredentials;
+
+import java.util.Collections;
+import java.util.Map;
+
+@Data
+public class TbAzureFunctionsNodeConfiguration extends TbRestApiCallNodeConfiguration {
+
+    private Map<String, String> queryParams;
+    private Map<String, String> inputKeys;
+
+    @Override
+    public TbAzureFunctionsNodeConfiguration defaultConfiguration() {
+        TbAzureFunctionsNodeConfiguration configuration = new TbAzureFunctionsNodeConfiguration();
+        configuration.setRestEndpointUrlPattern("http://localhost:<port>/api/<function-name>");
+        configuration.setRequestMethod("POST");
+        configuration.setHeaders(Collections.singletonMap(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE));
+        configuration.setQueryParams(Collections.emptyMap());
+        configuration.setInputKeys(Collections.emptyMap());
+        configuration.setCredentials(new AnonymousCredentials());
+        return configuration;
+    }
+}

rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/rest/TbAzureFunctionsNodeTest.java

@@ -0,0 +1,132 @@
+ /**
+ * Copyright © 2016-2024 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * 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 org.thingsboard.rule.engine.rest;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.rule.engine.api.TbContext;
+import org.thingsboard.rule.engine.api.TbNodeConfiguration;
+import org.thingsboard.rule.engine.api.TbNodeException;
+import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.common.data.msg.TbMsgType;
+import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.TbMsgMetaData;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatNoException;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.verify;
+
+@ExtendWith(MockitoExtension.class)
+public class TbAzureFunctionsNodeTest {
+
+    private TbAzureFunctionsNode node;
+    private TbAzureFunctionsNodeConfiguration config;
+
+    @Mock
+    private TbContext ctxMock;
+    @Mock
+    private TbHttpClient clientMock;
+
+    @BeforeEach
+    public void setUp() {
+        node = new TbAzureFunctionsNode();
+        config = new TbAzureFunctionsNodeConfiguration().defaultConfiguration();
+        ReflectionTestUtils.setField(node, "httpClient", clientMock);
+        ReflectionTestUtils.setField(node, "config", config);
+    }
+
+    @Test
+    public void givenDefaultConfig_whenInit_thenOk() {
+        var configuration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
+        assertThatNoException().isThrownBy(() -> node.init(ctxMock, configuration));
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"${msgKeyTemplate}", "msgKey", "10"})
+    public void givenInputKeys_whenOnMsg_thenTellSuccess(String msgKey) throws TbNodeException, ExecutionException, InterruptedException {
+        config.setInputKeys(Map.of("${funcKeyTemplate}", msgKey));
+
+        String data = """
+                {
+                "msgKey": 10
+                }
+                """;
+        Map<String, String> metadata = new HashMap<>();
+        metadata.put("funcKeyTemplate", "funcKey");
+        metadata.put("msgKeyTemplate", "msgKey");
+        DeviceId deviceId = new DeviceId(UUID.fromString("b8956804-147b-4595-a50d-19fca6dd0c7b"));
+        TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, deviceId, new TbMsgMetaData(metadata), data);
+        String processedData = "{\"funcKey\":\"10\"}";
+        TbMsg transformedMsg = TbMsg.transformMsgData(msg, processedData);
+        doAnswer(invocation -> {
+            Consumer<TbMsg> consumer = invocation.getArgument(2);
+            consumer.accept(transformedMsg);
+            return null;
+        }).when(clientMock).processMessage(any(), any(), any(), any());
+
+        node.onMsg(ctxMock, msg);
+
+        ArgumentCaptor<TbMsg> msgCaptor = ArgumentCaptor.forClass(TbMsg.class);
+        verify(clientMock).processMessage(eq(ctxMock), msgCaptor.capture(), any(), any());
+        TbMsg msgCaptorValue = msgCaptor.getValue();
+        assertThat(msgCaptorValue.getData()).isEqualTo(transformedMsg.getData());
+        verify(ctxMock).tellSuccess(eq(transformedMsg));
+    }
+
+    @Test
+    public void givenInvalidAccessToken_whenOnMsg_thenTellFailure() throws TbNodeException, ExecutionException, InterruptedException {
+        DeviceId deviceId = new DeviceId(UUID.fromString("b8956804-147b-4595-a50d-19fca6dd0c7b"));
+        Map<String, String> metadata = new HashMap<>();
+        TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, deviceId, new TbMsgMetaData(metadata), TbMsg.EMPTY_JSON_OBJECT);
+        Exception thrownException = new Exception("Unauthorized");
+
+        doAnswer(invocation -> {
+            metadata.put("error", thrownException.getClass() + ": " + thrownException.getMessage());
+            TbMsg transformedMsg = TbMsg.transformMsgMetadata(msg, new TbMsgMetaData(metadata));
+            BiConsumer<TbMsg, Exception> consumer = invocation.getArgument(3);
+            consumer.accept(transformedMsg, thrownException);
+            return null;
+        }).when(clientMock).processMessage(any(), any(), any(), any());
+
+        node.onMsg(ctxMock, msg);
+
+        ArgumentCaptor<TbMsg> msgCaptor = ArgumentCaptor.forClass(TbMsg.class);
+        ArgumentCaptor<Exception> exceptionCaptor = ArgumentCaptor.forClass(Exception.class);
+        verify(ctxMock).tellFailure(msgCaptor.capture(), exceptionCaptor.capture());
+        TbMsg msgCaptorValue = msgCaptor.getValue();
+        assertThat(msgCaptorValue.getMetaData().getData()).isEqualTo(metadata);
+        Exception exceptionCaptorValue = exceptionCaptor.getValue();
+        assertThat(exceptionCaptorValue.getMessage()).isEqualTo(thrownException.getMessage());
+    }
+}