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 907113a1bb80 CAMEL-22715: camel-jbang - receive command to accept prop
and properties to load configuration to use when receiving. Also added support
for receive local mode without existing running Camel
907113a1bb80 is described below
commit 907113a1bb80e9754b267d0d9d15b113d59fa78c
Author: Claus Ibsen <[email protected]>
AuthorDate: Fri Nov 21 12:13:31 2025 +0100
CAMEL-22715: camel-jbang - receive command to accept prop and properties to
load configuration to use when receiving. Also added support for receive local
mode without existing running Camel
---
.../core/commands/action/ActionBaseCommand.java | 4 +
.../core/commands/action/CamelReceiveAction.java | 202 ++++++++++++++++-----
.../core/commands/action/CamelSendAction.java | 29 +--
.../core/commands/action/CamelSourceAction.java | 2 +-
.../camel/dsl/jbang/core/commands/LoadTest.java | 1 -
5 files changed, 162 insertions(+), 76 deletions(-)
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/ActionBaseCommand.java
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/ActionBaseCommand.java
index 00bc6d1c0fe8..266c6b8ca673 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/ActionBaseCommand.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/ActionBaseCommand.java
@@ -39,6 +39,10 @@ abstract class ActionBaseCommand extends CamelCommand {
}
protected static JsonObject getJsonObject(Path outputFile) {
+ return getJsonObject(outputFile, 5000);
+ }
+
+ protected static JsonObject getJsonObject(Path outputFile, long timeout) {
StopWatch watch = new StopWatch();
while (watch.taken() < 5000) {
File f = outputFile.toFile();
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelReceiveAction.java
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelReceiveAction.java
index d2cd72954eaa..5d143e1f74da 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelReceiveAction.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelReceiveAction.java
@@ -33,7 +33,9 @@ import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Pattern;
import com.github.freva.asciitable.AsciiTable;
@@ -43,9 +45,11 @@ import com.github.freva.asciitable.OverflowBehaviour;
import org.apache.camel.catalog.impl.TimePatternConverter;
import org.apache.camel.dsl.jbang.core.commands.CamelJBangMain;
import org.apache.camel.dsl.jbang.core.commands.CommandHelper;
+import org.apache.camel.dsl.jbang.core.commands.Run;
import org.apache.camel.dsl.jbang.core.common.PathUtils;
import org.apache.camel.dsl.jbang.core.common.PidNameAgeCompletionCandidates;
import org.apache.camel.dsl.jbang.core.common.ProcessHelper;
+import org.apache.camel.main.KameletMain;
import org.apache.camel.util.StopWatch;
import org.apache.camel.util.StringHelper;
import org.apache.camel.util.TimeUtils;
@@ -90,8 +94,20 @@ public class CamelReceiveAction extends ActionBaseCommand {
}
}
- @CommandLine.Parameters(description = "Name or pid of running Camel
integration. (default selects all)", arity = "0..1")
- String name = "*";
+ @CommandLine.Parameters(description = "To use an existing running Camel
integration for receiving the message (name or pid)",
+ arity = "0..1")
+ String name;
+
+ @CommandLine.Option(names = { "--properties" },
+ description = "comma separated list of properties file
(only applicable when NOT using an existing running Camel)"
+ +
+ " (ex.
/path/to/file.properties,/path/to/other.properties")
+ String propertiesFiles;
+
+ @CommandLine.Option(names = { "--prop", "--property" },
+ description = "Additional properties; override
existing (only applicable when NOT using an existing running Camel)",
+ arity = "0")
+ String[] property;
@CommandLine.Option(names = { "--action" }, completionCandidates =
ActionCompletionCandidates.class,
defaultValue = "status",
@@ -131,6 +147,10 @@ public class CamelReceiveAction extends ActionBaseCommand {
description = "Filter messages to only output matching
text (ignore case).", arity = "0..*")
String[] grep;
+ @CommandLine.Option(names = { "--timeout" }, defaultValue = "20000",
+ description = "Timeout in millis waiting for message
to be received")
+ long timeout = 20000;
+
@CommandLine.Option(names = { "--show-exchange-properties" }, defaultValue
= "false",
description = "Show exchange properties in received
messages")
boolean showExchangeProperties;
@@ -177,6 +197,8 @@ public class CamelReceiveAction extends ActionBaseCommand {
@CommandLine.Option(names = { "--output" }, description = "Output format
(text or json)")
private String output;
+ private volatile long pid;
+
String findAnsi;
private int nameMaxWidth;
private boolean prefixShown;
@@ -202,6 +224,14 @@ public class CamelReceiveAction extends ActionBaseCommand {
return doStatusCall();
}
+ if (name != null) {
+ return doCall(name, autoDump);
+ } else {
+ return doCallLocal(autoDump);
+ }
+ }
+
+ private Integer doCall(String name, boolean autoDump) throws Exception {
List<Long> pids = findPids(name);
for (long pid : pids) {
if ("clear".equals(action)) {
@@ -210,48 +240,8 @@ public class CamelReceiveAction extends ActionBaseCommand {
Files.writeString(f, "{}");
}
} else {
- // ensure output file is deleted before executing action
- Path outputFile = getOutputFile(Long.toString(pid));
- PathUtils.deleteFile(outputFile);
-
- JsonObject root = new JsonObject();
- root.put("action", "receive");
- if ("start".equals(action)) {
- root.put("enabled", "true");
- if (endpoint != null) {
- root.put("endpoint", endpoint);
- } else {
- root.put("endpoint", "*");
- }
- } else if ("stop".equals(action)) {
- root.put("enabled", "false");
- }
- Path f = getActionFile(Long.toString(pid));
- Files.writeString(f, root.toJson());
-
- JsonObject jo = waitForOutputFile(outputFile);
- if (jo != null) {
- String error = jo.getString("error");
- if (error != null) {
- error = Jsoner.unescape(error);
- String url = jo.getString("url");
- List<String> stackTrace =
jo.getCollection("stackTrace");
- if (url != null) {
- printer().println("Error starting to receive
messages from: " + url + " due to: " + error);
-
- } else {
- printer().println("Error starting to receive
messages due to: " + error);
- }
- printer().println(StringHelper.fillChars('-', 120));
- printer().println(StringHelper.padString(1, 55) +
"STACK-TRACE");
- printer().println(StringHelper.fillChars('-', 120));
- StringBuilder sb = new StringBuilder();
- for (String s : stackTrace) {
- sb.append(String.format("\t%s%n", s));
- }
- printer().println(String.valueOf(sb));
- }
- }
+ Path outputFile = writeReceiveData();
+ showStatus(outputFile);
}
}
@@ -262,16 +252,125 @@ public class CamelReceiveAction extends
ActionBaseCommand {
return 0;
}
- protected JsonObject waitForOutputFile(Path outputFile) {
- return getJsonObject(outputFile);
+ private Integer doCallLocal(boolean autoDump) throws Exception {
+ AtomicReference<KameletMain> ref = new AtomicReference<>();
+ Run run = new Run(this.getMain()) {
+ @Override
+ protected int runKameletMain(KameletMain main) throws Exception {
+ ref.set(main);
+ return super.runKameletMain(main);
+ }
+ };
+ run.empty = true;
+ run.propertiesFiles = propertiesFiles;
+ run.property = property;
+
+ // spawn thread that waits for response file
+ final CountDownLatch latch = new CountDownLatch(1);
+ this.pid = ProcessHandle.current().pid();
+ Path outputFile = writeReceiveData();
+ Thread t = new Thread("CamelJBangSendStatus") {
+ @Override
+ public void run() {
+ try {
+ if (!showStatus(outputFile)) {
+ // some kind of error so exit
+ return;
+ }
+ if (autoDump) {
+ doDumpCall();
+ }
+ } catch (Exception e) {
+ // ignore
+ } finally {
+ latch.countDown();
+ // signal to main we are complete
+ KameletMain main = ref.get();
+ if (main != null) {
+ main.completed();
+ }
+ }
+ }
+ };
+ // keep thread running as we need it to show the status before
terminating
+ t.start();
+
+ Integer exit = run.call();
+ latch.await();
+
+ return exit;
+ }
+
+ protected Path writeReceiveData() {
+ // ensure output file is deleted before executing action
+ Path outputFile = getOutputFile(Long.toString(pid));
+ PathUtils.deleteFile(outputFile);
+
+ JsonObject root = new JsonObject();
+ root.put("action", "receive");
+ if ("start".equals(action)) {
+ root.put("enabled", "true");
+ if (endpoint != null) {
+ root.put("endpoint", endpoint);
+ } else {
+ root.put("endpoint", "*");
+ }
+ } else if ("stop".equals(action)) {
+ root.put("enabled", "false");
+ }
+ Path f = getActionFile(Long.toString(pid));
+ try {
+ String text = root.toJson();
+ Files.writeString(f, text);
+ } catch (Exception e) {
+ // ignore
+ }
+
+ return outputFile;
+ }
+
+ protected boolean showStatus(Path outputFile) throws Exception {
+ // wait longer than timeout
+ JsonObject jo = getJsonObject(outputFile, timeout + 10000);
+ if (jo != null) {
+ String error = jo.getString("error");
+ if (error != null) {
+ error = Jsoner.unescape(error);
+ String url = jo.getString("url");
+ List<String> stackTrace = jo.getCollection("stackTrace");
+ if (url != null) {
+ printer().println("Error starting to receive messages
from: " + url + " due to: " + error);
+
+ } else {
+ printer().println("Error starting to receive messages due
to: " + error);
+ }
+ printer().println(StringHelper.fillChars('-', 120));
+ printer().println(StringHelper.padString(1, 55) +
"STACK-TRACE");
+ printer().println(StringHelper.fillChars('-', 120));
+ StringBuilder sb = new StringBuilder();
+ for (String s : stackTrace) {
+ sb.append(String.format("\t%s%n", s));
+ }
+ printer().println(String.valueOf(sb));
+ return false;
+ }
+ }
+ return true;
}
protected Integer doStatusCall() {
List<StatusRow> rows = new ArrayList<>();
- List<Long> pids = findPids(name);
+ List<Long> pids;
+ if (pid != 0) {
+ pids = List.of(pid);
+ } else if (name != null) {
+ pids = findPids(name);
+ } else {
+ return 0;
+ }
ProcessHandle.allProcesses()
- .filter(ph -> pids.contains(ph.pid()))
+ .filter(ph -> pid == 0 && pids.contains(ph.pid()))
.forEach(ph -> {
JsonObject root = loadStatus(ph.pid());
if (root != null) {
@@ -481,7 +580,12 @@ public class CamelReceiveAction extends ActionBaseCommand {
}
private void updatePids(Map<Long, Pid> rows) {
- List<Long> pids = findPids(name);
+ List<Long> pids;
+ if (name != null) {
+ pids = findPids(name);
+ } else {
+ pids = List.of(pid);
+ }
ProcessHandle.allProcesses()
.filter(ph -> pids.contains(ph.pid()))
.forEach(ph -> {
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendAction.java
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendAction.java
index f7509bcb9794..3908bfb73df2 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendAction.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendAction.java
@@ -31,7 +31,6 @@ import
org.apache.camel.dsl.jbang.core.commands.CamelJBangMain;
import org.apache.camel.dsl.jbang.core.commands.Run;
import org.apache.camel.dsl.jbang.core.common.PathUtils;
import org.apache.camel.main.KameletMain;
-import org.apache.camel.util.StopWatch;
import org.apache.camel.util.StringHelper;
import org.apache.camel.util.TimeUtils;
import org.apache.camel.util.json.JsonArray;
@@ -56,7 +55,8 @@ public class CamelSendAction extends ActionBaseCommand {
" (ex.
/path/to/file.properties,/path/to/other.properties")
String propertiesFiles;
- @CommandLine.Option(names = { "--prop", "--property" }, description =
"Additional properties; override existing (only applicable when NOT using an
existing running Camel)",
+ @CommandLine.Option(names = { "--prop", "--property" },
+ description = "Additional properties; override
existing (only applicable when NOT using an existing running Camel)",
arity = "0")
String[] property;
@@ -207,7 +207,8 @@ public class CamelSendAction extends ActionBaseCommand {
protected void showStatus(Path outputFile) throws Exception {
try {
- JsonObject jo = waitForOutputFile(outputFile);
+ // wait longer than timeout
+ JsonObject jo = getJsonObject(outputFile, timeout + 10000);
if (jo != null) {
printStatusLine(jo);
String exchangeId = jo.getString("exchangeId");
@@ -385,26 +386,4 @@ public class CamelSendAction extends ActionBaseCommand {
}
}
- protected JsonObject waitForOutputFile(Path outputFile) {
- StopWatch watch = new StopWatch();
- long wait = timeout + 10000; // wait longer than timeout
- while (watch.taken() < wait) {
- File f = outputFile.toFile();
- try {
- // give time for response to be ready
- Thread.sleep(20);
-
- if (Files.exists(outputFile) && f.length() > 0) {
- String text = Files.readString(outputFile);
- return (JsonObject) Jsoner.deserialize(text);
- }
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- } catch (Exception e) {
- // ignore
- }
- }
- return null;
- }
-
}
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSourceAction.java
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSourceAction.java
index b18e390e172f..00c55fb70e94 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSourceAction.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSourceAction.java
@@ -166,7 +166,7 @@ public class CamelSourceAction extends ActionBaseCommand {
}
protected JsonObject waitForOutputFile(Path outputFile) {
- return getJsonObject((Path) outputFile);
+ return getJsonObject(outputFile);
}
public static String extractSourceName(String loc) {
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/LoadTest.java
b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/LoadTest.java
index 4839eaf27317..47c5df47ae13 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/LoadTest.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/LoadTest.java
@@ -56,7 +56,6 @@ class LoadTest {
Runnable r = () -> {
try {
Thread.sleep(2000);
- System.out.println("Calling load...");
CamelLoadAction load = new CamelLoadAction(new
CamelJBangMain());
CommandLine.populateCommand(load, getPid(),
"--source=src/test/resources/load.yaml");
int exit = load.doCall();
