This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch rcmd in repository https://gitbox.apache.org/repos/asf/camel.git
commit f59476a3388d218757130a0d8cb2a4fc2d9878a0 Author: Claus Ibsen <[email protected]> AuthorDate: Wed Oct 9 21:00:51 2024 +0200 CAMEL-21193: camel-jbang - Add listen command --- .../apache/camel/impl/console/ReceiveDevConsole.java | 17 ++++++++++++++++- .../core/commands/action/CamelListenAction.java | 20 +++++++++++++++++--- 2 files changed, 33 insertions(+), 4 deletions(-) diff --git a/core/camel-console/src/main/java/org/apache/camel/impl/console/ReceiveDevConsole.java b/core/camel-console/src/main/java/org/apache/camel/impl/console/ReceiveDevConsole.java index 1d794f55a52..2a74dd49a66 100644 --- a/core/camel-console/src/main/java/org/apache/camel/impl/console/ReceiveDevConsole.java +++ b/core/camel-console/src/main/java/org/apache/camel/impl/console/ReceiveDevConsole.java @@ -24,6 +24,7 @@ import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; +import netscape.javascript.JSObject; import org.apache.camel.CamelContext; import org.apache.camel.Consumer; import org.apache.camel.Endpoint; @@ -73,6 +74,8 @@ public class ReceiveDevConsole extends AbstractDevConsole { private final AtomicBoolean enabled = new AtomicBoolean(); private final AtomicLong uuid = new AtomicLong(); private Queue<JsonObject> queue; + private long firstTimestamp; + private long lastTimestamp; public ReceiveDevConsole() { super("camel", "receive", "Camel Receive", "Consume messages from endpoints"); @@ -127,6 +130,10 @@ public class ReceiveDevConsole extends AbstractDevConsole { if (removeOnDump) { queue.clear(); } + JsonObject jo = (JsonObject) arr.get(0); + firstTimestamp = jo.getLongOrDefault("timestamp", 0); + jo = (JsonObject) arr.get(arr.size() - 1); + lastTimestamp = jo.getLongOrDefault("timestamp", 0); String json = arr.toJson(); sb.append(json).append("\n"); return sb.toString(); @@ -181,7 +188,8 @@ public class ReceiveDevConsole extends AbstractDevConsole { json.put("uid", uuid.incrementAndGet()); json.put("endpointUri", exchange.getFromEndpoint().toString()); json.put("remoteEndpoint", exchange.getFromEndpoint().isRemote()); - json.put("timestamp", exchange.getMessage().getMessageTimestamp()); + lastTimestamp = exchange.getMessage().getMessageTimestamp(); + json.put("timestamp", lastTimestamp); // ensure there is space on the queue by polling until at least single slot is free int drain = queue.size() - capacity + 1; @@ -240,6 +248,10 @@ public class ReceiveDevConsole extends AbstractDevConsole { queue.clear(); } root.put("messages", arr); + JsonObject jo = (JsonObject) arr.get(0); + firstTimestamp = jo.getLongOrDefault("timestamp", 0); + jo = (JsonObject) arr.get(arr.size() - 1); + lastTimestamp = jo.getLongOrDefault("timestamp", 0); return root; } @@ -271,6 +283,9 @@ public class ReceiveDevConsole extends AbstractDevConsole { root.put("enabled", this.enabled.get()); root.put("total", uuid.get()); + root.put("firstTimestamp", firstTimestamp); + root.put("lastTimestamp", lastTimestamp); + JsonArray arr = new JsonArray(); for (Consumer c : consumers) { JsonObject jo = new JsonObject(); diff --git a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelListenAction.java b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelListenAction.java index 33dce4074ac..30b213f9b23 100644 --- a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelListenAction.java +++ b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelListenAction.java @@ -33,6 +33,7 @@ import java.util.List; import java.util.Map; import java.util.Queue; import java.util.Set; +import java.util.StringJoiner; import java.util.concurrent.ArrayBlockingQueue; import java.util.regex.Pattern; @@ -88,7 +89,7 @@ public class CamelListenAction extends ActionBaseCommand { @CommandLine.Parameters(description = "Name or pid of running Camel integration. (default selects all)", arity = "0..1") String name = "*"; - @CommandLine.Option(names = { "--action" }, completionCandidates = CamelTraceAction.ActionCompletionCandidates.class, + @CommandLine.Option(names = { "--action" }, completionCandidates = ActionCompletionCandidates.class, defaultValue = "status", description = "Action to start, stop, clear, list status, or dump messages") String action; @@ -114,7 +115,7 @@ public class CamelListenAction extends ActionBaseCommand { boolean follow = true; @CommandLine.Option(names = { "--prefix" }, defaultValue = "auto", - completionCandidates = CamelTraceAction.PrefixCompletionCandidates.class, + completionCandidates = PrefixCompletionCandidates.class, description = "Print prefix with running Camel integration name. auto=only prefix when running multiple integrations. true=always prefix. false=prefix off.") String prefix = "auto"; @@ -237,6 +238,8 @@ public class CamelListenAction extends ActionBaseCommand { if (jo != null) { row.enabled = jo.getBoolean("enabled"); row.counter = jo.getLong("total"); + row.firstTimestamp = jo.getLongOrDefault("firstTimestamp", 0); + row.lastTimestamp = jo.getLongOrDefault("lastTimestamp", 0); JsonArray arr = jo.getCollection("endpoints"); if (arr != null) { for (Object e : arr) { @@ -265,7 +268,9 @@ public class CamelListenAction extends ActionBaseCommand { .with(r -> r.name), new Column().header("AGE").headerAlign(HorizontalAlign.CENTER).with(r -> r.age), new Column().header("STATUS").with(this::getStatus), - new Column().header("COUNTER").with(r -> "" + r.counter), + new Column().header("TOTAL").with(r -> "" + r.counter), + new Column().header("SINCE").headerAlign(HorizontalAlign.CENTER) + .with(this::getMessageAgo), new Column().header("ENDPOINT").visible(!wideUri).dataAlign(HorizontalAlign.LEFT) .maxWidth(90, OverflowBehaviour.ELLIPSIS_RIGHT) .with(this::getEndpointUri), @@ -731,6 +736,13 @@ public class CamelListenAction extends ActionBaseCommand { return u; } + protected String getMessageAgo(StatusRow r) { + if (r.lastTimestamp > 0) { + return TimeUtils.printSince(r.lastTimestamp); + } + return ""; + } + private static class Pid { String pid; String name; @@ -761,6 +773,8 @@ public class CamelListenAction extends ActionBaseCommand { long uptime; boolean enabled; long counter; + long firstTimestamp; + long lastTimestamp; String uri; StatusRow copy() {
