This is an automated email from the ASF dual-hosted git repository.
dominikriemer pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new 6b0c0bdfe5 #4683 fix(msteams): migrate sink to Power Automate workflow
format (#4684)
6b0c0bdfe5 is described below
commit 6b0c0bdfe5db494a47f2c360a35442c98da22931
Author: Stefan Obermeier <[email protected]>
AuthorDate: Thu Jul 16 21:38:29 2026 +0200
#4683 fix(msteams): migrate sink to Power Automate workflow format (#4684)
Co-authored-by: obermeier <[email protected]>
Co-authored-by: Dominik Riemer <[email protected]>
---
.../notifications/jvm/msteams/MSTeamsSink.java | 518 +++++++++++----------
.../documentation.md | 21 +-
.../notifications/jvm/msteams/TestMSTeamsSink.java | 230 +++++----
3 files changed, 428 insertions(+), 341 deletions(-)
diff --git
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/msteams/MSTeamsSink.java
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/msteams/MSTeamsSink.java
index e62cf725e8..86e0469da6 100644
---
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/msteams/MSTeamsSink.java
+++
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/msteams/MSTeamsSink.java
@@ -40,285 +40,309 @@ import
org.apache.streampipes.wrapper.standalone.StreamPipesNotificationSink;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.http.Header;
import org.apache.http.HttpHost;
+import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.ContentType;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;
-import org.apache.http.impl.client.HttpClients;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.IOException;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.time.Duration;
+import java.net.URI;
+import java.net.URISyntaxException;
import java.util.Map;
public class MSTeamsSink extends StreamPipesNotificationSink {
- public static final String ID =
"org.apache.streampipes.sinks.notifications.jvm.msteams";
-
- private static final String KEY_MESSAGE_ADVANCED = "messageAdvanced";
- private static final String KEY_MESSAGE_ADVANCED_CONTENT =
"messageContentAdvanced";
- private static final String KEY_MESSAGE_SIMPLE = "messageSimple";
- private static final String KEY_MESSAGE_SIMPLE_CONTENT =
"messageContentSimple";
- private static final String KEY_MESSAGE_TYPE_ALTERNATIVES = "messageType";
- private static final String KEY_WEBHOOK_URL = "webhookUrl";
- public static final String KEY_PROXY_ALTERNATIVES = "proxy";
- public static final String KEY_PROXY_DISABLED = "proxyDisabled";
- public static final String KEY_PROXY_ENABLED = "proxyEnabled";
- public static final String KEY_PROXY_GROUP = "proxyConfigurationGroup";
- public static final String KEY_PROXY_URL = "proxyUrl";
- protected static final String SIMPLE_MESSAGE_TEMPLATE = "{\"text\": \"%s\"}";
- private static final int MAX_RETRIES = 5;
- private static final int HTTP_TOO_MANY_REQUESTS = 429;
- private static final Duration BASE_BACKOFF = Duration.ofSeconds(1);
-
- private String messageContent;
- private boolean isSimpleMessageMode;
- private String webhookUrl;
- private ObjectMapper objectMapper;
- private CloseableHttpClient httpClient;
-
- public MSTeamsSink() {
- super();
- this.objectMapper = JacksonSerializer.getObjectMapper(Map.of(
- DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, true
- ));
- }
-
- @Override
- public IDataSinkConfiguration declareConfig() {
- var builder = declareModelWithoutSilentPeriod();
- addSilentPeriodParameter(builder);
-
- return DataSinkConfiguration.create(
- MSTeamsSink::new,
- builder.build()
- );
- }
-
- @Override
- public void onPipelineStarted(
- IDataSinkParameters parameters,
- EventSinkRuntimeContext runtimeContext
- ) {
- super.onPipelineStarted(parameters, runtimeContext);
-
- this.objectMapper = JacksonSerializer.getObjectMapper(Map.of(
- DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, true
- ));
-
- var extractor = parameters.extractor();
- webhookUrl = extractor.secretValue(KEY_WEBHOOK_URL);
-
- validateWebhookUrl(webhookUrl);
-
- var selectedAlternative =
extractor.selectedAlternativeInternalId(KEY_MESSAGE_TYPE_ALTERNATIVES);
- if (selectedAlternative.equals(KEY_MESSAGE_ADVANCED)) {
- isSimpleMessageMode = false;
- messageContent =
extractor.singleValueParameter(KEY_MESSAGE_ADVANCED_CONTENT, String.class);
- } else {
- isSimpleMessageMode = true;
- messageContent =
extractor.singleValueParameter(KEY_MESSAGE_SIMPLE_CONTENT, String.class);
+ public static final String ID =
"org.apache.streampipes.sinks.notifications.jvm.msteams";
+
+ private static final String KEY_MESSAGE_ADVANCED = "messageAdvanced";
+ private static final String KEY_MESSAGE_ADVANCED_CONTENT =
"messageContentAdvanced";
+ private static final String KEY_MESSAGE_SIMPLE = "messageSimple";
+ private static final String KEY_MESSAGE_SIMPLE_CONTENT =
"messageContentSimple";
+ private static final String KEY_MESSAGE_TYPE_ALTERNATIVES = "messageType";
+ private static final String KEY_WEBHOOK_URL = "webhookUrl";
+ public static final String KEY_PROXY_ALTERNATIVES = "proxy";
+ public static final String KEY_PROXY_DISABLED = "proxyDisabled";
+ public static final String KEY_PROXY_ENABLED = "proxyEnabled";
+ public static final String KEY_PROXY_GROUP = "proxyConfigurationGroup";
+ public static final String KEY_PROXY_URL = "proxyUrl";
+
+ private static final int MAX_ATTEMPTS = 3;
+ private static final long RETRY_DELAY_MS = 2000;
+ private static final int CONNECT_TIMEOUT_MS = 5_000;
+ private static final int SOCKET_TIMEOUT_MS = 10_000;
+ private static final int CONNECTION_REQUEST_TIMEOUT_MS = 5_000;
+
+ private static final Logger LOG =
LoggerFactory.getLogger(MSTeamsSink.class);
+
+ private String messageContent;
+ private boolean isSimpleMessageMode;
+ private URI webhookUrl;
+ private ObjectMapper objectMapper;
+ private CloseableHttpClient httpClient;
+
+ public MSTeamsSink() {
+ super();
+ this.objectMapper = JacksonSerializer
+
.getObjectMapper(Map.of(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES,
true));
}
- var selectedProxyAlternative =
extractor.selectedAlternativeInternalId(KEY_PROXY_ALTERNATIVES);
- if (selectedProxyAlternative.equals(KEY_PROXY_DISABLED)) {
- this.httpClient = HttpClients.createDefault();
- } else {
- var proxyUrl = extractor.singleValueParameter(KEY_PROXY_URL,
String.class);
- this.httpClient = HttpClientBuilder
- .create()
- .setProxy(HttpHost.create(proxyUrl))
- .build();
+ @Override
+ public IDataSinkConfiguration declareConfig() {
+ var builder = declareModelWithoutSilentPeriod();
+ return DataSinkConfiguration.create(MSTeamsSink::new, builder.build());
}
- }
- @Override
- public void onNotificationEvent(Event event) {
+ @Override
+ public void onPipelineStarted(IDataSinkParameters parameters,
EventSinkRuntimeContext runtimeContext) {
+ super.onPipelineStarted(parameters, runtimeContext);
- // This sink allows to use placeholders for event properties when defining
the message content in the UI
- // Therefore, we need to replace these placeholders based on the actual
event before actually sending the message
- var processedMessageContent =
PlaceholderExtractor.replacePlaceholders(event, messageContent);
+ this.objectMapper = JacksonSerializer
+
.getObjectMapper(Map.of(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES,
true));
- String teamsMessageContent;
- if (isSimpleMessageMode) {
- teamsMessageContent =
createMessageFromSimpleContent(processedMessageContent);
- } else {
- teamsMessageContent =
createMessageFromAdvancedContent(processedMessageContent);
- }
- sendPayloadToWebhook(httpClient, teamsMessageContent, webhookUrl);
- }
-
- @Override
- public DataSinkBuilder declareModelWithoutSilentPeriod() {
- return DataSinkBuilder
- .create(ID, 1)
- .withLocales(Locales.EN)
- .withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON)
- .category(DataSinkType.NOTIFICATION)
- .requiredStream(
- StreamRequirementsBuilder
- .create()
- .requiredProperty(EpRequirements.anyProperty())
- .build()
- )
- .requiredSecret(Labels.withId(KEY_WEBHOOK_URL))
- .requiredAlternatives(
- Labels.withId(KEY_PROXY_ALTERNATIVES),
- Alternatives.from(Labels.withId(KEY_PROXY_DISABLED)),
- Alternatives.from(Labels.withId(KEY_PROXY_ENABLED),
- StaticProperties.group(Labels.withId(KEY_PROXY_GROUP),
-
StaticProperties.stringFreeTextProperty(Labels.withId(KEY_PROXY_URL))
- )
- ))
- .requiredAlternatives(
- Labels.withId(KEY_MESSAGE_TYPE_ALTERNATIVES),
- Alternatives.from(
- Labels.withId(KEY_MESSAGE_SIMPLE),
- StaticProperties.stringFreeTextProperty(
- Labels.withId(KEY_MESSAGE_SIMPLE_CONTENT),
- true,
- true
- ),
- true
- ),
- Alternatives.from(
- Labels.withId(KEY_MESSAGE_ADVANCED),
- StaticProperties.stringFreeTextProperty(
- Labels.withId(KEY_MESSAGE_ADVANCED_CONTENT),
- true,
- true
- )
- )
- );
- }
-
- @Override
- public void onPipelineStopped() {
- try {
- this.httpClient.close();
- } catch (IOException e) {
- throw new SpRuntimeException("Error when closing MSTeams client:
%s".formatted(e.getMessage()));
- }
- }
-
- /**
- * Creates a JSON string intended for the MS Teams Webhook URL based on the
provided plain message content.
- * <p>
- * This method utilizes a basic approach for constructing messages to be
sent to MS Teams.
- * If you intend to provide text in the form of Adaptive Cards, consider
using
- * {@link #createMessageFromAdvancedContent(String)} for a more advanced and
interactive message format.
- * </p>
- *
- * @param messageContent The plain message content to be included in the
Teams message.
- * @return A JSON string formatted using a predefined template with the
provided message content.
- */
- protected String createMessageFromSimpleContent(String messageContent) {
- return SIMPLE_MESSAGE_TEMPLATE.formatted(messageContent);
- }
-
- /**
- * Creates a message for MS Teams from a JSON string, specifically designed
for use with Adaptive Cards.
- * <p>
- * This method takes a JSON string as input, which is expected to represent
the content of the message.
- * The content is directly forwarded to MS Teams, allowing for the
utilization of Adaptive Cards.
- * Adaptive Cards provide a flexible and interactive way to present content
in Microsoft Teams.
- * Learn more about Adaptive Cards: <a
href="https://learn.microsoft.com/en-us/adaptive-cards/";>here</a>
- * </p>
- *
- * @param messageContent The JSON string representing the content of the
message.
- * @return The original JSON string, unchanged.
- * @throws SpRuntimeException If the provided message is not a valid JSON
string.
- */
- protected String createMessageFromAdvancedContent(String messageContent) {
- try {
- objectMapper.readValue(messageContent, Object.class);
- } catch (JsonProcessingException e) {
- throw new SpRuntimeException(
- "Advanced message content provided is not a valid JSON string:
%s".formatted(messageContent),
- e
- );
- }
- return messageContent;
- }
-
- /**
- * Sends a payload to a webhook using the provided HTTP client, payload, and
webhook URL.
- *
- * @param mockedClient The HTTP client used to send the payload.
- * @param payload The payload to be sent to the webhook.
- * @param webhookUrl The URL of the webhook to which the payload will be
sent.
- * @throws SpRuntimeException If an I/O error occurs while sending the
payload to the webhook or
- * the payload sent is not accepted by the API.
- */
- protected void sendPayloadToWebhook(CloseableHttpClient mockedClient, String
payload, String webhookUrl) {
-
- for (int attempt = 1; ; attempt++) {
- HttpPost request = new HttpPost(webhookUrl);
- request.setEntity(new StringEntity(payload,
ContentType.APPLICATION_JSON));
-
- if (Thread.currentThread().isInterrupted()) {
- throw new SpRuntimeException("Interrupted while sending MS Teams
webhook");
- }
-
- try (CloseableHttpResponse response = mockedClient.execute(request)) {
- int status = response.getStatusLine().getStatusCode();
- if (status >= 200 && status < 300) {
- return;
+ var extractor = parameters.extractor();
+ webhookUrl =
validateWebhookUrl(extractor.secretValue(KEY_WEBHOOK_URL));
+
+ var selectedAlternative =
extractor.selectedAlternativeInternalId(KEY_MESSAGE_TYPE_ALTERNATIVES);
+ if (selectedAlternative.equals(KEY_MESSAGE_ADVANCED)) {
+ isSimpleMessageMode = false;
+ messageContent =
extractor.singleValueParameter(KEY_MESSAGE_ADVANCED_CONTENT, String.class);
+ } else {
+ isSimpleMessageMode = true;
+ messageContent =
extractor.singleValueParameter(KEY_MESSAGE_SIMPLE_CONTENT, String.class);
}
- if (status != HTTP_TOO_MANY_REQUESTS && (status < 500 || status >=
600)) {
- throw new SpRuntimeException("MS Teams webhook rejected request
(status=%d)".formatted(status));
+ var selectedProxyAlternative =
extractor.selectedAlternativeInternalId(KEY_PROXY_ALTERNATIVES);
+ var requestConfig =
RequestConfig.custom().setConnectTimeout(CONNECT_TIMEOUT_MS)
+
.setSocketTimeout(SOCKET_TIMEOUT_MS).setConnectionRequestTimeout(CONNECTION_REQUEST_TIMEOUT_MS).build();
+
+ if (selectedProxyAlternative.equals(KEY_PROXY_DISABLED)) {
+ this.httpClient =
HttpClientBuilder.create().setDefaultRequestConfig(requestConfig).build();
+ LOG.info("MS Teams sink initialized (no proxy), webhook host={}",
webhookUrl.getHost());
+ } else {
+ var proxyUrl = extractor.singleValueParameter(KEY_PROXY_URL,
String.class);
+ this.httpClient =
HttpClientBuilder.create().setDefaultRequestConfig(requestConfig)
+ .setProxy(HttpHost.create(proxyUrl)).build();
+ LOG.info("MS Teams sink initialized via proxy {}", proxyUrl);
}
+ }
+
+ @Override
+ public void onNotificationEvent(Event event) {
- if (attempt > MAX_RETRIES) {
- throw new SpRuntimeException("MS Teams webhook failed after %d
attempts (status=%d)"
- .formatted(attempt - 1, status));
+ // This sink allows to use placeholders for event properties when
defining the
+ // message content in the UI
+ // Therefore, we need to replace these placeholders based on the
actual event
+ // before actually sending the message
+ var processedMessageContent =
PlaceholderExtractor.replacePlaceholders(event, messageContent);
+
+ String teamsMessageContent;
+ if (isSimpleMessageMode) {
+ teamsMessageContent =
createMessageFromSimpleContent(processedMessageContent);
+ } else {
+ teamsMessageContent =
createMessageFromAdvancedContent(processedMessageContent);
}
+ sendPayloadToWebhook(httpClient, teamsMessageContent, webhookUrl);
+ }
- long backoffMs = BASE_BACKOFF.toMillis() << Math.min(attempt, 6);
+ @Override
+ public DataSinkBuilder declareModelWithoutSilentPeriod() {
+ return DataSinkBuilder.create(ID, 1).withLocales(Locales.EN)
+ .withAssets(ExtensionAssetType.DOCUMENTATION,
ExtensionAssetType.ICON)
+ .category(DataSinkType.NOTIFICATION)
+ .requiredStream(
+
StreamRequirementsBuilder.create().requiredProperty(EpRequirements.anyProperty()).build())
+ .requiredSecret(Labels.withId(KEY_WEBHOOK_URL))
+ .requiredAlternatives(Labels.withId(KEY_PROXY_ALTERNATIVES),
+ Alternatives.from(Labels.withId(KEY_PROXY_DISABLED)),
+ Alternatives.from(Labels.withId(KEY_PROXY_ENABLED),
+
StaticProperties.group(Labels.withId(KEY_PROXY_GROUP),
+
StaticProperties.stringFreeTextProperty(Labels.withId(KEY_PROXY_URL)))))
+
.requiredAlternatives(Labels.withId(KEY_MESSAGE_TYPE_ALTERNATIVES),
+ Alternatives.from(Labels.withId(KEY_MESSAGE_SIMPLE),
+
StaticProperties.stringFreeTextProperty(Labels.withId(KEY_MESSAGE_SIMPLE_CONTENT),
true,
+ true),
+ true),
+ Alternatives.from(Labels.withId(KEY_MESSAGE_ADVANCED),
StaticProperties
+
.stringFreeTextProperty(Labels.withId(KEY_MESSAGE_ADVANCED_CONTENT), true,
true)));
+ }
- Header retryAfter = response.getFirstHeader("Retry-After");
- if (retryAfter != null) {
- try {
- backoffMs = Long.parseLong(retryAfter.getValue()) * 1000;
- } catch (NumberFormatException ignored) {}
+ @Override
+ public void onPipelineStopped() {
+ if (httpClient != null) {
+ try {
+ httpClient.close();
+ LOG.info("MS Teams sink stopped, HTTP client closed");
+ } catch (IOException e) {
+ LOG.warn("Error closing MS Teams HTTP client: {}",
e.getMessage());
+ }
}
+ }
+
+ protected String createMessageFromSimpleContent(String messageContent) {
+ var card = objectMapper.createObjectNode();
+ card.put("$schema",
"http://adaptivecards.io/schemas/adaptive-card.json";);
+ card.put("type", "AdaptiveCard");
+ card.put("version", "1.4");
- Thread.sleep(backoffMs);
- } catch (IOException | InterruptedException e) {
- if (attempt > MAX_RETRIES) {
- throw new SpRuntimeException("I/O error sending MS Teams webhook",
e);
+ var textBlock = objectMapper.createObjectNode();
+ textBlock.put("type", "TextBlock");
+ textBlock.put("text", messageContent);
+ textBlock.put("wrap", true);
+ card.putArray("body").add(textBlock);
+
+ var attachment = objectMapper.createObjectNode();
+ attachment.put("contentType",
"application/vnd.microsoft.card.adaptive");
+ attachment.set("content", card);
+
+ var message = objectMapper.createObjectNode();
+ message.put("type", "message");
+ message.putArray("attachments").add(attachment);
+
+ try {
+ return objectMapper.writeValueAsString(message);
+ } catch (JsonProcessingException e) {
+ throw new SpRuntimeException("Could not serialize MS Teams message
content", e);
}
+ }
+ /**
+ * Creates a message for MS Teams from a JSON string, specifically
designed for
+ * use with Adaptive Cards.
+ * <p>
+ * This method takes a JSON string as input, which is expected to
represent the
+ * content of the message. The content is directly forwarded to MS Teams,
+ * allowing for the utilization of Adaptive Cards. Adaptive Cards provide a
+ * flexible and interactive way to present content in Microsoft Teams.
Learn
+ * more about Adaptive Cards:
+ * <a href="https://learn.microsoft.com/en-us/adaptive-cards/";>here</a>
+ * </p>
+ *
+ * @param messageContent The JSON string representing the content of the
+ * message.
+ * @return The original JSON string, unchanged.
+ * @throws SpRuntimeException If the provided message is not a valid JSON
+ * string.
+ */
+ protected String createMessageFromAdvancedContent(String messageContent) {
try {
- Thread.sleep(BASE_BACKOFF.toMillis() << Math.min(attempt, 6));
- } catch (InterruptedException ie) {
- Thread.currentThread().interrupt();
- throw new SpRuntimeException("Interrupted while retrying MS Teams
webhook", ie);
+ objectMapper.readValue(messageContent, Object.class);
+ } catch (JsonProcessingException e) {
+ throw new SpRuntimeException(
+ "Advanced message content provided is not a valid JSON
string: %s".formatted(messageContent), e);
}
- }
+ return messageContent;
}
- }
-
- /**
- * Validates a webhook URL to ensure it is not null, not empty, and has a
valid URL format.
- *
- * @param webhookUrl The webhook URL to be validated.
- * @throws SpRuntimeException If the webhook URL is null or empty, or if it
is not a valid URL.
- */
- protected void validateWebhookUrl(String webhookUrl) {
- if (webhookUrl == null || webhookUrl.isEmpty()) {
- throw new SpRuntimeException("Given webhook URL is empty");
+
+ /**
+ * Sends a payload to the configured MS Teams webhook, retrying transient
+ * failures with a fixed delay.
+ * <p>
+ * A request is retried up to {@value #MAX_ATTEMPTS} times (waiting
+ * {@value #RETRY_DELAY_MS} ms between attempts) when the call fails with
an I/O
+ * error or the webhook responds with HTTP 429 or a 5xx status. A 4xx
status
+ * other than 429 is treated as permanent (e.g. an invalid webhook, revoked
+ * token, or malformed card) and fails immediately without retrying.
+ * </p>
+ * <p>
+ * If the calling thread is interrupted (for example because the pipeline
is
+ * being stopped), the method restores the interrupt flag and returns
quietly
+ * without throwing, since an aborted send during shutdown is not a
failure.
+ * </p>
+ *
+ * @param client The HTTP client used to send the payload.
+ * @param payload The payload to be sent to the webhook.
+ * @param url The URL of the webhook to which the payload will be sent.
+ * @throws SpRuntimeException If the webhook rejects the message with a
+ * permanent (non-429 4xx) status, or if all
retry
+ * attempts are exhausted without success.
+ */
+ void sendPayloadToWebhook(CloseableHttpClient client, String payload, URI
url) {
+ var post = new HttpPost(url);
+ post.setEntity(new StringEntity(payload,
ContentType.APPLICATION_JSON));
+
+ SpRuntimeException last = null;
+
+ for (int attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
+ LOG.debug("Sending notification to MS Teams (attempt {}/{})",
attempt, MAX_ATTEMPTS);
+
+ try (CloseableHttpResponse response = client.execute(post)) {
+ int status = response.getStatusLine().getStatusCode();
+
+ if (status >= 200 && status < 300) {
+ if (attempt > 1) {
+ LOG.info("MS Teams notification succeeded on attempt
{}/{}", attempt, MAX_ATTEMPTS);
+ } else {
+ LOG.debug("MS Teams notification sent (HTTP {})",
status);
+ }
+ return;
+ }
+
+ if (status >= 400 && status < 500 && status != 429) {
+ // client error: retrying won't help (bad webhook, revoked
token, malformed
+ // card)
+ LOG.error("MS Teams rejected the message with HTTP {} -
not retrying", status);
+ throw new SpRuntimeException("Teams rejected the message:
HTTP " + status);
+ }
+
+ // 5xx or 429 -> transient, worth retrying
+ LOG.warn("MS Teams returned HTTP {} (attempt {}/{})", status,
attempt, MAX_ATTEMPTS);
+ last = new SpRuntimeException("Teams returned HTTP " + status);
+
+ } catch (IOException e) {
+ if (Thread.currentThread().isInterrupted()) {
+ LOG.debug("MS Teams request aborted because the pipeline
is stopping - ignoring");
+ return; // clean shutdown, not a failure
+ }
+ LOG.warn("MS Teams request failed (attempt {}/{}): {}",
attempt, MAX_ATTEMPTS, e.getMessage());
+ last = new SpRuntimeException("Sending notification to MS
Teams failed.", e);
+ }
+
+ if (attempt < MAX_ATTEMPTS) {
+ LOG.info("Retrying MS Teams notification in {} ms",
RETRY_DELAY_MS);
+ try {
+ Thread.sleep(RETRY_DELAY_MS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOG.debug("Interrupted while waiting to retry - pipeline
stopping, giving up");
+ return;
+ }
+ }
+ }
+
+ LOG.error("Giving up on MS Teams notification after {} attempts",
MAX_ATTEMPTS);
+ // 'last' is always assigned before reaching this point, but guard
defensively.
+ throw last != null ? last : new SpRuntimeException("Sending
notification to MS Teams failed.");
}
- try {
- new URL(webhookUrl);
- } catch (MalformedURLException e) {
- throw new SpRuntimeException("The given webhook is not a valid URL:
%s".formatted(webhookUrl));
+
+ /**
+ * Validates a webhook URL to ensure it is not null, not empty, and has a
valid
+ * URL format.
+ *
+ * @param webhookUrl The webhook URL to be validated.
+ * @throws SpRuntimeException If the webhook URL is null or empty, or if
it is
+ * not a valid URL.
+ */
+ protected URI validateWebhookUrl(String webhookUrl) {
+ if (webhookUrl == null || webhookUrl.isEmpty()) {
+ throw new SpRuntimeException("Given webhook URL is empty");
+ }
+ try {
+ URI uri = new URI(webhookUrl);
+ if (uri.getScheme() == null || uri.getHost() == null) {
+ throw new SpRuntimeException("The given webhook URL is not
absolute or has no host");
+ }
+ if (!"http".equalsIgnoreCase(uri.getScheme()) &&
!"https".equalsIgnoreCase(uri.getScheme())) {
+ throw new SpRuntimeException("The given webhook URL must use
http or https");
+ }
+ return uri;
+ } catch (URISyntaxException e) {
+ throw new SpRuntimeException("The given webhook URL is not valid",
e);
+ }
}
- }
-}
+}
\ No newline at end of file
diff --git
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/resources/org.apache.streampipes.sinks.notifications.jvm.msteams/documentation.md
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/resources/org.apache.streampipes.sinks.notifications.jvm.msteams/documentation.md
index 3c6d41d4f9..f055757a66 100644
---
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/resources/org.apache.streampipes.sinks.notifications.jvm.msteams/documentation.md
+++
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/resources/org.apache.streampipes.sinks.notifications.jvm.msteams/documentation.md
@@ -43,28 +43,21 @@ with any type of incoming event, making it a versatile
choice for various use ca
#### Webhook URL
-To configure the MS Teams Sink, you need to provide the Webhook URL that
enables the sink to send messages to a specific
-MS Teams channel. If you don't have a Webhook URL, you can learn how to create
-one
[here](https://learn.microsoft.com/en-us/microsoftteams/platform/webhooks-and-connectors/how-to/add-incoming-webhook?tabs=dotnet#create-incoming-webhooks-1).
+To configure the MS Teams Sink, you need to provide the Webhook URL that
enables the sink to send
+messages to a specific MS Teams channel. The webhook must be created as a
Power Automate workflow
+using the *"Post to a channel when a webhook request is received"* workflow
template.
#### Message Content Options
You can choose between two message content formats:
-- **Simple Message Content:** Supports plain text and basic markdown
formatting.
-- **Advanced Message Content:** Expects JSON input directly forwarded to Teams
without modification. This format is
- highly customizable and can be used for Adaptive Cards.
+- **Simple Message Content:** Send plain text. The text is automatically
wrapped in an Adaptive Card
+ before being sent to Teams.
+- **Advanced Message Content:** Expects JSON input directly forwarded to Teams
without modification.
+ This format is highly customizable and can be used for Adaptive Cards.
Choose the format that best suits your messaging needs.
-### Silent Period
-
-The *Silent Period* is the duration, expressed in minutes, during which
notifications are temporarily disabled after one
-has been sent. This feature is implemented to prevent overwhelming the target
with frequent notifications, avoiding
-potential spam behavior.
-
----
-
## Usage
#### Simple Message Format
diff --git
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/test/java/org/apache/streampipes/sinks/notifications/jvm/msteams/TestMSTeamsSink.java
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/test/java/org/apache/streampipes/sinks/notifications/jvm/msteams/TestMSTeamsSink.java
index 8c5617f3fa..55dd51aca6 100644
---
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/test/java/org/apache/streampipes/sinks/notifications/jvm/msteams/TestMSTeamsSink.java
+++
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/test/java/org/apache/streampipes/sinks/notifications/jvm/msteams/TestMSTeamsSink.java
@@ -32,7 +32,10 @@ import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
@@ -43,84 +46,151 @@ import static org.mockito.Mockito.when;
public class TestMSTeamsSink {
- @Test
- public void createMessageFromSimpleContent() {
- var messageContent = "This is test";
- var sink = new MSTeamsSink();
+ @Test
+ public void createMessageFromSimpleContent() {
+ var sink = new MSTeamsSink();
+
+ var expectedTeamsMessage = """
+ {
+ "type" : "message",
+ "attachments" : [ {
+ "contentType" : "application/vnd.microsoft.card.adaptive",
+ "content" : {
+ "$schema" :
"http://adaptivecards.io/schemas/adaptive-card.json";,
+ "type" : "AdaptiveCard",
+ "version" : "1.4",
+ "body" : [ {
+ "type" : "TextBlock",
+ "text" : "This is test",
+ "wrap" : true
+ } ]
+ }
+ } ]
+ }""";
+
+ var createdTeamsMessage = sink.createMessageFromSimpleContent("This is
test");
+
+ assertEquals(
+ expectedTeamsMessage.replace("\r\n", "\n"),
+ createdTeamsMessage.replace("\r\n", "\n")
+ );
+ }
+
+ @Test
+ public void createMessageFromAdvancedContent() {
+ var messageContent = "{\"text\": \"Hi this is a message from Apache
StreamPipes\"}";
+
+ var sink = new MSTeamsSink();
+ assertEquals(messageContent,
sink.createMessageFromAdvancedContent(messageContent));
+ }
+
+ @Test
+ public void createMessageFromAdvancedContentCheckException() {
+ var messageContent = "invalid-complex-input";
+
+ var sink = new MSTeamsSink();
+
+ assertThrows(SpRuntimeException.class, () ->
sink.createMessageFromAdvancedContent(messageContent));
+ }
+
+ @Test
+ public void sendPayloadToWebhook() throws IOException, URISyntaxException {
+
+ var mockedClient = mock(CloseableHttpClient.class);
+ var mockedResponse = mock(CloseableHttpResponse.class);
+ var mockedStatusLine = mock(StatusLine.class);
+ var argumentCaptor = ArgumentCaptor.forClass(HttpPost.class);
+
+ when(mockedStatusLine.getStatusCode()).thenReturn(HttpStatus.SC_OK);
+ when(mockedResponse.getStatusLine()).thenReturn(mockedStatusLine);
+ when(mockedClient.execute(any())).thenReturn(mockedResponse);
+
+ var payload = "This is a test";
+ var webhook = "https://webhook.com";;
+ var sink = new MSTeamsSink();
+
+ sink.sendPayloadToWebhook(mockedClient, payload, new URI(webhook));
+ verify(mockedClient, times(1)).execute(argumentCaptor.capture());
+
+ var capturedPost = argumentCaptor.getValue();
+
+ Assertions.assertNotNull(capturedPost);
+ assertEquals(webhook, capturedPost.getURI().toString());
+ assertEquals(ContentType.APPLICATION_JSON.toString(),
capturedPost.getEntity().getContentType().getValue());
+ assertEquals(payload, EntityUtils.toString(capturedPost.getEntity()));
+ }
+
+ @Test
+ public void sendPayloadToWebhookBadResponse() throws IOException {
+ CloseableHttpClient mockedClient = mock(CloseableHttpClient.class);
+ var mockedResponse = mock(CloseableHttpResponse.class);
+ var mockedStatusLine = mock(StatusLine.class);
+
+
when(mockedStatusLine.getStatusCode()).thenReturn(HttpStatus.SC_BAD_REQUEST);
+ when(mockedResponse.getStatusLine()).thenReturn(mockedStatusLine);
+ when(mockedClient.execute(any())).thenReturn(mockedResponse);
+
+ var sink = new MSTeamsSink();
+ var payload = "<a>invalid</a>";
+ var url = "https://webhook.com";;
+
+ assertThrows(SpRuntimeException.class, () ->
sink.sendPayloadToWebhook(mockedClient, payload, new URI(url)));
+
+ // A 4xx (other than 429) is permanent: it must fail on the first
attempt, no
+ // retries.
+ verify(mockedClient, times(1)).execute(any());
+ }
+
+ @Test
+ public void sendPayloadToWebhookRetriesOnServerErrorThenGivesUp() throws
IOException {
+ var mockedClient = mock(CloseableHttpClient.class);
+ var mockedResponse = mock(CloseableHttpResponse.class);
+ var mockedStatusLine = mock(StatusLine.class);
+
+ // Every attempt returns a 500 -> transient, should be retried up to
+ // MAX_ATTEMPTS.
+
when(mockedStatusLine.getStatusCode()).thenReturn(HttpStatus.SC_INTERNAL_SERVER_ERROR);
+ when(mockedResponse.getStatusLine()).thenReturn(mockedStatusLine);
+ when(mockedClient.execute(any())).thenReturn(mockedResponse);
+
+ var sink = new MSTeamsSink();
+
+ assertThrows(SpRuntimeException.class,
+ () -> sink.sendPayloadToWebhook(mockedClient, "payload", new
URI("https://webhook.com";)));
+
+ // After exhausting all attempts the call must have been made
MAX_ATTEMPTS
+ // times.
+ verify(mockedClient, times(3)).execute(any());
+ }
+
+ @Test
+ public void sendPayloadToWebhookRecoversAfterTransientFailure() throws
IOException {
+ var mockedClient = mock(CloseableHttpClient.class);
+ var failResponse = mock(CloseableHttpResponse.class);
+ var failStatusLine = mock(StatusLine.class);
+ var okResponse = mock(CloseableHttpResponse.class);
+ var okStatusLine = mock(StatusLine.class);
+
+
when(failStatusLine.getStatusCode()).thenReturn(HttpStatus.SC_SERVICE_UNAVAILABLE);
+ when(failResponse.getStatusLine()).thenReturn(failStatusLine);
+ when(okStatusLine.getStatusCode()).thenReturn(HttpStatus.SC_OK);
+ when(okResponse.getStatusLine()).thenReturn(okStatusLine);
- assertEquals(MSTeamsSink.SIMPLE_MESSAGE_TEMPLATE.formatted(messageContent),
-
sink.createMessageFromSimpleContent(messageContent));
- }
-
- @Test
- public void createMessageFromAdvancedContent() {
- var messageContent = "{\"text\": \"Hi this is a message from Apache
StreamPipes\"}";
-
- var sink = new MSTeamsSink();
- assertEquals(messageContent,
sink.createMessageFromAdvancedContent(messageContent));
- }
-
- @Test
- public void createMessageFromAdvancedContentCheckException() {
- var messageContent = "invalid-complex-input";
-
- var sink = new MSTeamsSink();
-
- assertThrows(SpRuntimeException.class, () ->
sink.createMessageFromAdvancedContent(messageContent));
- }
-
- @Test
- public void sendPayloadToWebhook() throws IOException {
-
- var mockedClient = mock(CloseableHttpClient.class);
- var mockedResponse = mock(CloseableHttpResponse.class);
- var mockedStatusLine = mock(StatusLine.class);
- var argumentCaptor = ArgumentCaptor.forClass(HttpPost.class);
-
- when(mockedStatusLine.getStatusCode()).thenReturn(HttpStatus.SC_OK);
- when(mockedResponse.getStatusLine()).thenReturn(mockedStatusLine);
- when(mockedClient.execute(any())).thenReturn(mockedResponse);
-
- var payload = "This is a test";
- var webhook = "https://webhook.com";;
- var sink = new MSTeamsSink();
-
- sink.sendPayloadToWebhook(mockedClient, payload, webhook);
- verify(mockedClient, times(1)).execute(argumentCaptor.capture());
-
-
- var capturedPost = argumentCaptor.getValue();
-
- Assertions.assertNotNull(capturedPost);
- assertEquals(webhook,
- capturedPost.getURI().toString()
- );
- assertEquals(ContentType.APPLICATION_JSON.toString(),
-
capturedPost.getEntity().getContentType().getValue());
- assertEquals(payload, EntityUtils.toString(capturedPost.getEntity()));
- }
-
- @Test
- public void sendPayloadToWebhookBadResponse() throws IOException {
- CloseableHttpClient mockedClient = mock(CloseableHttpClient.class);
- var mockedResponse = mock(CloseableHttpResponse.class);
- var mockedStatusLine = mock(StatusLine.class);
-
-
when(mockedStatusLine.getStatusCode()).thenReturn(HttpStatus.SC_BAD_REQUEST);
- when(mockedResponse.getStatusLine()).thenReturn(mockedStatusLine);
- when(mockedClient.execute(any())).thenReturn(mockedResponse);
-
- var sink = new MSTeamsSink();
- var payload = "<a>invalid</a>";
- var url = "https://webhook.com";;
-
- assertThrows(SpRuntimeException.class, () ->
sink.sendPayloadToWebhook(mockedClient, payload, url));
- }
-
- @Test
- public void validateWebhookUrl() {
- var sink = new MSTeamsSink();
- assertThrows(SpRuntimeException.class, () -> sink.validateWebhookUrl(""));
- assertThrows(SpRuntimeException.class, () ->
sink.validateWebhookUrl("some-string"));
- }
-}
+ // First call fails with 503, second call succeeds.
+
when(mockedClient.execute(any())).thenReturn(failResponse).thenReturn(okResponse);
+
+ var sink = new MSTeamsSink();
+
+ assertDoesNotThrow(() -> sink.sendPayloadToWebhook(mockedClient,
"payload", new URI("https://webhook.com";)));
+
+ // One failed attempt + one successful retry = two executions, no
third.
+ verify(mockedClient, times(2)).execute(any());
+ }
+
+ @Test
+ public void validateWebhookUrl() {
+ var sink = new MSTeamsSink();
+ assertThrows(SpRuntimeException.class, () ->
sink.validateWebhookUrl(""));
+ assertThrows(SpRuntimeException.class, () ->
sink.validateWebhookUrl("some-string"));
+ }
+}
\ No newline at end of file
