This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-4.18.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.18.x by this push:
new 97a92844f0cb [backport camel-4.18.x] CAMEL-26027: camel-telegram -
concurrent sends through one endpoint each build their own request (#28335)
97a92844f0cb is described below
commit 97a92844f0cb717410dcfa969a88a6af9a17769d
Author: Guillaume Nodet <[email protected]>
AuthorDate: Mon Nov 2 09:56:40 2026 +0100
[backport camel-4.18.x] CAMEL-26027: camel-telegram - concurrent sends
through one endpoint each build their own request (#28335)
…s through one endpoint each build their own request
…18.x
---
.../telegram/service/OutgoingMessageHandler.java | 8 +-
.../telegram/service/TelegramHandlerRegistry.java | 2 +-
.../telegram/service/TelegramMessageHandler.java | 17 +---
.../service/TelegramServiceRestBotAPIAdapter.java | 58 ++++++------
.../telegram/TelegramProducerConcurrentTest.java | 100 +++++++++++++++++++++
5 files changed, 138 insertions(+), 47 deletions(-)
diff --git
a/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/OutgoingMessageHandler.java
b/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/OutgoingMessageHandler.java
index 2daa1d77fbe1..18e22962f29e 100644
---
a/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/OutgoingMessageHandler.java
+++
b/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/OutgoingMessageHandler.java
@@ -27,10 +27,10 @@ abstract class OutgoingMessageHandler<T extends
OutgoingMessage> extends Telegra
super(apiClient, uri, contentType, resultClass);
}
- protected void fillCommonMediaParts(OutgoingMessage message) {
+ protected void fillCommonMediaParts(TelegramBodyPublisher bodyPublisher,
OutgoingMessage message) {
- buildTextPart("chat_id", message.getChatId());
- buildTextPart("reply_to_message_id", message.getReplyToMessageId());
- buildTextPart("disable_notification",
message.getDisableNotification());
+ buildTextPart(bodyPublisher, "chat_id", message.getChatId());
+ buildTextPart(bodyPublisher, "reply_to_message_id",
message.getReplyToMessageId());
+ buildTextPart(bodyPublisher, "disable_notification",
message.getDisableNotification());
}
}
diff --git
a/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramHandlerRegistry.java
b/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramHandlerRegistry.java
index 314835d17169..628c4886cf42 100644
---
a/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramHandlerRegistry.java
+++
b/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramHandlerRegistry.java
@@ -82,7 +82,7 @@ final class TelegramHandlerRegistry {
}
@Override
- protected void addBody(TelegramMessage message) {
+ protected void addBody(TelegramBodyPublisher bodyPublisher,
TelegramMessage message) {
try {
final String body = mapper.writeValueAsString(message);
bodyPublisher.addBodyPart(new
TelegramBodyPublisher.SingleBodyPart(body));
diff --git
a/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramMessageHandler.java
b/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramMessageHandler.java
index fea04895e7ff..e019f705059c 100644
---
a/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramMessageHandler.java
+++
b/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramMessageHandler.java
@@ -29,7 +29,6 @@ import
org.apache.camel.component.telegram.model.MessageResult;
abstract class TelegramMessageHandler<T extends TelegramMessage> {
protected final ObjectMapper mapper;
- protected final TelegramBodyPublisher bodyPublisher;
private final TelegramApiClient apiClient;
private final String contentType;
@@ -43,14 +42,14 @@ abstract class TelegramMessageHandler<T extends
TelegramMessage> {
this.uri = uri;
this.contentType = contentType;
this.resultClass = resultClass;
-
- bodyPublisher = new TelegramBodyPublisher(apiClient.bufferSize());
}
public void sendMessage(Exchange exchange, AsyncCallback callback, T
message) {
HttpRequest.Builder builder =
HttpRequest.newBuilder().uri(URI.create(uri));
- addBody(message);
+ // the body of this request only: the producer may send from several
threads at once
+ TelegramBodyPublisher bodyPublisher = new
TelegramBodyPublisher(apiClient.bufferSize());
+ addBody(bodyPublisher, message);
if (bodyPublisher.getBodyParts().size() > 1) {
builder.setHeader("Content-type", "multipart/form-data;
boundary=\"" + bodyPublisher.getBoundary() + "\"");
@@ -77,11 +76,7 @@ abstract class TelegramMessageHandler<T extends
TelegramMessage> {
}
}
- protected abstract void addBody(T message);
-
- protected <T> void buildTextPart(String name, T value) {
- buildTextPart(bodyPublisher, name, value);
- }
+ protected abstract void addBody(TelegramBodyPublisher bodyPublisher, T
message);
static <T> void buildTextPart(TelegramBodyPublisher bodyPublisher, String
name, T value) {
if (value != null) {
@@ -92,10 +87,6 @@ abstract class TelegramMessageHandler<T extends
TelegramMessage> {
}
}
- protected void buildMediaPart(String name, String fileNameWithExtension,
byte[] value) {
- buildMediaPart(bodyPublisher, name, fileNameWithExtension, value);
- }
-
void buildMediaPart(TelegramBodyPublisher bodyPublisher, String name,
String fileNameWithExtension, byte[] value) {
TelegramBodyPublisher.MultilineBodyPart<byte[]> bodyPart
= new TelegramBodyPublisher.MultilineBodyPart<>(name, value,
"application/octet-stream", null);
diff --git
a/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramServiceRestBotAPIAdapter.java
b/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramServiceRestBotAPIAdapter.java
index a509dfbb7eb1..087e4656e393 100644
---
a/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramServiceRestBotAPIAdapter.java
+++
b/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramServiceRestBotAPIAdapter.java
@@ -236,13 +236,13 @@ public class TelegramServiceRestBotAPIAdapter implements
TelegramService {
}
@Override
- protected void addBody(OutgoingAudioMessage message) {
- fillCommonMediaParts(message);
- buildMediaPart("audio", message.getFilenameWithExtension(),
message.getAudio());
- buildTextPart("title", message.getTitle());
- buildTextPart("duration", message.getDurationSeconds());
- buildTextPart("performer", message.getPerformer());
- buildTextPart("reply_markup", message.replyMarkupJson());
+ protected void addBody(TelegramBodyPublisher bodyPublisher,
OutgoingAudioMessage message) {
+ fillCommonMediaParts(bodyPublisher, message);
+ buildMediaPart(bodyPublisher, "audio",
message.getFilenameWithExtension(), message.getAudio());
+ buildTextPart(bodyPublisher, "title", message.getTitle());
+ buildTextPart(bodyPublisher, "duration",
message.getDurationSeconds());
+ buildTextPart(bodyPublisher, "performer", message.getPerformer());
+ buildTextPart(bodyPublisher, "reply_markup",
message.replyMarkupJson());
}
}
@@ -254,14 +254,14 @@ public class TelegramServiceRestBotAPIAdapter implements
TelegramService {
}
@Override
- protected void addBody(OutgoingVideoMessage message) {
- fillCommonMediaParts(message);
- buildMediaPart("video", message.getFilenameWithExtension(),
message.getVideo());
- buildTextPart("caption", message.getCaption());
- buildTextPart("duration", message.getDurationSeconds());
- buildTextPart("width", message.getWidth());
- buildTextPart("height", message.getHeight());
- buildTextPart("reply_markup", message.replyMarkupJson());
+ protected void addBody(TelegramBodyPublisher bodyPublisher,
OutgoingVideoMessage message) {
+ fillCommonMediaParts(bodyPublisher, message);
+ buildMediaPart(bodyPublisher, "video",
message.getFilenameWithExtension(), message.getVideo());
+ buildTextPart(bodyPublisher, "caption", message.getCaption());
+ buildTextPart(bodyPublisher, "duration",
message.getDurationSeconds());
+ buildTextPart(bodyPublisher, "width", message.getWidth());
+ buildTextPart(bodyPublisher, "height", message.getHeight());
+ buildTextPart(bodyPublisher, "reply_markup",
message.replyMarkupJson());
}
}
@@ -272,11 +272,11 @@ public class TelegramServiceRestBotAPIAdapter implements
TelegramService {
}
@Override
- protected void addBody(OutgoingDocumentMessage message) {
- fillCommonMediaParts(message);
- buildMediaPart("document", message.getFilenameWithExtension(),
message.getDocument());
- buildTextPart("caption", message.getCaption());
- buildTextPart("reply_markup", message.replyMarkupJson());
+ protected void addBody(TelegramBodyPublisher bodyPublisher,
OutgoingDocumentMessage message) {
+ fillCommonMediaParts(bodyPublisher, message);
+ buildMediaPart(bodyPublisher, "document",
message.getFilenameWithExtension(), message.getDocument());
+ buildTextPart(bodyPublisher, "caption", message.getCaption());
+ buildTextPart(bodyPublisher, "reply_markup",
message.replyMarkupJson());
}
}
@@ -288,11 +288,11 @@ public class TelegramServiceRestBotAPIAdapter implements
TelegramService {
}
@Override
- protected void addBody(OutgoingPhotoMessage message) {
- fillCommonMediaParts(message);
- buildMediaPart("photo", message.getFilenameWithExtension(),
message.getPhoto());
- buildTextPart("caption", message.getCaption());
- buildTextPart("reply_markup", message.replyMarkupJson());
+ protected void addBody(TelegramBodyPublisher bodyPublisher,
OutgoingPhotoMessage message) {
+ fillCommonMediaParts(bodyPublisher, message);
+ buildMediaPart(bodyPublisher, "photo",
message.getFilenameWithExtension(), message.getPhoto());
+ buildTextPart(bodyPublisher, "caption", message.getCaption());
+ buildTextPart(bodyPublisher, "reply_markup",
message.replyMarkupJson());
}
}
@@ -303,12 +303,12 @@ public class TelegramServiceRestBotAPIAdapter implements
TelegramService {
}
@Override
- protected void addBody(OutgoingStickerMessage message) {
- fillCommonMediaParts(message);
+ protected void addBody(TelegramBodyPublisher bodyPublisher,
OutgoingStickerMessage message) {
+ fillCommonMediaParts(bodyPublisher, message);
if (message.getSticker() != null) {
- buildTextPart("sticker", message.getSticker());
+ buildTextPart(bodyPublisher, "sticker", message.getSticker());
} else {
- buildMediaPart("sticker", message.getFilenameWithExtension(),
message.getStickerImage());
+ buildMediaPart(bodyPublisher, "sticker",
message.getFilenameWithExtension(), message.getStickerImage());
}
}
}
diff --git
a/components/camel-telegram/src/test/java/org/apache/camel/component/telegram/TelegramProducerConcurrentTest.java
b/components/camel-telegram/src/test/java/org/apache/camel/component/telegram/TelegramProducerConcurrentTest.java
new file mode 100644
index 000000000000..eb9d9676e2d2
--- /dev/null
+++
b/components/camel-telegram/src/test/java/org/apache/camel/component/telegram/TelegramProducerConcurrentTest.java
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.camel.component.telegram;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.telegram.model.OutgoingTextMessage;
+import org.apache.camel.component.telegram.util.TelegramMockRoutes;
+import
org.apache.camel.component.telegram.util.TelegramMockRoutes.MockProcessor;
+import org.apache.camel.component.telegram.util.TelegramTestSupport;
+import org.apache.camel.component.telegram.util.TelegramTestUtil;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Sends from several threads at once through one endpoint: each request
carries the message of its own send only.
+ */
+public class TelegramProducerConcurrentTest extends TelegramTestSupport {
+
+ private static final int MESSAGES = 100;
+
+ @Test
+ public void testConcurrentSends() throws Exception {
+ final MockProcessor<String> mockProcessor =
getMockRoutes().getMock("sendMessage");
+ mockProcessor.clearRecordedMessages();
+
+ ExecutorService executor = Executors.newFixedThreadPool(10);
+ try {
+ List<Future<?>> sends = new ArrayList<>();
+ for (int i = 0; i < MESSAGES; i++) {
+ final int n = i;
+ sends.add(executor.submit(() ->
template.sendBodyAndHeader("direct:telegram", "msg-" + n,
+ TelegramConstants.TELEGRAM_CHAT_ID, "chat-" + n)));
+ }
+ for (Future<?> send : sends) {
+ send.get(30, TimeUnit.SECONDS);
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+
+ // each request is the JSON of one message, sent to the chat of that
message
+ ObjectMapper mapper = new ObjectMapper();
+ Set<String> texts = new HashSet<>();
+ for (String request : mockProcessor.awaitRecordedMessages(MESSAGES,
10000)) {
+ OutgoingTextMessage message = mapper.readValue(request,
OutgoingTextMessage.class);
+ assertEquals(message.getText().replace("msg-", "chat-"),
message.getChatId(), request);
+ texts.add(message.getText());
+ }
+ assertEquals(MESSAGES, texts.size());
+ }
+
+ @Override
+ protected RoutesBuilder[] createRouteBuilders() {
+ return new RoutesBuilder[] {
+ getMockRoutes(),
+ new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:telegram")
+
.to("telegram:bots?authorizationToken=mock-token");
+ }
+ } };
+ }
+
+ @Override
+ protected TelegramMockRoutes createMockRoutes() {
+ return new TelegramMockRoutes(port)
+ .addEndpoint(
+ "sendMessage",
+ "POST",
+ String.class,
+
TelegramTestUtil.stringResource("messages/send-message.json"));
+ }
+}
