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 001ade0c3090 CAMEL-25609: camel-browse - Release the exchanges that
browseLimit drops (#27796)
001ade0c3090 is described below
commit 001ade0c3090821cfdcc58f9c8312181735c6821
Author: allthingssecurity <[email protected]>
AuthorDate: Thu Oct 15 11:18:46 2026 +0530
CAMEL-25609: camel-browse - Release the exchanges that browseLimit drops
(#27796)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/component/browse/BrowseEndpoint.java | 23 +-
.../org/apache/camel/spi/BrowsableEndpoint.java | 20 +-
.../component/browse/BrowseLimitMemoryTest.java | 302 +++++++++++++++++++++
.../management/mbean/ManagedBrowsableEndpoint.java | 50 ++--
.../ManagedBrowsableEndpointSnapshotTest.java | 174 ++++++++++++
5 files changed, 536 insertions(+), 33 deletions(-)
diff --git
a/components/camel-browse/src/main/java/org/apache/camel/component/browse/BrowseEndpoint.java
b/components/camel-browse/src/main/java/org/apache/camel/component/browse/BrowseEndpoint.java
index 5462f9dae33f..35e842ba0eee 100644
---
a/components/camel-browse/src/main/java/org/apache/camel/component/browse/BrowseEndpoint.java
+++
b/components/camel-browse/src/main/java/org/apache/camel/component/browse/BrowseEndpoint.java
@@ -18,6 +18,8 @@ package org.apache.camel.component.browse;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Predicate;
import org.apache.camel.Category;
@@ -55,6 +57,7 @@ public class BrowseEndpoint extends DefaultEndpoint
implements BrowsableEndpoint
description = "To use a predicate to filter whether to include
the message for browsing. Return true to include, false to exclude.")
private Predicate<Exchange> filter;
+ private final Lock limitLock = new ReentrantLock();
private List<Exchange> exchanges;
private volatile Processor onExchangeProcessor;
@@ -140,14 +143,22 @@ public class BrowseEndpoint extends DefaultEndpoint
implements BrowsableEndpoint
*/
protected void onExchange(Exchange exchange) throws Exception {
if (browseLimit > 0) {
- // make room if end of capacity
- int max = exchanges.size();
- if (max >= browseLimit) {
- int space = Math.abs(max - (browseLimit - 1));
- exchanges = exchanges.subList(space, max);
+ // add first and then remove the oldest from the list itself: a
sub list is only a view that keeps them
+ // all in memory, and a reader never sees fewer than the limit
once it was reached. Under a lock, as
+ // concurrent producers would otherwise remove too many or
overtake the limit
+ limitLock.lock();
+ try {
+ exchanges.add(exchange);
+ int excess = exchanges.size() - browseLimit;
+ if (excess > 0) {
+ exchanges.subList(0, excess).clear();
+ }
+ } finally {
+ limitLock.unlock();
}
+ } else {
+ exchanges.add(exchange);
}
- exchanges.add(exchange);
// now fire the consumer
if (onExchangeProcessor != null) {
diff --git
a/core/camel-api/src/main/java/org/apache/camel/spi/BrowsableEndpoint.java
b/core/camel-api/src/main/java/org/apache/camel/spi/BrowsableEndpoint.java
index d65845200dd9..2420f2526543 100644
--- a/core/camel-api/src/main/java/org/apache/camel/spi/BrowsableEndpoint.java
+++ b/core/camel-api/src/main/java/org/apache/camel/spi/BrowsableEndpoint.java
@@ -57,16 +57,26 @@ public interface BrowsableEndpoint extends Endpoint {
* @return the status
*/
default BrowseStatus getBrowseStatus(int limit) {
- List<Exchange> list = getExchanges();
+ // iterate once, so size, first and last come from the same snapshot
(the iterator of a copy-on-write list), as
+ // the list can change between size() and get(index), such as a browse
limit removing the oldest exchange
+ int size = 0;
+ Exchange first = null;
+ Exchange last = null;
+ for (Exchange exchange : getExchanges()) {
+ if (size++ == 0) {
+ first = exchange;
+ }
+ last = exchange;
+ }
long ts = 0;
long ts2 = 0;
- if (!list.isEmpty()) {
- Long v1 =
list.get(0).getMessage().getHeader(Exchange.MESSAGE_TIMESTAMP, 0L, long.class);
- Long v2 = list.get(list.size() -
1).getMessage().getHeader(Exchange.MESSAGE_TIMESTAMP, 0L, long.class);
+ if (size > 0) {
+ Long v1 = first.getMessage().getHeader(Exchange.MESSAGE_TIMESTAMP,
0L, long.class);
+ Long v2 = last.getMessage().getHeader(Exchange.MESSAGE_TIMESTAMP,
0L, long.class);
ts = v1 != null ? v1 : 0L;
ts2 = v2 != null ? v2 : 0L;
}
- return new BrowseStatus(list.size(), ts, ts2);
+ return new BrowseStatus(size, ts, ts2);
}
/**
diff --git
a/core/camel-core/src/test/java/org/apache/camel/component/browse/BrowseLimitMemoryTest.java
b/core/camel-core/src/test/java/org/apache/camel/component/browse/BrowseLimitMemoryTest.java
new file mode 100644
index 000000000000..841551b3af7f
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/component/browse/BrowseLimitMemoryTest.java
@@ -0,0 +1,302 @@
+/*
+ * 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.browse;
+
+import java.util.AbstractList;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Producer;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * With browseLimit the endpoint must keep at most that many exchanges in
memory, a list handed out to a browser must
+ * stay usable when more exchanges arrive, and producers that send at the same
time must not break the endpoint.
+ */
+public class BrowseLimitMemoryTest extends ContextTestSupport {
+
+ private static final int LIMIT = 10;
+ private static final int TOTAL = 1000;
+
+ @Test
+ public void testOldExchangesAreReleased() throws Exception {
+ AtomicReference<List<Exchange>> store = new AtomicReference<>();
+ BrowseEndpoint endpoint = new BrowseEndpoint("browse:store",
context.getComponent("browse")) {
+ @Override
+ protected List<Exchange> createExchangeList() {
+ // the list that holds the exchanges in memory
+ List<Exchange> answer = super.createExchangeList();
+ store.set(answer);
+ return answer;
+ }
+ };
+ endpoint.setName("store");
+ endpoint.setBrowseLimit(LIMIT);
+ context.addEndpoint("browse:store", endpoint);
+
+ for (int i = 0; i < TOTAL; i++) {
+ template.sendBody(endpoint, "msg-" + i);
+ }
+
+ // what a browser sees: the last ten
+ List<Exchange> browsed = endpoint.getExchanges();
+ assertEquals(LIMIT, browsed.size());
+ assertEquals("msg-" + (TOTAL - LIMIT),
browsed.get(0).getMessage().getBody());
+ assertEquals("msg-" + (TOTAL - 1), browsed.get(LIMIT -
1).getMessage().getBody());
+
+ // what is kept in memory: not more than the limit
+ assertEquals(LIMIT, store.get().size(), "exchanges kept in memory with
browseLimit=" + LIMIT);
+ }
+
+ @Test
+ public void testBrowsedListStaysUsableWhenMoreExchangesArrive() throws
Exception {
+ String uri = "browse:live?browseLimit=2";
+ template.sendBody(uri, "one");
+ template.sendBody(uri, "two");
+ template.sendBody(uri, "three");
+
+ BrowseEndpoint endpoint = context.getEndpoint(uri,
BrowseEndpoint.class);
+ List<Exchange> browsed = endpoint.getExchanges();
+ assertEquals(2, browsed.size());
+
+ // a new exchange arrives while the browser still holds the list
+ template.sendBody(uri, "four");
+
+ assertEquals(2, browsed.size());
+ assertEquals("three", browsed.get(0).getMessage().getBody());
+ assertEquals("four", browsed.get(1).getMessage().getBody());
+ }
+
+ @Test
+ public void testSecondProducerWhileFirstMakesRoom() throws Exception {
+ // producer A is held at the moment it asks the list for a sub list
(the step that makes room), and producer B
+ // sends meanwhile: B either completes, or waits for A when the
endpoint serializes the two
+ Queue<String> failures = new ConcurrentLinkedQueue<>();
+ AtomicReference<Thread> second = new AtomicReference<>();
+ CountDownLatch secondDone = new CountDownLatch(1);
+ AtomicReference<Runnable> onSubList = new AtomicReference<>();
+
+ BrowseEndpoint endpoint = hookedEndpoint("pair", 2, null, onSubList);
+ Producer producer = endpoint.createProducer();
+ producer.process(exchange(endpoint, "one"));
+ producer.process(exchange(endpoint, "two"));
+
+ Exchange a = exchange(endpoint, "A");
+ Exchange b = exchange(endpoint, "B");
+ Exchange c = exchange(endpoint, "C");
+ second.set(new Thread(() -> {
+ try {
+ producer.process(b);
+ } catch (Throwable e) {
+ failures.add("producer B: " + e);
+ } finally {
+ secondDone.countDown();
+ }
+ }, "browse-producer-B"));
+
+ onSubList.set(() -> {
+ // only the first sub list is held
+ onSubList.set(null);
+ Thread t = second.get();
+ t.start();
+ await().atMost(20, TimeUnit.SECONDS)
+ .until(() -> secondDone.getCount() == 0 || t.getState() ==
Thread.State.WAITING);
+ });
+ try {
+ producer.process(a);
+ } catch (Throwable e) {
+ failures.add("producer A: " + e);
+ }
+ assertTrue(secondDone.await(20, TimeUnit.SECONDS), "producer B did not
complete");
+
+ // the endpoint must still accept exchanges and be readable afterwards
+ try {
+ producer.process(c);
+ } catch (Throwable e) {
+ failures.add("producer C, after A and B: " + e);
+ }
+ try {
+ assertEquals(2, endpoint.getExchanges().size());
+ assertEquals("C",
endpoint.getExchanges().get(1).getMessage().getBody());
+ } catch (Throwable e) {
+ failures.add("browsing, after A, B and C: " + e);
+ }
+ assertTrue(failures.isEmpty(), "failures: " + failures);
+ }
+
+ @Test
+ public void testManyProducerThreads() throws Exception {
+ int threads = 4;
+ int perThread = 250;
+ AtomicReference<List<Exchange>> store = new AtomicReference<>();
+ BrowseEndpoint endpoint = new BrowseEndpoint("browse:many",
context.getComponent("browse")) {
+ @Override
+ protected List<Exchange> createExchangeList() {
+ List<Exchange> answer = super.createExchangeList();
+ store.set(answer);
+ return answer;
+ }
+ };
+ endpoint.setName("many");
+ endpoint.setBrowseLimit(LIMIT);
+ context.addEndpoint("browse:many", endpoint);
+ Producer producer = endpoint.createProducer();
+
+ Queue<String> failures = new ConcurrentLinkedQueue<>();
+ CountDownLatch go = new CountDownLatch(1);
+ List<Thread> producers = new ArrayList<>();
+ for (int t = 0; t < threads; t++) {
+ List<Exchange> mine = new ArrayList<>();
+ for (int i = 0; i < perThread; i++) {
+ mine.add(exchange(endpoint, t + "-" + i));
+ }
+ Thread thread = new Thread(() -> {
+ try {
+ if (!go.await(20, TimeUnit.SECONDS)) {
+ failures.add("not started");
+ return;
+ }
+ for (Exchange exchange : mine) {
+ producer.process(exchange);
+ }
+ } catch (Throwable e) {
+ failures.add(e.toString());
+ }
+ }, "browse-producer-" + t);
+ producers.add(thread);
+ thread.start();
+ }
+ go.countDown();
+ for (Thread thread : producers) {
+ thread.join(20000);
+ assertFalse(thread.isAlive(), thread.getName() + " did not
complete");
+ }
+
+ assertTrue(failures.isEmpty(), "failures: " + failures);
+ assertEquals(LIMIT, endpoint.getExchanges().size());
+ assertEquals(LIMIT, store.get().size(), "exchanges kept in memory with
browseLimit=" + LIMIT);
+ }
+
+ @Test
+ public void testReaderDoesNotSeeTheEndpointEmptyWithLimitOne() throws
Exception {
+ // a reader that looks while a producer is inside the list (before the
add, and before the oldest is removed)
+ List<Integer> seen = new ArrayList<>();
+ AtomicReference<Runnable> look = new AtomicReference<>();
+ BrowseEndpoint endpoint = hookedEndpoint("single", 1, look, look);
+ Producer producer = endpoint.createProducer();
+ producer.process(exchange(endpoint, "one"));
+
+ look.set(() -> seen.add(endpoint.getBrowseStatus(0).size()));
+ producer.process(exchange(endpoint, "two"));
+ producer.process(exchange(endpoint, "three"));
+ look.set(null);
+
+ assertFalse(seen.isEmpty());
+ assertFalse(seen.contains(0), "sizes seen by a reader while exchanges
arrive: " + seen);
+ assertEquals(1, endpoint.getExchanges().size());
+ assertEquals("three",
endpoint.getExchanges().get(0).getMessage().getBody());
+ }
+
+ private static Exchange exchange(BrowseEndpoint endpoint, String body) {
+ Exchange answer = endpoint.createExchange();
+ answer.getMessage().setBody(body);
+ return answer;
+ }
+
+ private BrowseEndpoint hookedEndpoint(
+ String name, int limit, AtomicReference<Runnable> beforeAdd,
AtomicReference<Runnable> afterSubList)
+ throws Exception {
+ BrowseEndpoint endpoint = new BrowseEndpoint("browse:" + name,
context.getComponent("browse")) {
+ @Override
+ protected List<Exchange> createExchangeList() {
+ return new HookedList(super.createExchangeList(), beforeAdd,
afterSubList);
+ }
+ };
+ endpoint.setName(name);
+ endpoint.setBrowseLimit(limit);
+ context.addEndpoint("browse:" + name, endpoint);
+ return endpoint;
+ }
+
+ /**
+ * Delegates to the list of the endpoint (and to its sub lists) and lets a
test run code just before an add and just
+ * after a sub list was created. It does not change what the list does.
+ */
+ private static final class HookedList extends AbstractList<Exchange> {
+ private final List<Exchange> delegate;
+ private final AtomicReference<Runnable> beforeAdd;
+ private final AtomicReference<Runnable> afterSubList;
+
+ HookedList(List<Exchange> delegate, AtomicReference<Runnable>
beforeAdd, AtomicReference<Runnable> afterSubList) {
+ this.delegate = delegate;
+ this.beforeAdd = beforeAdd;
+ this.afterSubList = afterSubList;
+ }
+
+ private static void run(AtomicReference<Runnable> hook) {
+ Runnable task = hook != null ? hook.get() : null;
+ if (task != null) {
+ task.run();
+ }
+ }
+
+ @Override
+ public Exchange get(int index) {
+ return delegate.get(index);
+ }
+
+ @Override
+ public int size() {
+ return delegate.size();
+ }
+
+ @Override
+ public boolean add(Exchange exchange) {
+ run(beforeAdd);
+ return delegate.add(exchange);
+ }
+
+ @Override
+ public Exchange remove(int index) {
+ return delegate.remove(index);
+ }
+
+ @Override
+ public void clear() {
+ delegate.clear();
+ }
+
+ @Override
+ public List<Exchange> subList(int fromIndex, int toIndex) {
+ List<Exchange> answer = new HookedList(delegate.subList(fromIndex,
toIndex), beforeAdd, afterSubList);
+ run(afterSubList);
+ return answer;
+ }
+ }
+}
diff --git
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedBrowsableEndpoint.java
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedBrowsableEndpoint.java
index 0a6a7c36cef0..78171b147016 100644
---
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedBrowsableEndpoint.java
+++
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedBrowsableEndpoint.java
@@ -16,8 +16,6 @@
*/
package org.apache.camel.management.mbean;
-import java.util.List;
-
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.api.management.ManagedResource;
@@ -45,6 +43,14 @@ public class ManagedBrowsableEndpoint extends
ManagedEndpoint implements Managed
return getEndpoint();
}
+ /**
+ * The exchanges of the endpoint as one snapshot, as the list can change
between size() and get(index) (such as a
+ * browse limit removing the oldest exchange).
+ */
+ private Exchange[] snapshot() {
+ return getEndpoint().getExchanges().toArray(new Exchange[0]);
+ }
+
@Override
public int getBrowseLimit() {
return getEndpoint().getBrowseLimit();
@@ -57,12 +63,12 @@ public class ManagedBrowsableEndpoint extends
ManagedEndpoint implements Managed
@Override
public String browseExchange(Integer index) {
- List<Exchange> exchanges = getEndpoint().getExchanges();
+ Exchange[] exchanges = snapshot();
- if (index >= exchanges.size()) {
+ if (index >= exchanges.length) {
return null;
}
- Exchange exchange = exchanges.get(index);
+ Exchange exchange = exchanges[index];
if (exchange == null) {
return null;
}
@@ -72,12 +78,12 @@ public class ManagedBrowsableEndpoint extends
ManagedEndpoint implements Managed
@Override
public String browseMessageBody(Integer index) {
- List<Exchange> exchanges = getEndpoint().getExchanges();
+ Exchange[] exchanges = snapshot();
- if (index >= exchanges.size()) {
+ if (index >= exchanges.length) {
return null;
}
- Exchange exchange = exchanges.get(index);
+ Exchange exchange = exchanges[index];
if (exchange == null) {
return null;
}
@@ -88,12 +94,12 @@ public class ManagedBrowsableEndpoint extends
ManagedEndpoint implements Managed
@Override
public String browseMessageAsXml(Integer index, Boolean includeBody) {
- List<Exchange> exchanges = getEndpoint().getExchanges();
+ Exchange[] exchanges = snapshot();
- if (index >= exchanges.size()) {
+ if (index >= exchanges.length) {
return null;
}
- Exchange exchange = exchanges.get(index);
+ Exchange exchange = exchanges[index];
if (exchange == null) {
return null;
}
@@ -120,15 +126,15 @@ public class ManagedBrowsableEndpoint extends
ManagedEndpoint implements Managed
"From index cannot be larger than to index, was: " +
fromIndex + " > " + toIndex);
}
- List<Exchange> exchanges = getEndpoint().getExchanges();
- if (exchanges.isEmpty()) {
+ Exchange[] exchanges = snapshot();
+ if (exchanges.length == 0) {
return null;
}
StringBuilder sb = new StringBuilder();
sb.append("<messages>");
- for (int i = fromIndex; i < exchanges.size() && i <= toIndex; i++) {
- Exchange exchange = exchanges.get(i);
+ for (int i = fromIndex; i < exchanges.length && i <= toIndex; i++) {
+ Exchange exchange = exchanges[i];
Message msg = exchange.getMessage();
String xml = MessageHelper.dumpAsXml(msg, includeBody);
sb.append("\n").append(xml);
@@ -139,12 +145,12 @@ public class ManagedBrowsableEndpoint extends
ManagedEndpoint implements Managed
@Override
public String browseMessageAsJSon(Integer index, Boolean includeBody) {
- List<Exchange> exchanges = getEndpoint().getExchanges();
+ Exchange[] exchanges = snapshot();
- if (index >= exchanges.size()) {
+ if (index >= exchanges.length) {
return null;
}
- Exchange exchange = exchanges.get(index);
+ Exchange exchange = exchanges[index];
if (exchange == null) {
return null;
}
@@ -171,14 +177,14 @@ public class ManagedBrowsableEndpoint extends
ManagedEndpoint implements Managed
"From index cannot be larger than to index, was: " +
fromIndex + " > " + toIndex);
}
- List<Exchange> exchanges = getEndpoint().getExchanges();
- if (exchanges.isEmpty()) {
+ Exchange[] exchanges = snapshot();
+ if (exchanges.length == 0) {
return null;
}
JsonArray arr = new JsonArray();
- for (int i = fromIndex; i < exchanges.size() && i <= toIndex; i++) {
- Exchange exchange = exchanges.get(i);
+ for (int i = fromIndex; i < exchanges.length && i <= toIndex; i++) {
+ Exchange exchange = exchanges[i];
Message msg = exchange.getMessage();
JsonObject jo = MessageHelper.dumpAsJSonObject(msg, false, false,
includeBody, true, true, true, 128 * 1024);
arr.add(jo);
diff --git
a/core/camel-management/src/test/java/org/apache/camel/management/ManagedBrowsableEndpointSnapshotTest.java
b/core/camel-management/src/test/java/org/apache/camel/management/ManagedBrowsableEndpointSnapshotTest.java
new file mode 100644
index 000000000000..326c315bee26
--- /dev/null
+++
b/core/camel-management/src/test/java/org/apache/camel/management/ManagedBrowsableEndpointSnapshotTest.java
@@ -0,0 +1,174 @@
+/*
+ * 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.management;
+
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.browse.BrowseEndpoint;
+import org.apache.camel.management.mbean.ManagedBrowsableEndpoint;
+import org.apache.camel.spi.BrowsableEndpoint.BrowseStatus;
+import org.apache.camel.util.json.JsonArray;
+import org.apache.camel.util.json.Jsoner;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The readers of a browse endpoint must read one snapshot of its list: with a
browse limit a producer adds the new
+ * exchange and then removes the oldest, so a reader that calls size() and
then get(index) can find the list shorter
+ * than the size it read.
+ */
+public class ManagedBrowsableEndpointSnapshotTest extends ContextTestSupport {
+
+ private static final int LIMIT = 3;
+
+ private GatedList list;
+ private BrowseEndpoint endpoint;
+
+ @BeforeEach
+ void createEndpoint() throws Exception {
+ endpoint = new BrowseEndpoint("browse:gated",
context.getComponent("browse")) {
+ @Override
+ protected List<Exchange> createExchangeList() {
+ list = new GatedList();
+ return list;
+ }
+ };
+ endpoint.setName("gated");
+ endpoint.setBrowseLimit(LIMIT);
+ context.addEndpoint("browse:gated", endpoint);
+
+ send("A", 1);
+ send("B", 2);
+ send("C", 3);
+ }
+
+ @Test
+ void testBrowseMessageBodyOfLastWhileOldestIsRemoved() throws Exception {
+ ManagedBrowsableEndpoint mbean = new
ManagedBrowsableEndpoint(endpoint);
+
+ // the reader sees A, B, C and D and asks for D, and the producer
removes A before the reader gets it
+ String body = readWhileProducerMakesRoom(1, () ->
mbean.browseMessageBody(LIMIT));
+
+ assertEquals("D", body);
+ }
+
+ @Test
+ void testBrowseStatusWhileOldestIsRemoved() throws Exception {
+ // the reader gets the first (A), and the producer removes A before
the reader gets the last
+ BrowseStatus status = readWhileProducerMakesRoom(2, () ->
endpoint.getBrowseStatus(0));
+
+ // size, first and last all come from the same list: A, B, C and D
+ assertEquals(new BrowseStatus(LIMIT + 1, 1, 4), status);
+ }
+
+ @Test
+ void testBrowseRangeWhileOldestIsRemoved() throws Exception {
+ ManagedBrowsableEndpoint mbean = new
ManagedBrowsableEndpoint(endpoint);
+
+ // the reader sees four exchanges, and the producer removes A before
the reader gets the first
+ String json = readWhileProducerMakesRoom(1, () ->
mbean.browseRangeMessagesAsJSon(0, Integer.MAX_VALUE, true));
+
+ JsonArray arr = (JsonArray) Jsoner.deserialize(json);
+ assertEquals(LIMIT + 1, arr.size(), json);
+ assertTrue(json.contains("\"A\"") && json.contains("\"D\""), json);
+ }
+
+ /**
+ * Sends D from another thread and holds that producer after it added D
(so the list has one more than the limit)
+ * until the reader calls get(index) for the given time; the producer then
removes the oldest before that get runs.
+ * A reader that reads one snapshot does not call get(index), and the
producer completes after the reader.
+ */
+ private <T> T readWhileProducerMakesRoom(int trimBeforeGet, Callable<T>
reader) throws Exception {
+ Thread producer = new Thread(() -> send("D", 4), "browse-producer");
+ list.trimBeforeGet.set(trimBeforeGet);
+ list.producer = producer;
+ list.holdAfterAdd = true;
+ producer.start();
+ assertTrue(list.added.await(20, TimeUnit.SECONDS), "producer did not
add");
+ assertEquals(LIMIT + 1, list.size());
+ try {
+ return reader.call();
+ } finally {
+ list.trim.countDown();
+ producer.join(20000);
+ assertFalse(producer.isAlive(), "producer did not complete");
+ assertEquals(LIMIT, endpoint.getExchanges().size());
+ }
+ }
+
+ private void send(String body, long timestamp) {
+ template.sendBodyAndHeader(endpoint, body, Exchange.MESSAGE_TIMESTAMP,
timestamp);
+ }
+
+ /**
+ * The list of the endpoint, which can hold the producer after an add and
let it continue at the n-th get(index).
+ */
+ private static final class GatedList extends
CopyOnWriteArrayList<Exchange> {
+ private final CountDownLatch added = new CountDownLatch(1);
+ private final CountDownLatch trim = new CountDownLatch(1);
+ private final AtomicInteger trimBeforeGet = new AtomicInteger();
+ private volatile boolean holdAfterAdd;
+ private volatile Thread producer;
+
+ @Override
+ public boolean add(Exchange exchange) {
+ boolean answer = super.add(exchange);
+ if (holdAfterAdd) {
+ holdAfterAdd = false;
+ added.countDown();
+ try {
+ if (!trim.await(20, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("producer was not
released");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ }
+ return answer;
+ }
+
+ @Override
+ public Exchange get(int index) {
+ if (trimBeforeGet.get() > 0 && trimBeforeGet.decrementAndGet() ==
0) {
+ // let the producer remove the oldest, and wait until it did
+ trim.countDown();
+ try {
+ producer.join(20000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ if (producer.isAlive()) {
+ throw new IllegalStateException("producer did not
complete");
+ }
+ }
+ return super.get(index);
+ }
+ }
+}
