This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 1bc77dcd6774 CAMEL-26027: camel-telegram - concurrent sends through
one endpoint each build their own request (#28311)
1bc77dcd6774 is described below
commit 1bc77dcd677472a89178b75cbce570f55edf6034
Author: Claus Ibsen <[email protected]>
AuthorDate: Fri Oct 23 10:44:41 2026 +0200
CAMEL-26027: camel-telegram - concurrent sends through one endpoint each
build their own request (#28311)
Each message handler of the telegram producer kept one
TelegramBodyPublisher shared by all sends, so concurrent sends through one
endpoint mixed the parts of their requests. Each send now builds its request
body in its own TelegramBodyPublisher.
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
---
.../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 982945905532..30ac93fc271d 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 4da26ab2c504..891f7db4ac14 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
@@ -31,7 +31,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;
@@ -45,14 +44,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() + "\"");
@@ -90,11 +89,7 @@ abstract class TelegramMessageHandler<T extends
TelegramMessage> {
callback.done(false);
}
- 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) {
@@ -105,10 +100,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..444f3f713e89
--- /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.getPort())
+ .addEndpoint(
+ "sendMessage",
+ "POST",
+ String.class,
+
TelegramTestUtil.stringResource("messages/send-message.json"));
+ }
+}
