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 c78c70da9652 CAMEL-25523: camel-slack - Read the following pages of
the history so the consumer does not skip messages (#27674)
c78c70da9652 is described below
commit c78c70da96521b110d5cf7d0de28660b6d6d921e
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Oct 11 12:31:58 2026 +0530
CAMEL-25523: camel-slack - Read the following pages of the history so the
consumer does not skip messages (#27674)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../org/apache/camel/catalog/components/slack.json | 2 +-
.../apache/camel/catalog/docs/slack-component.adoc | 6 +
.../org/apache/camel/component/slack/slack.json | 2 +-
.../camel-slack/src/main/docs/slack-component.adoc | 6 +
.../camel/component/slack/SlackConsumer.java | 59 ++++-
.../camel/component/slack/SlackEndpoint.java | 3 +-
.../component/slack/SlackConsumerHasMoreTest.java | 292 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 9 +
.../endpoint/dsl/SlackEndpointBuilderFactory.java | 4 +-
9 files changed, 374 insertions(+), 9 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/slack.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/slack.json
index 2acc5b4d73b0..40dcfb95686e 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/slack.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/slack.json
@@ -36,7 +36,7 @@
"channel": { "index": 0, "kind": "path", "displayName": "Channel",
"group": "common", "label": "", "required": true, "type": "string", "javaType":
"java.lang.String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The channel name (syntax #name) or
slack user (syntax userName) to send a message directly to an user." },
"token": { "index": 1, "kind": "parameter", "displayName": "Token",
"group": "common", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": true, "security": "secret", "description": "The token to access
Slack. This app needs to have channels:history, groups:history, im:history,
mpim:history, channels:read, groups:read, im:read and mpim:read permissions.
The User OAuth Token is the kind of token needed." },
"conversationType": { "index": 2, "kind": "parameter", "displayName":
"Conversation Type", "group": "consumer", "label": "consumer", "required":
false, "type": "enum", "javaType": "com.slack.api.model.ConversationType",
"enum": [ "PUBLIC_CHANNEL", "PRIVATE_CHANNEL", "MPIM", "IM" ], "deprecated":
false, "autowired": false, "secret": false, "defaultValue": "PUBLIC_CHANNEL",
"description": "Type of conversation" },
- "maxResults": { "index": 3, "kind": "parameter", "displayName": "Max
Results", "group": "consumer", "label": "consumer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": "10", "description": "The Max Result
for the poll" },
+ "maxResults": { "index": 3, "kind": "parameter", "displayName": "Max
Results", "group": "consumer", "label": "consumer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": "10", "description": "The maximum
number of messages that the consumer requests from the conversation history at
a time. When more new messages are available, the consumer reads the following
pages in the same poll." },
"naturalOrder": { "index": 4, "kind": "parameter", "displayName": "Natural
Order", "group": "consumer", "label": "consumer", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": false, "description": "Create exchanges in
natural order (oldest to newest) or not" },
"sendEmptyMessageWhenIdle": { "index": 5, "kind": "parameter",
"displayName": "Send Empty Message When Idle", "group": "consumer", "label":
"consumer", "required": false, "type": "boolean", "javaType": "boolean",
"deprecated": false, "autowired": false, "secret": false, "defaultValue":
false, "description": "If the polling consumer did not poll any files, you can
enable this option to send an empty message (no body) instead." },
"serverUrl": { "index": 6, "kind": "parameter", "displayName": "Server
Url", "group": "consumer", "label": "consumer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": "https:\/\/slack.com", "description":
"The Server URL of the Slack instance" },
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/slack-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/slack-component.adoc
index 4f7350e13eab..b0708b1d502b 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/slack-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/slack-component.adoc
@@ -188,6 +188,12 @@ YAML::
This way you'll get the last message from `general` channel.
The consumer will track the timestamp of the last message consumed,
and in the next poll it will consume only newer messages in the channel.
+When more than `maxResults` new messages were posted since the previous poll,
the consumer reads the following pages of
+the conversation history in the same poll (`maxResults` messages per request),
so no message is skipped. For a busy
+channel, set a higher `maxResults` to need fewer requests. If the request for
a following page fails, for example
+because Slack rate limits it (some apps may send only one
`conversations.history` request per minute), the consumer
+routes the most recent messages it has already read, as when it reads one page
only, and logs a WARN that the older
+messages of that poll were skipped.
You'll need to create a Slack app and use it in your workspace.
diff --git
a/components/camel-slack/src/generated/resources/META-INF/org/apache/camel/component/slack/slack.json
b/components/camel-slack/src/generated/resources/META-INF/org/apache/camel/component/slack/slack.json
index 2acc5b4d73b0..40dcfb95686e 100644
---
a/components/camel-slack/src/generated/resources/META-INF/org/apache/camel/component/slack/slack.json
+++
b/components/camel-slack/src/generated/resources/META-INF/org/apache/camel/component/slack/slack.json
@@ -36,7 +36,7 @@
"channel": { "index": 0, "kind": "path", "displayName": "Channel",
"group": "common", "label": "", "required": true, "type": "string", "javaType":
"java.lang.String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The channel name (syntax #name) or
slack user (syntax userName) to send a message directly to an user." },
"token": { "index": 1, "kind": "parameter", "displayName": "Token",
"group": "common", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": true, "security": "secret", "description": "The token to access
Slack. This app needs to have channels:history, groups:history, im:history,
mpim:history, channels:read, groups:read, im:read and mpim:read permissions.
The User OAuth Token is the kind of token needed." },
"conversationType": { "index": 2, "kind": "parameter", "displayName":
"Conversation Type", "group": "consumer", "label": "consumer", "required":
false, "type": "enum", "javaType": "com.slack.api.model.ConversationType",
"enum": [ "PUBLIC_CHANNEL", "PRIVATE_CHANNEL", "MPIM", "IM" ], "deprecated":
false, "autowired": false, "secret": false, "defaultValue": "PUBLIC_CHANNEL",
"description": "Type of conversation" },
- "maxResults": { "index": 3, "kind": "parameter", "displayName": "Max
Results", "group": "consumer", "label": "consumer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": "10", "description": "The Max Result
for the poll" },
+ "maxResults": { "index": 3, "kind": "parameter", "displayName": "Max
Results", "group": "consumer", "label": "consumer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": "10", "description": "The maximum
number of messages that the consumer requests from the conversation history at
a time. When more new messages are available, the consumer reads the following
pages in the same poll." },
"naturalOrder": { "index": 4, "kind": "parameter", "displayName": "Natural
Order", "group": "consumer", "label": "consumer", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": false, "description": "Create exchanges in
natural order (oldest to newest) or not" },
"sendEmptyMessageWhenIdle": { "index": 5, "kind": "parameter",
"displayName": "Send Empty Message When Idle", "group": "consumer", "label":
"consumer", "required": false, "type": "boolean", "javaType": "boolean",
"deprecated": false, "autowired": false, "secret": false, "defaultValue":
false, "description": "If the polling consumer did not poll any files, you can
enable this option to send an empty message (no body) instead." },
"serverUrl": { "index": 6, "kind": "parameter", "displayName": "Server
Url", "group": "consumer", "label": "consumer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": "https:\/\/slack.com", "description":
"The Server URL of the Slack instance" },
diff --git a/components/camel-slack/src/main/docs/slack-component.adoc
b/components/camel-slack/src/main/docs/slack-component.adoc
index 4f7350e13eab..b0708b1d502b 100644
--- a/components/camel-slack/src/main/docs/slack-component.adoc
+++ b/components/camel-slack/src/main/docs/slack-component.adoc
@@ -188,6 +188,12 @@ YAML::
This way you'll get the last message from `general` channel.
The consumer will track the timestamp of the last message consumed,
and in the next poll it will consume only newer messages in the channel.
+When more than `maxResults` new messages were posted since the previous poll,
the consumer reads the following pages of
+the conversation history in the same poll (`maxResults` messages per request),
so no message is skipped. For a busy
+channel, set a higher `maxResults` to need fewer requests. If the request for
a following page fails, for example
+because Slack rate limits it (some apps may send only one
`conversations.history` request per minute), the consumer
+routes the most recent messages it has already read, as when it reads one page
only, and logs a WARN that the older
+messages of that poll were skipped.
You'll need to create a Slack app and use it in your workspace.
diff --git
a/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackConsumer.java
b/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackConsumer.java
index c0d0f6fce121..216534e61dd5 100644
---
a/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackConsumer.java
+++
b/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackConsumer.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.slack;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
@@ -39,11 +40,15 @@ import org.apache.camel.component.slack.helper.SlackHelper;
import org.apache.camel.support.ScheduledBatchPollingConsumer;
import org.apache.camel.util.CastUtils;
import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class SlackConsumer extends ScheduledBatchPollingConsumer {
public static final long DEFAULT_CONSUMER_DELAY = 10 * 1000L;
+ private static final Logger LOG =
LoggerFactory.getLogger(SlackConsumer.class);
+
private static final int CONVERSATIONS_LIST_LIMIT = 200;
private final SlackEndpoint slackEndpoint;
private Slack slack;
@@ -77,20 +82,64 @@ public class SlackConsumer extends
ScheduledBatchPollingConsumer {
// Maximum limit is 1000. Slack recommends no more than 200 results at
a time.
// https://api.slack.com/methods/conversations.history
// We set the limit to 1 the first call to set the timestamp of the
last message of the history
+ ConversationsHistoryResponse response = conversationsHistory(null);
+
+ // okay we have some response from slack so lets mark the consumer as
ready
+ forceConsumerAsReady();
+
+ List<Message> messages = response.getMessages();
+ if (timestamp != null && response.isHasMore()) {
+ // the most recent messages of the range come first, so the
messages that did not fit into this response
+ // are older ones: read the following pages, or they are skipped
when the timestamp moves past them
+ messages = messages != null ? new ArrayList<>(messages) : new
ArrayList<>();
+ String cursor = nextCursor(response);
+ while (cursor != null) {
+ if (!isPollAllowed()) {
+ // stopping or suspended: do not keep calling Slack. The
timestamp is not moved and nothing of this
+ // poll is routed, so the first poll after a restart or
resume reads these messages again; routing
+ // the pages read so far would move the timestamp past the
older messages that were not read
+ LOG.debug("Stopped reading the conversation history of
channel {} as the consumer is stopping or"
+ + " suspended",
+ slackEndpoint.getChannel());
+ return 0;
+ }
+ try {
+ response = conversationsHistory(cursor);
+ } catch (IOException | SlackApiException |
RuntimeCamelException e) {
+ // for example rate limited: route the most recent
messages already read, as when only one page was
+ // read, instead of failing every poll
+ LOG.warn("Cannot read the following page of the
conversation history of channel {}: {}."
+ + " Routing the {} most recent new messages, the
older messages of this poll are skipped.",
+ slackEndpoint.getChannel(), e.getMessage(),
messages.size());
+ break;
+ }
+ if (response.getMessages() != null) {
+ messages.addAll(response.getMessages());
+ }
+ cursor = response.isHasMore() ? nextCursor(response) : null;
+ }
+ }
+
+ Queue<Exchange> exchanges = createExchanges(messages);
+ return processBatch(CastUtils.cast(exchanges));
+ }
+
+ private ConversationsHistoryResponse conversationsHistory(String cursor)
throws IOException, SlackApiException {
ConversationsHistoryResponse response =
slack.methods(slackEndpoint.getToken()).conversationsHistory(req -> req
.channel(channelId)
.oldest(timestamp)
+ .cursor(cursor)
.limit(timestamp != null ?
Integer.parseInt(slackEndpoint.getMaxResults()) : 1));
if (!response.isOk()) {
throw new RuntimeCamelException("API request conversations.history
to Slack failed: " + response);
}
+ return response;
+ }
- // okay we have some response from slack so lets mark the consumer as
ready
- forceConsumerAsReady();
-
- Queue<Exchange> exchanges = createExchanges(response.getMessages());
- return processBatch(CastUtils.cast(exchanges));
+ private static String nextCursor(ConversationsHistoryResponse response) {
+ String cursor = response.getResponseMetadata() != null ?
response.getResponseMetadata().getNextCursor() : null;
+ return ObjectHelper.isNotEmpty(cursor) ? cursor : null;
}
private Queue<Exchange> createExchanges(final List<Message> list) {
diff --git
a/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackEndpoint.java
b/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackEndpoint.java
index f9ff2e37d243..fb94edc48f68 100644
---
a/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackEndpoint.java
+++
b/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackEndpoint.java
@@ -182,7 +182,8 @@ public class SlackEndpoint extends ScheduledPollEndpoint {
}
/**
- * The Max Result for the poll
+ * The maximum number of messages that the consumer requests from the
conversation history at a time. When more new
+ * messages are available, the consumer reads the following pages in the
same poll.
*/
public void setMaxResults(String maxResult) {
this.maxResults = maxResult;
diff --git
a/components/camel-slack/src/test/java/org/apache/camel/component/slack/SlackConsumerHasMoreTest.java
b/components/camel-slack/src/test/java/org/apache/camel/component/slack/SlackConsumerHasMoreTest.java
new file mode 100644
index 000000000000..adbe579dcca8
--- /dev/null
+++
b/components/camel-slack/src/test/java/org/apache/camel/component/slack/SlackConsumerHasMoreTest.java
@@ -0,0 +1,292 @@
+/*
+ * 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.slack;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.math.BigDecimal;
+import java.net.InetSocketAddress;
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+import com.slack.api.model.Message;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.PollingConsumerPollStrategy;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The consumer must not skip messages when more new messages than {@code
maxResults} were posted between two polls:
+ * conversations.history returns the most recent messages of the range first,
and the older ones on the following pages
+ * (has_more and response_metadata.next_cursor).
+ */
+class SlackConsumerHasMoreTest extends CamelTestSupport {
+
+ private final List<String> channel = new CopyOnWriteArrayList<>();
+ private HttpServer server;
+ private final AtomicInteger rateLimitedRequests = new AtomicInteger();
+ private boolean naturalOrder;
+ private volatile boolean rateLimitFollowingPages;
+ private final List<String> cursors = new CopyOnWriteArrayList<>();
+ private final AtomicReference<Runnable> beforeNextFollowingPage = new
AtomicReference<>();
+ private final CountDownLatch suspendedPollDone = new CountDownLatch(1);
+ private volatile boolean suspendedWhilePaging;
+
+ @Override
+ protected void doPreSetup() throws Exception {
+ server = HttpServer.create(new InetSocketAddress("localhost", 0), 0);
+ server.createContext("/api/conversations.list",
this::conversationsList);
+ server.createContext("/api/conversations.history",
this::conversationsHistory);
+ server.start();
+ }
+
+ @AfterEach
+ void stopServer() {
+ // stop polling before the server goes away
+ context.stop();
+ if (server != null) {
+ server.stop(0);
+ }
+ }
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ @Test
+ void consumesEveryMessageOfABurstLargerThanMaxResults() throws Exception {
+ MockEndpoint mock = startRouteAndReceiveFirstMessage();
+
+ // five messages are posted before the next poll, more than
maxResults=2
+ postBurst();
+
+ mock.expectedMessageCount(5);
+ mock.setAssertPeriod(500);
+ mock.assertIsSatisfied();
+ assertEquals(List.of("m6", "m5", "m4", "m3", "m2"), texts(mock),
+ "every new message must be routed once, the most recent
first");
+ }
+
+ @Test
+ void consumesEveryMessageOfABurstInNaturalOrder() throws Exception {
+ naturalOrder = true;
+ MockEndpoint mock = startRouteAndReceiveFirstMessage();
+
+ postBurst();
+
+ mock.expectedMessageCount(5);
+ mock.setAssertPeriod(500);
+ mock.assertIsSatisfied();
+ assertEquals(List.of("m2", "m3", "m4", "m5", "m6"), texts(mock),
+ "every new message must be routed once, the oldest first");
+ }
+
+ @Test
+ void routesTheMostRecentMessagesWhenAFollowingPageIsRateLimited() throws
Exception {
+ // apps that Slack limits to one conversations.history request per
minute get HTTP 429 for the second page
+ rateLimitFollowingPages = true;
+ MockEndpoint mock = startRouteAndReceiveFirstMessage();
+
+ postBurst();
+
+ // as when only one page is read: the most recent maxResults messages
are routed, once, and the poll does not
+ // fail (a failed poll would read the whole range again at every
following poll)
+ mock.expectedMessageCount(2);
+ mock.setAssertPeriod(500);
+ mock.assertIsSatisfied();
+ assertEquals(List.of("m6", "m5"), texts(mock));
+ assertTrue(rateLimitedRequests.get() > 0, "the following page must
have been requested");
+ }
+
+ @Test
+ void stopsPagingWhenSuspendedAndReadsTheMessagesAgainAfterResume() throws
Exception {
+ MockEndpoint mock = startRouteAndReceiveFirstMessage();
+ SlackConsumer consumer = (SlackConsumer)
context.getRoute("slack").getConsumer();
+ // the consumer is suspended while it waits for the second of the
three pages
+ beforeNextFollowingPage.set(() -> {
+ consumer.suspend();
+ suspendedWhilePaging = true;
+ });
+
+ postBurst();
+
+ assertTrue(suspendedPollDone.await(20, TimeUnit.SECONDS), "the poll
that was reading the pages did not end");
+ assertEquals(List.of("2"), cursors, "the third page must not be
requested once the consumer is suspended");
+ assertEquals(0, mock.getReceivedCounter(), "a part of the burst was
routed, so its older messages are skipped");
+
+ mock.expectedMessageCount(5);
+ mock.setAssertPeriod(500);
+ consumer.resume();
+
+ mock.assertIsSatisfied();
+ assertEquals(List.of("m6", "m5", "m4", "m3", "m2"), texts(mock),
+ "every message of the burst must be routed once after the
resume");
+ assertEquals(List.of("2", "2", "4"), cursors);
+ }
+
+ private MockEndpoint startRouteAndReceiveFirstMessage() throws Exception {
+ channel.add("m1");
+ // tells when the poll during which the route was suspended has ended
+ context.getRegistry().bind("pollEnded", new
PollingConsumerPollStrategy() {
+ @Override
+ public boolean begin(Consumer consumer, Endpoint endpoint) {
+ return true;
+ }
+
+ @Override
+ public void commit(Consumer consumer, Endpoint endpoint, int
polledMessages) {
+ if (suspendedWhilePaging) {
+ suspendedPollDone.countDown();
+ }
+ }
+
+ @Override
+ public boolean rollback(Consumer consumer, Endpoint endpoint, int
retryCounter, Exception e) {
+ return false;
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
fromF("slack://general?token=RAW(xoxp-test)&serverUrl=http://localhost:%d&maxResults=2&delay=100";
+ + "&naturalOrder=%s&pollStrategy=#pollEnded",
+ server.getAddress().getPort(), naturalOrder)
+ .routeId("slack")
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ // the first poll reads the last message of the history
+ mock.expectedMessageCount(1);
+ mock.assertIsSatisfied();
+ assertEquals(List.of("m1"), texts(mock));
+ mock.reset();
+ return mock;
+ }
+
+ private void postBurst() {
+ // added at once, so that no poll sees only a part of the burst
+ channel.addAll(List.of("m2", "m3", "m4", "m5", "m6"));
+ }
+
+ private static List<String> texts(MockEndpoint mock) {
+ List<String> answer = new ArrayList<>();
+ for (Exchange exchange : mock.getReceivedExchanges()) {
+ answer.add(exchange.getMessage().getBody(Message.class).getText());
+ }
+ return answer;
+ }
+
+ private void conversationsList(HttpExchange http) throws IOException {
+ respond(http,
"{\"ok\":true,\"channels\":[{\"id\":\"C1\",\"name\":\"general\"}],"
+ + "\"response_metadata\":{\"next_cursor\":\"\"}}");
+ }
+
+ /**
+ * Answers as Slack does: the messages after {@code oldest} (exclusive),
the most recent first, {@code limit} at a
+ * time, with has_more and a next_cursor when there are more.
+ */
+ private void conversationsHistory(HttpExchange http) throws IOException {
+ Map<String, String> params = form(http);
+ String oldest = params.get("oldest");
+ int limit = Integer.parseInt(params.get("limit"));
+ if (rateLimitFollowingPages && params.get("cursor") != null) {
+ rateLimitedRequests.incrementAndGet();
+ http.getResponseHeaders().add("Retry-After", "60");
+ respond(http, 429, "{\"ok\":false,\"error\":\"ratelimited\"}");
+ return;
+ }
+ if (params.get("cursor") != null) {
+ cursors.add(params.get("cursor"));
+ Runnable task = beforeNextFollowingPage.getAndSet(null);
+ if (task != null) {
+ task.run();
+ }
+ }
+ int offset = params.get("cursor") != null ?
Integer.parseInt(params.get("cursor")) : 0;
+
+ List<Integer> range = new ArrayList<>();
+ for (int i = 0; i < channel.size(); i++) {
+ if (oldest == null || new BigDecimal(ts(i)).compareTo(new
BigDecimal(oldest)) > 0) {
+ range.add(i);
+ }
+ }
+ range.sort(Comparator.reverseOrder());
+ List<Integer> page = range.subList(Math.min(offset, range.size()),
Math.min(offset + limit, range.size()));
+ boolean hasMore = offset + limit < range.size();
+
+ String messages = page.stream()
+ .map(i -> "{\"type\":\"message\",\"text\":\"" + channel.get(i)
+ "\",\"ts\":\"" + ts(i) + "\"}")
+ .collect(Collectors.joining(","));
+ respond(http, "{\"ok\":true,\"messages\":[" + messages +
"],\"has_more\":" + hasMore
+ + ",\"response_metadata\":{\"next_cursor\":\"" +
(hasMore ? offset + limit : "") + "\"}}");
+ }
+
+ private static String ts(int index) {
+ return (1700000000 + index) + ".000100";
+ }
+
+ private static Map<String, String> form(HttpExchange http) throws
IOException {
+ Map<String, String> answer = new HashMap<>();
+ String body = new String(http.getRequestBody().readAllBytes(),
StandardCharsets.UTF_8);
+ for (String pair : body.split("&")) {
+ int idx = pair.indexOf('=');
+ if (idx > 0) {
+ answer.put(URLDecoder.decode(pair.substring(0, idx),
StandardCharsets.UTF_8),
+ URLDecoder.decode(pair.substring(idx + 1),
StandardCharsets.UTF_8));
+ }
+ }
+ return answer;
+ }
+
+ private static void respond(HttpExchange http, String json) throws
IOException {
+ respond(http, 200, json);
+ }
+
+ private static void respond(HttpExchange http, int status, String json)
throws IOException {
+ byte[] bytes = json.getBytes(StandardCharsets.UTF_8);
+ http.getResponseHeaders().add("Content-Type", "application/json;
charset=utf-8");
+ http.sendResponseHeaders(status, bytes.length);
+ try (OutputStream os = http.getResponseBody()) {
+ os.write(bytes);
+ }
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index ef55a568c050..ff9c0d0a3386 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -1163,6 +1163,15 @@ the message body as if the route had succeeded, which
usually echoed the request
body now replies with an empty (`null`) body instead of not replying, so the
sender no longer waits until its reply timeout (30 seconds by default).
+=== camel-slack - the consumer reads every new message
+
+When more messages than `maxResults` (default 10) were posted to the channel
since the previous poll, the consumer
+routed only the most recent `maxResults` of them and skipped the older ones.
It now reads the following pages of the
+conversation history in the same poll, so a poll can route more than
`maxResults` messages, with one request per
+`maxResults` messages. For a busy channel, raise `maxResults` to keep the
number of requests within the Slack rate
+limits. If the request for a following page fails (for example rate limited),
the consumer routes the most recent
+messages it has already read, as before, and logs a WARN that the older
messages of that poll were skipped.
+
=== camel-rocketmq - a message whose route failed is consumed again
The consumer acknowledged a message (`CONSUME_SUCCESS`) whenever the processor
returned, also when the route failed:
diff --git
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/SlackEndpointBuilderFactory.java
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/SlackEndpointBuilderFactory.java
index a0db5f36cbcf..fb820d61dfc1 100644
---
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/SlackEndpointBuilderFactory.java
+++
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/SlackEndpointBuilderFactory.java
@@ -93,7 +93,9 @@ public interface SlackEndpointBuilderFactory {
return this;
}
/**
- * The Max Result for the poll.
+ * The maximum number of messages that the consumer requests from the
+ * conversation history at a time. When more new messages are
available,
+ * the consumer reads the following pages in the same poll.
*
* The option is a: <code>java.lang.String</code> type.
*
