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 0f95633e7e7f CAMEL-25576: camel-box - getEnterpriseEvents reads every
chunk of the event log (#27760)
0f95633e7e7f is described below
commit 0f95633e7e7fe6db4bc42d992688a3baa60e5f02
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Oct 13 12:25:45 2026 +0530
CAMEL-25576: camel-box - getEnterpriseEvents reads every chunk of the event
log (#27760)
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../component/box/api/BoxEventLogsManager.java | 29 +++--
.../box/BoxEventLogsManagerPagingTest.java | 140 +++++++++++++++++++++
2 files changed, 162 insertions(+), 7 deletions(-)
diff --git
a/components/camel-box/camel-box-api/src/main/java/org/apache/camel/component/box/api/BoxEventLogsManager.java
b/components/camel-box/camel-box-api/src/main/java/org/apache/camel/component/box/api/BoxEventLogsManager.java
index 094e42585b0f..c9ec9ddcc2c1 100644
---
a/components/camel-box/camel-box-api/src/main/java/org/apache/camel/component/box/api/BoxEventLogsManager.java
+++
b/components/camel-box/camel-box-api/src/main/java/org/apache/camel/component/box/api/BoxEventLogsManager.java
@@ -19,7 +19,9 @@ package org.apache.camel.component.box.api;
import java.text.DateFormat;
import java.util.ArrayList;
import java.util.Date;
+import java.util.LinkedHashSet;
import java.util.List;
+import java.util.Set;
import com.box.sdk.BoxAPIConnection;
import com.box.sdk.BoxAPIException;
@@ -63,7 +65,9 @@ public class BoxEventLogsManager {
* @param before - the upper bound on the timestamp of the events
returned.
* @param types - an optional list of event types to filter by.
*
- * @return A list of all the events that met the given criteria.
+ * @return A list of all the events that met the given criteria.
The event log is read chunk by chunk (at
+ * most 500 events per request) until its end, and all
the events are kept in memory, so use a
+ * narrow time window for a large event log.
*/
public List<BoxEvent> getEnterpriseEvents(String position, Date after,
Date before, BoxEvent.EventType... types) {
try {
@@ -80,15 +84,26 @@ public class BoxEventLogsManager {
}
EnterpriseEventsRequest request = new EnterpriseEventsRequest();
-
request.position(position).after(after).before(before).types(types);
- EventLog eventLog = EventLog.getEnterpriseEvents(boxConnection,
request);
+ request.after(after).before(before).types(types);
- List<BoxEvent> results = new ArrayList<>();
- for (BoxEvent event : eventLog) {
- results.add(event);
+ // The service returns the events in chunks (of at most 500
events) with the position of the next
+ // chunk. Read on from that position until a chunk is empty, which
is the end of the events within
+ // bounds. An event that the service returns in two chunks is kept
once.
+ Set<BoxEvent> results = new LinkedHashSet<>();
+ String streamPosition = position;
+ while (true) {
+ EventLog eventLog =
EventLog.getEnterpriseEvents(boxConnection, request.position(streamPosition));
+ for (BoxEvent event : eventLog) {
+ results.add(event);
+ }
+ String nextStreamPosition = eventLog.getNextStreamPosition();
+ if (eventLog.getSize() == 0 || nextStreamPosition == null ||
nextStreamPosition.equals(streamPosition)) {
+ break;
+ }
+ streamPosition = nextStreamPosition;
}
- return results;
+ return new ArrayList<>(results);
} catch (BoxAPIException e) {
throw new RuntimeCamelException(
buildBoxApiErrorMessage(e), e);
diff --git
a/components/camel-box/camel-box-component/src/test/java/org/apache/camel/component/box/BoxEventLogsManagerPagingTest.java
b/components/camel-box/camel-box-component/src/test/java/org/apache/camel/component/box/BoxEventLogsManagerPagingTest.java
new file mode 100644
index 000000000000..f4f2595f9898
--- /dev/null
+++
b/components/camel-box/camel-box-component/src/test/java/org/apache/camel/component/box/BoxEventLogsManagerPagingTest.java
@@ -0,0 +1,140 @@
+/*
+ * 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.box;
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+
+import com.box.sdk.BoxAPIConnection;
+import com.box.sdk.BoxAPIRequest;
+import com.box.sdk.BoxAPIResponse;
+import com.box.sdk.BoxEvent;
+import com.box.sdk.BoxJSONResponse;
+import com.eclipsesource.json.JsonArray;
+import com.eclipsesource.json.JsonObject;
+import org.apache.camel.component.box.api.BoxEventLogsManager;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * The Box connection answers the enterprise events request from memory, as
the service does: at most {@link #MAX_CHUNK}
+ * events from the requested stream position, and the position of the next
chunk.
+ */
+class BoxEventLogsManagerPagingTest {
+
+ private static final int MAX_CHUNK = 500;
+ private static final int MAX_REQUESTS = 20;
+ private static final Date AFTER = new Date(0);
+ private static final Date BEFORE = new Date(86_400_000L);
+
+ /** The stream positions of the requests, in order ({@code null}: a
request without position). */
+ private final List<String> positions = new ArrayList<>();
+ private int events;
+ private boolean samePositionAgain;
+ private BoxEventLogsManager manager;
+
+ @BeforeEach
+ void setUp() {
+ BoxAPIConnection connection = new BoxAPIConnection("test-token");
+ connection.setRequestInterceptor(this::respond);
+ manager = new BoxEventLogsManager(connection);
+ }
+
+ private BoxAPIResponse respond(BoxAPIRequest request) {
+ String position = null;
+ int limit = MAX_CHUNK;
+ for (String parameter : request.getUrl().getQuery().split("&")) {
+ if (parameter.startsWith("stream_position=")) {
+ position = parameter.substring("stream_position=".length());
+ } else if (parameter.startsWith("limit=")) {
+ limit = Math.min(MAX_CHUNK,
Integer.parseInt(parameter.substring("limit=".length())));
+ }
+ }
+ positions.add(position);
+ if (positions.size() > MAX_REQUESTS) {
+ throw new IllegalStateException("The manager sent more than " +
MAX_REQUESTS + " requests");
+ }
+ int from = position == null ? 0 : Integer.parseInt(position);
+ int to = Math.min(from + limit, events);
+ JsonArray entries = new JsonArray();
+ for (int i = from; i < to; i++) {
+ entries.add(new JsonObject().add("type", "event").add("event_id",
"event-" + i)
+ .add("event_type", "LOGIN").add("created_at",
"2026-01-01T00:00:00-00:00"));
+ }
+ JsonObject body = new JsonObject()
+ .add("chunk_size", entries.size())
+ .add("next_stream_position", samePositionAgain ?
String.valueOf(from) : String.valueOf(to))
+ .add("entries", entries);
+ return new BoxJSONResponse(200, request.getMethod(),
request.getUrl().toString(), Map.of(), body);
+ }
+
+ private static List<String> ids(List<BoxEvent> result) {
+ List<String> answer = new ArrayList<>();
+ for (BoxEvent event : result) {
+ answer.add(event.getID());
+ }
+ return answer;
+ }
+
+ private static List<String> expectedIds(int from, int to) {
+ List<String> answer = new ArrayList<>();
+ for (int i = from; i < to; i++) {
+ answer.add("event-" + i);
+ }
+ return answer;
+ }
+
+ @Test
+ void returnsTheEventsOfEveryChunk() {
+ events = 2 * MAX_CHUNK + 200;
+
+ List<BoxEvent> result = manager.getEnterpriseEvents(null, AFTER,
BEFORE);
+
+ assertEquals(events, result.size(), "all the events within bounds, not
only the first chunk");
+ assertEquals(expectedIds(0, events), ids(result), "each event once and
in the order of the service");
+ List<String> expectedPositions = new ArrayList<>();
+ expectedPositions.add(null);
+ expectedPositions.addAll(List.of("500", "1000", "1200"));
+ assertEquals(expectedPositions, positions, "each chunk is requested at
the next position of the chunk before");
+ }
+
+ @Test
+ void startsAtThePositionOfTheCallerAndReadsToTheEnd() {
+ events = 2 * MAX_CHUNK + 200;
+
+ List<BoxEvent> result = manager.getEnterpriseEvents("100", AFTER,
BEFORE);
+
+ assertEquals(expectedIds(100, events), ids(result));
+ assertEquals(List.of("100", "600", "1100", "1200"), positions);
+ }
+
+ @Test
+ void stopsWhenTheServiceDoesNotMoveThePosition() {
+ // a service that answers with the position it was asked for must not
be asked for ever
+ events = 2;
+ samePositionAgain = true;
+
+ List<BoxEvent> result = manager.getEnterpriseEvents(null, AFTER,
BEFORE);
+
+ assertEquals(expectedIds(0, 2), ids(result), "the two events once,
although both responses contain them");
+ assertEquals(2, positions.size());
+ }
+}
