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 5bc24e361bf3 CAMEL-26191: camel-docker - the streaming operations give
their result and fail the exchange on errors, and the events since time is in
seconds (#29021)
5bc24e361bf3 is described below
commit 5bc24e361bf3b0f6645e3debc916539b911dd7c9
Author: Claus Ibsen <[email protected]>
AuthorDate: Tue Nov 3 11:43:36 2026 +0100
CAMEL-26191: camel-docker - the streaming operations give their result and
fail the exchange on errors, and the events since time is in seconds (#29021)
…t and fail the exchange on errors, and the events since time is in seconds
The async producer (imagebuild, imagepull, imagepush, containerattach,
containerlog, containerwait, execstart) only logged its errors, so the
exchange succeeded. It now sets the exception on the exchange. imagebuild
accepts a byte[] or any body that converts to an InputStream, and gives the
image id of the build (also for a tagged build), failing on a failed build.
containerattach, containerlog and execstart give the output as text,
containerwait the exit code, and imagepull/imagepush the last progress item
instead of the docker-java callback; a failed push fails the exchange.
The events consumer passes the since time in seconds, as the Docker API
expects. camel-docker adds the Netty native transports for arm64, so the
default Netty cmdExecFactory works on Apple silicon and Linux arm64, and
DEFAULT_CMD_EXEC_FACTORY now names that default.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../apache/camel/catalog/components/docker.json | 6 +-
.../camel/catalog/docs/docker-component.adoc | 15 ++
components/camel-docker/pom.xml | 14 +
.../org/apache/camel/component/docker/docker.json | 6 +-
.../src/main/docs/docker-component.adoc | 15 ++
.../component/docker/DockerConfiguration.java | 2 +-
.../camel/component/docker/DockerConstants.java | 6 +-
.../camel/component/docker/DockerEndpoint.java | 38 +--
.../docker/consumer/DockerEventsConsumer.java | 5 +-
.../docker/producer/AsyncDockerProducer.java | 79 ++++--
.../docker/consumer/DockerEventsConsumerTest.java | 15 ++
.../docker/producer/AsyncDockerProducerTest.java | 284 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 25 ++
.../endpoint/dsl/DockerEndpointBuilderFactory.java | 3 +-
14 files changed, 457 insertions(+), 56 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/docker.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/docker.json
index 1c820359545b..07c7b96edc4d 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/docker.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/docker.json
@@ -72,7 +72,7 @@
"CamelDockerTerm": { "index": 22, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The term to search", "constantName":
"org.apache.camel.component.docker.DockerConstants#DOCKER_TERM" },
"CamelDockerForce": { "index": 23, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "Boolean",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "With force flag", "constantName":
"org.apache.camel.component.docker.DockerConstants#DOCKER_FORCE" },
"CamelDockerNoPrune": { "index": 24, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "Boolean",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "With no prune flag", "constantName":
"org.apache.camel.component.docker.DockerConstants#DOCKER_NO_PRUNE" },
- "CamelDockerInitialRange": { "index": 25, "kind": "header", "displayName":
"", "group": "common", "label": "", "required": false, "javaType": "Long",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The initial range", "constantName":
"org.apache.camel.component.docker.DockerConstants#DOCKER_INITIAL_RANGE" },
+ "CamelDockerInitialRange": { "index": 25, "kind": "header", "displayName":
"", "group": "common", "label": "", "required": false, "javaType": "Long",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The time in milliseconds before the start of the events
consumer from which it also receives the past events", "constantName":
"org.apache.camel.component.docker.DockerConstants#DOCKER_INITIAL_RANGE" },
"CamelDockerBefore": { "index": 26, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "With before", "constantName":
"org.apache.camel.component.docker.DockerConstants#DOCKER_BEFORE" },
"CamelDockerLimit": { "index": 27, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "Integer",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "With limit", "constantName":
"org.apache.camel.component.docker.DockerConstants#DOCKER_LIMIT" },
"CamelDockerShowSize": { "index": 28, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "Boolean",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "With show size flag", "constantName":
"org.apache.camel.component.docker.DockerConstants#DOCKER_SHOW_SIZE" },
@@ -132,8 +132,8 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "pojo" ], "byOption": [ { "option":
"operation", "value": "events", "carriers": [ "pojo" ], "javaType":
"com.github.dockerjava.api.model.Event" }, { "option": "operation", "value":
"stats", "carriers": [ "pojo" ], "javaType":
"com.github.dockerjava.api.model.Statistics" } ], "description": "The object of
docker-java for each item of the stream: an Event of the daemon with events,
the Statistics of the container in CamelDockerContainerId with stats" },
- "producerIn": { "carriers": [ "any" ], "byOption": [ { "option":
"operation", "value": "imagecreate", "carriers": [ "stream", "bytes", "text",
"file" ] }, { "option": "operation", "value": "imagebuild", "carriers": [
"stream", "file" ] } ], "description": "Not used by most operations, which take
their input from the options and headers, such as the container in
CamelDockerContainerId. imagecreate reads the image as a stream; imagebuild
needs an InputStream or a java.io.File of the bu [...]
- "producerOut": { "carriers": [ "pojo", "list", "text", "stream", "null",
"unchanged" ], "byOption": [ { "option": "operation", "value": "auth",
"carriers": [ "pojo" ], "javaType":
"com.github.dockerjava.api.model.AuthResponse" }, { "option": "operation",
"value": "info", "carriers": [ "pojo" ], "javaType":
"com.github.dockerjava.api.model.Info" }, { "option": "operation", "value":
"ping", "carriers": [ "unchanged" ] }, { "option": "operation", "value":
"version", "carriers": [ "pojo" [...]
+ "producerIn": { "carriers": [ "any" ], "byOption": [ { "option":
"operation", "value": "imagecreate", "carriers": [ "stream", "bytes", "text",
"file" ] }, { "option": "operation", "value": "imagebuild", "carriers": [
"stream", "bytes", "file" ] } ], "description": "Not used by most operations,
which take their input from the options and headers, such as the container in
CamelDockerContainerId. imagecreate reads the image as a stream; imagebuild
needs the build context as a tar archiv [...]
+ "producerOut": { "carriers": [ "pojo", "list", "text", "stream", "value",
"unchanged" ], "byOption": [ { "option": "operation", "value": "auth",
"carriers": [ "pojo" ], "javaType":
"com.github.dockerjava.api.model.AuthResponse" }, { "option": "operation",
"value": "info", "carriers": [ "pojo" ], "javaType":
"com.github.dockerjava.api.model.Info" }, { "option": "operation", "value":
"ping", "carriers": [ "unchanged" ] }, { "option": "operation", "value":
"version", "carriers": [ "pojo [...]
},
"properties": {
"operation": { "index": 0, "kind": "path", "displayName": "Operation",
"group": "common", "label": "", "required": true, "type": "enum", "javaType":
"org.apache.camel.component.docker.DockerOperation", "enum": [ "events",
"stats", "auth", "info", "ping", "version", "imagebuild", "imagecreate",
"imageinspect", "imagelist", "imagepull", "imagepush", "imageremove",
"imagesearch", "imagetag", "containerattach", "containercommit",
"containercopyfile", "containercreate", "containerdiff", " [...]
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/docker-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/docker-component.adoc
index 55711082149b..e0e3157e838e 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/docker-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/docker-component.adoc
@@ -49,6 +49,21 @@ below
|containerId |CamelDockerContainerId
|=======================================================================
+=== Streaming operations
+
+The operations `imagebuild`, `imagepull`, `imagepush`, `containerattach`,
`containerlog`, `containerwait` and
+`execstart` wait until Docker has finished, and set the result as the message
body:
+
+* `imagebuild`: the id of the built image. The body is the build context: a
tar archive as an `InputStream`, a
+ `byte[]` or another body that converts to an `InputStream`, or a
`java.io.File` of the Dockerfile or its directory.
+* `containerattach`, `containerlog` and `execstart`: the output of stdout and
stderr as a `String`.
+* `containerwait`: the exit code of the container as an `Integer`.
+* `imagepull` and `imagepush`: the last `PullResponseItem` or
`PushResponseItem` of the progress.
+
+When the operation fails, for example a build step that fails, the exchange
fails.
+
+NOTE: With `followStream=true`, `containerattach` and `containerlog` wait
until the container stops.
+
== Examples
diff --git a/components/camel-docker/pom.xml b/components/camel-docker/pom.xml
index 8af44fec78fc..1dde3e18827a 100644
--- a/components/camel-docker/pom.xml
+++ b/components/camel-docker/pom.xml
@@ -44,6 +44,20 @@
<artifactId>docker-java</artifactId>
<version>${docker-java-version}</version>
</dependency>
+ <!-- docker-java brings the native unix socket transports of Netty for
x86_64 only:
+ add them for arm64 so the default Netty cmdExecFactory works on
Apple silicon and Linux arm64 -->
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-transport-native-kqueue</artifactId>
+ <classifier>osx-aarch_64</classifier>
+ <scope>runtime</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-transport-native-epoll</artifactId>
+ <classifier>linux-aarch_64</classifier>
+ <scope>runtime</scope>
+ </dependency>
<!-- testing -->
<dependency>
diff --git
a/components/camel-docker/src/generated/resources/META-INF/org/apache/camel/component/docker/docker.json
b/components/camel-docker/src/generated/resources/META-INF/org/apache/camel/component/docker/docker.json
index 1c820359545b..07c7b96edc4d 100644
---
a/components/camel-docker/src/generated/resources/META-INF/org/apache/camel/component/docker/docker.json
+++
b/components/camel-docker/src/generated/resources/META-INF/org/apache/camel/component/docker/docker.json
@@ -72,7 +72,7 @@
"CamelDockerTerm": { "index": 22, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The term to search", "constantName":
"org.apache.camel.component.docker.DockerConstants#DOCKER_TERM" },
"CamelDockerForce": { "index": 23, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "Boolean",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "With force flag", "constantName":
"org.apache.camel.component.docker.DockerConstants#DOCKER_FORCE" },
"CamelDockerNoPrune": { "index": 24, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "Boolean",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "With no prune flag", "constantName":
"org.apache.camel.component.docker.DockerConstants#DOCKER_NO_PRUNE" },
- "CamelDockerInitialRange": { "index": 25, "kind": "header", "displayName":
"", "group": "common", "label": "", "required": false, "javaType": "Long",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The initial range", "constantName":
"org.apache.camel.component.docker.DockerConstants#DOCKER_INITIAL_RANGE" },
+ "CamelDockerInitialRange": { "index": 25, "kind": "header", "displayName":
"", "group": "common", "label": "", "required": false, "javaType": "Long",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The time in milliseconds before the start of the events
consumer from which it also receives the past events", "constantName":
"org.apache.camel.component.docker.DockerConstants#DOCKER_INITIAL_RANGE" },
"CamelDockerBefore": { "index": 26, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "With before", "constantName":
"org.apache.camel.component.docker.DockerConstants#DOCKER_BEFORE" },
"CamelDockerLimit": { "index": 27, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "Integer",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "With limit", "constantName":
"org.apache.camel.component.docker.DockerConstants#DOCKER_LIMIT" },
"CamelDockerShowSize": { "index": 28, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "Boolean",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "With show size flag", "constantName":
"org.apache.camel.component.docker.DockerConstants#DOCKER_SHOW_SIZE" },
@@ -132,8 +132,8 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "pojo" ], "byOption": [ { "option":
"operation", "value": "events", "carriers": [ "pojo" ], "javaType":
"com.github.dockerjava.api.model.Event" }, { "option": "operation", "value":
"stats", "carriers": [ "pojo" ], "javaType":
"com.github.dockerjava.api.model.Statistics" } ], "description": "The object of
docker-java for each item of the stream: an Event of the daemon with events,
the Statistics of the container in CamelDockerContainerId with stats" },
- "producerIn": { "carriers": [ "any" ], "byOption": [ { "option":
"operation", "value": "imagecreate", "carriers": [ "stream", "bytes", "text",
"file" ] }, { "option": "operation", "value": "imagebuild", "carriers": [
"stream", "file" ] } ], "description": "Not used by most operations, which take
their input from the options and headers, such as the container in
CamelDockerContainerId. imagecreate reads the image as a stream; imagebuild
needs an InputStream or a java.io.File of the bu [...]
- "producerOut": { "carriers": [ "pojo", "list", "text", "stream", "null",
"unchanged" ], "byOption": [ { "option": "operation", "value": "auth",
"carriers": [ "pojo" ], "javaType":
"com.github.dockerjava.api.model.AuthResponse" }, { "option": "operation",
"value": "info", "carriers": [ "pojo" ], "javaType":
"com.github.dockerjava.api.model.Info" }, { "option": "operation", "value":
"ping", "carriers": [ "unchanged" ] }, { "option": "operation", "value":
"version", "carriers": [ "pojo" [...]
+ "producerIn": { "carriers": [ "any" ], "byOption": [ { "option":
"operation", "value": "imagecreate", "carriers": [ "stream", "bytes", "text",
"file" ] }, { "option": "operation", "value": "imagebuild", "carriers": [
"stream", "bytes", "file" ] } ], "description": "Not used by most operations,
which take their input from the options and headers, such as the container in
CamelDockerContainerId. imagecreate reads the image as a stream; imagebuild
needs the build context as a tar archiv [...]
+ "producerOut": { "carriers": [ "pojo", "list", "text", "stream", "value",
"unchanged" ], "byOption": [ { "option": "operation", "value": "auth",
"carriers": [ "pojo" ], "javaType":
"com.github.dockerjava.api.model.AuthResponse" }, { "option": "operation",
"value": "info", "carriers": [ "pojo" ], "javaType":
"com.github.dockerjava.api.model.Info" }, { "option": "operation", "value":
"ping", "carriers": [ "unchanged" ] }, { "option": "operation", "value":
"version", "carriers": [ "pojo [...]
},
"properties": {
"operation": { "index": 0, "kind": "path", "displayName": "Operation",
"group": "common", "label": "", "required": true, "type": "enum", "javaType":
"org.apache.camel.component.docker.DockerOperation", "enum": [ "events",
"stats", "auth", "info", "ping", "version", "imagebuild", "imagecreate",
"imageinspect", "imagelist", "imagepull", "imagepush", "imageremove",
"imagesearch", "imagetag", "containerattach", "containercommit",
"containercopyfile", "containercreate", "containerdiff", " [...]
diff --git a/components/camel-docker/src/main/docs/docker-component.adoc
b/components/camel-docker/src/main/docs/docker-component.adoc
index 55711082149b..e0e3157e838e 100644
--- a/components/camel-docker/src/main/docs/docker-component.adoc
+++ b/components/camel-docker/src/main/docs/docker-component.adoc
@@ -49,6 +49,21 @@ below
|containerId |CamelDockerContainerId
|=======================================================================
+=== Streaming operations
+
+The operations `imagebuild`, `imagepull`, `imagepush`, `containerattach`,
`containerlog`, `containerwait` and
+`execstart` wait until Docker has finished, and set the result as the message
body:
+
+* `imagebuild`: the id of the built image. The body is the build context: a
tar archive as an `InputStream`, a
+ `byte[]` or another body that converts to an `InputStream`, or a
`java.io.File` of the Dockerfile or its directory.
+* `containerattach`, `containerlog` and `execstart`: the output of stdout and
stderr as a `String`.
+* `containerwait`: the exit code of the container as an `Integer`.
+* `imagepull` and `imagepush`: the last `PullResponseItem` or
`PushResponseItem` of the progress.
+
+When the operation fails, for example a build step that fails, the exchange
fails.
+
+NOTE: With `followStream=true`, `containerattach` and `containerlog` wait
until the container stops.
+
== Examples
diff --git
a/components/camel-docker/src/main/java/org/apache/camel/component/docker/DockerConfiguration.java
b/components/camel-docker/src/main/java/org/apache/camel/component/docker/DockerConfiguration.java
index a93e8450fcff..9214894cf5c0 100644
---
a/components/camel-docker/src/main/java/org/apache/camel/component/docker/DockerConfiguration.java
+++
b/components/camel-docker/src/main/java/org/apache/camel/component/docker/DockerConfiguration.java
@@ -66,7 +66,7 @@ public class DockerConfiguration implements Cloneable {
@UriParam(label = "advanced", defaultValue = "true")
private boolean socket;
@UriParam(label = "advanced", defaultValue =
"com.github.dockerjava.netty.NettyDockerCmdExecFactory")
- private String cmdExecFactory =
"com.github.dockerjava.netty.NettyDockerCmdExecFactory";
+ private String cmdExecFactory = DockerConstants.DEFAULT_CMD_EXEC_FACTORY;
@UriParam(label = "advanced")
private Map<String, Object> parameters = new HashMap<>();
diff --git
a/components/camel-docker/src/main/java/org/apache/camel/component/docker/DockerConstants.java
b/components/camel-docker/src/main/java/org/apache/camel/component/docker/DockerConstants.java
index 88145836f2b2..8b22cd4b2cc8 100644
---
a/components/camel-docker/src/main/java/org/apache/camel/component/docker/DockerConstants.java
+++
b/components/camel-docker/src/main/java/org/apache/camel/component/docker/DockerConstants.java
@@ -33,7 +33,7 @@ public final class DockerConstants {
/**
* Endpoint configuration defaults
*/
- public static final String DEFAULT_CMD_EXEC_FACTORY =
"com.github.dockerjava.jaxrs.JerseyDockerCmdExecFactory";
+ public static final String DEFAULT_CMD_EXEC_FACTORY =
"com.github.dockerjava.netty.NettyDockerCmdExecFactory";
/**
* Connectivity *
@@ -128,7 +128,9 @@ public final class DockerConstants {
/**
* Events *
*/
- @Metadata(description = "The initial range", javaType = "Long")
+ @Metadata(description = "The time in milliseconds before the start of the
events consumer from which it also receives"
+ + " the past events",
+ javaType = "Long")
public static final String DOCKER_INITIAL_RANGE =
"CamelDockerInitialRange";
/**
diff --git
a/components/camel-docker/src/main/java/org/apache/camel/component/docker/DockerEndpoint.java
b/components/camel-docker/src/main/java/org/apache/camel/component/docker/DockerEndpoint.java
index 48981f10659d..c697a4dc94e8 100644
---
a/components/camel-docker/src/main/java/org/apache/camel/component/docker/DockerEndpoint.java
+++
b/components/camel-docker/src/main/java/org/apache/camel/component/docker/DockerEndpoint.java
@@ -56,14 +56,15 @@ import org.apache.camel.support.DefaultEndpoint;
carriers = {
Carrier.STREAM, Carrier.BYTES, Carrier.TEXT, Carrier.FILE }),
@BodyType.When(option = "operation",
value = "imagebuild",
- carriers = {
Carrier.STREAM, Carrier.FILE }) },
+ carriers = {
Carrier.STREAM, Carrier.BYTES, Carrier.FILE }) },
description = "Not used by most operations,
which take their input from the options and"
+ " headers, such as the
container in CamelDockerContainerId. imagecreate"
- + " reads the image as a
stream; imagebuild needs an InputStream or a"
- + " java.io.File of the build
context, and another body is logged as an"
- + " error"),
+ + " reads the image as a
stream; imagebuild needs the build context as"
+ + " a tar archive (an
InputStream, a byte[] or another body that converts"
+ + " to an InputStream) or as a
java.io.File of the Dockerfile or its"
+ + " directory, and fails
without one"),
producerOut = @BodyType(value = {
- Carrier.POJO, Carrier.LIST, Carrier.TEXT, Carrier.STREAM,
Carrier.NULL,
+ Carrier.POJO, Carrier.LIST, Carrier.TEXT, Carrier.STREAM,
Carrier.VALUE,
Carrier.UNCHANGED },
byOption = {
@BodyType.When(option =
"operation", value = "auth", carriers = Carrier.POJO,
@@ -74,22 +75,23 @@ import org.apache.camel.support.DefaultEndpoint;
@BodyType.When(option =
"operation", value = "version", carriers = Carrier.POJO,
javaType =
"com.github.dockerjava.api.model.Version"),
@BodyType.When(option =
"operation", value = "imagebuild",
- carriers = {
Carrier.TEXT, Carrier.NULL }),
+ carriers =
Carrier.TEXT, javaType = "java.lang.String"),
@BodyType.When(option =
"operation", value = "imagecreate", carriers = Carrier.POJO,
javaType =
"com.github.dockerjava.api.command.CreateImageResponse"),
@BodyType.When(option =
"operation", value = "imageinspect", carriers = Carrier.POJO,
javaType =
"com.github.dockerjava.api.command.InspectImageResponse"),
@BodyType.When(option =
"operation", value = "imagelist", carriers = Carrier.LIST),
@BodyType.When(option =
"operation", value = "imagepull", carriers = Carrier.POJO,
- javaType =
"com.github.dockerjava.api.command.PullImageResultCallback"),
- @BodyType.When(option =
"operation", value = "imagepush", carriers = Carrier.POJO),
+ javaType =
"com.github.dockerjava.api.model.PullResponseItem"),
+ @BodyType.When(option =
"operation", value = "imagepush", carriers = Carrier.POJO,
+ javaType =
"com.github.dockerjava.api.model.PushResponseItem"),
@BodyType.When(option =
"operation", value = "imageremove",
carriers =
Carrier.UNCHANGED),
@BodyType.When(option =
"operation", value = "imagesearch", carriers = Carrier.LIST),
@BodyType.When(option =
"operation", value = "imagetag",
carriers =
Carrier.UNCHANGED),
@BodyType.When(option =
"operation", value = "containerattach",
- carriers =
Carrier.POJO),
+ carriers =
Carrier.TEXT, javaType = "java.lang.String"),
@BodyType.When(option =
"operation", value = "containercommit",
carriers =
Carrier.TEXT,
javaType =
"java.lang.String"),
@@ -108,7 +110,7 @@ import org.apache.camel.support.DefaultEndpoint;
@BodyType.When(option =
"operation", value = "containerlist",
carriers =
Carrier.LIST),
@BodyType.When(option =
"operation", value = "containerlog",
- carriers =
Carrier.POJO),
+ carriers =
Carrier.TEXT, javaType = "java.lang.String"),
@BodyType.When(option =
"operation", value = "containerpause",
carriers =
Carrier.UNCHANGED),
@BodyType.When(option =
"operation", value = "containerrestart",
@@ -124,11 +126,11 @@ import org.apache.camel.support.DefaultEndpoint;
@BodyType.When(option =
"operation", value = "containerunpause",
carriers =
Carrier.UNCHANGED),
@BodyType.When(option =
"operation", value = "containerwait",
- carriers =
Carrier.POJO,
- javaType =
"com.github.dockerjava.api.command.WaitContainerResultCallback"),
+ carriers =
Carrier.VALUE, javaType = "java.lang.Integer"),
@BodyType.When(option =
"operation", value = "execcreate", carriers = Carrier.POJO,
javaType =
"com.github.dockerjava.api.command.ExecCreateCmdResponse"),
- @BodyType.When(option =
"operation", value = "execstart", carriers = Carrier.POJO),
+ @BodyType.When(option =
"operation", value = "execstart", carriers = Carrier.TEXT,
+ javaType =
"java.lang.String"),
@BodyType.When(option =
"operation", value = "networkconnect",
carriers =
Carrier.UNCHANGED),
@BodyType.When(option =
"operation", value = "networkcreate",
@@ -139,11 +141,11 @@ import org.apache.camel.support.DefaultEndpoint;
description = "The response of docker-java
for the operation, such as the Version of"
+ " version and a List of
Containers for containerlist; the operations"
+ " without a response, such
as ping and containerstart, keep the body."
- + " The streaming operations
(imagepull, imagepush, containerattach,"
- + " containerlog,
containerwait, execstart) give the docker-java callback"
- + " once the stream is done,
not its frames; imagebuild gives the image id"
- + " of the last line of the
output, null when it is not the"
- + " Successfully built line"))
+ + " The streaming operations
wait until the stream is done:"
+ + " containerattach,
containerlog and execstart give the output of stdout"
+ + " and stderr as text,
containerwait the exit code of the container,"
+ + " imagebuild the id of the
built image, imagepull and imagepush the last"
+ + " item of the progress; a
failed build, pull or push fails the exchange"))
public class DockerEndpoint extends DefaultEndpoint implements
EndpointServiceLocation {
@UriParam
diff --git
a/components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java
b/components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java
index 62b329a5b502..b0c7917c3476 100644
---
a/components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java
+++
b/components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java
@@ -76,7 +76,8 @@ public class DockerEventsConsumer extends DefaultConsumer {
}
/**
- * Determine the point in time to begin streaming events
+ * Determine the point in time to begin streaming events, in seconds since
the epoch as the Docker API expects. The
+ * initial range is in milliseconds.
*/
private long processInitialEvent() {
long currentTime = System.currentTimeMillis();
@@ -86,7 +87,7 @@ public class DockerEventsConsumer extends DefaultConsumer {
currentTime = currentTime - initialRange;
}
- return currentTime;
+ return TimeUnit.MILLISECONDS.toSeconds(currentTime);
}
@Override
diff --git
a/components/camel-docker/src/main/java/org/apache/camel/component/docker/producer/AsyncDockerProducer.java
b/components/camel-docker/src/main/java/org/apache/camel/component/docker/producer/AsyncDockerProducer.java
index 2f652cfa2af2..ce8986214dee 100644
---
a/components/camel-docker/src/main/java/org/apache/camel/component/docker/producer/AsyncDockerProducer.java
+++
b/components/camel-docker/src/main/java/org/apache/camel/component/docker/producer/AsyncDockerProducer.java
@@ -16,13 +16,15 @@
*/
package org.apache.camel.component.docker.producer;
+import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.InputStream;
+import java.nio.charset.Charset;
import java.util.Collections;
+import java.util.concurrent.atomic.AtomicReference;
import com.github.dockerjava.api.DockerClient;
import com.github.dockerjava.api.async.ResultCallback.Adapter;
-import com.github.dockerjava.api.async.ResultCallbackTemplate;
import com.github.dockerjava.api.command.AsyncDockerCmd;
import com.github.dockerjava.api.command.AttachContainerCmd;
import com.github.dockerjava.api.command.BuildImageCmd;
@@ -52,6 +54,7 @@ import org.apache.camel.component.docker.DockerHelper;
import org.apache.camel.component.docker.DockerOperation;
import org.apache.camel.component.docker.exception.DockerException;
import org.apache.camel.support.DefaultAsyncProducer;
+import org.apache.camel.support.ExchangeHelper;
import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -117,10 +120,10 @@ public class AsyncDockerProducer extends
DefaultAsyncProducer {
throw new DockerException("Invalid operation: " +
operation);
}
} catch (InterruptedException e) {
- LOG.error("Interrupted while processing", e);
Thread.currentThread().interrupt();
- } catch (DockerException e) {
- LOG.error(e.getMessage(), e);
+ exchange.setException(e);
+ } catch (Exception e) {
+ exchange.setException(e);
}
callback.done(false);
@@ -130,7 +133,7 @@ public class AsyncDockerProducer extends
DefaultAsyncProducer {
private void runAsyncImageBuild(Exchange exchange, Message message,
DockerClient client)
throws DockerException, InterruptedException {
// result contain an image id value
- try (BuildImageCmd cmd = executeBuildImageRequest(client, message)) {
+ try (BuildImageCmd cmd = executeBuildImageRequest(exchange, client,
message)) {
BuildImageResultCallback item = cmd.exec(new
BuildImageResultCallback() {
@Override
@@ -138,26 +141,34 @@ public class AsyncDockerProducer extends
DefaultAsyncProducer {
super.onNext(item);
LOG.trace("build image callback {}", item);
-
- exchange.getIn().setBody(item.getImageId());
}
});
- setResponse(exchange, item);
+ item.awaitCompletion();
+ // the image id is in the Successfully built line, which is not
the last line of a tagged build
+ exchange.getIn().setBody(item.awaitImageId());
}
}
private void runAsyncWithFrameResponse(Exchange exchange,
AsyncDockerCmd<?, Frame> cmd) throws InterruptedException {
+ ByteArrayOutputStream output = new ByteArrayOutputStream();
Adapter<Frame> item = cmd.exec(new Adapter<Frame>() {
@Override
public void onNext(Frame item) {
LOG.trace("running framed callback {}", item);
super.onNext(item);
+
+ byte[] payload = item.getPayload();
+ if (payload != null) {
+ output.writeBytes(payload);
+ }
}
});
- setResponse(exchange, item);
+ item.awaitCompletion();
+ // the frames of stdout and stderr in the order they were received
+
exchange.getIn().setBody(output.toString(Charset.forName(ExchangeHelper.getCharsetName(exchange))));
}
private void runAsyncExecStart(Exchange exchange, Message message,
DockerClient client) throws InterruptedException {
@@ -178,14 +189,8 @@ public class AsyncDockerProducer extends
DefaultAsyncProducer {
});
- setResponse(exchange, item);
- }
- }
-
- private void setResponse(Exchange exchange, ResultCallbackTemplate item)
throws InterruptedException {
- if (item != null) {
- exchange.getIn().setBody(item);
item.awaitCompletion();
+ exchange.getIn().setBody(item.awaitStatusCode());
}
}
@@ -201,24 +206,36 @@ public class AsyncDockerProducer extends
DefaultAsyncProducer {
}
}
- private void runAsyncPush(Exchange exchange, Message message, DockerClient
client) throws InterruptedException {
+ private void runAsyncPush(Exchange exchange, Message message, DockerClient
client)
+ throws InterruptedException, DockerException {
try (PushImageCmd cmd = executePushImageRequest(client, message)) {
+ AtomicReference<PushResponseItem> last = new AtomicReference<>();
+ AtomicReference<String> error = new AtomicReference<>();
Adapter<PushResponseItem> item = cmd.exec(new
Adapter<PushResponseItem>() {
@Override
public void onNext(PushResponseItem item) {
super.onNext(item);
LOG.trace("push image callback {}", item);
+
+ last.set(item);
+ if (item.isErrorIndicated()) {
+ error.compareAndSet(null, item.getError());
+ }
}
});
- setResponse(exchange, item);
+ item.awaitCompletion();
+ if (error.get() != null) {
+ throw new DockerException("Could not push image: " +
error.get());
+ }
+ exchange.getIn().setBody(last.get());
}
}
private void runAsyncPull(Message message, DockerClient client, Exchange
exchange) throws InterruptedException {
try (PullImageCmd cmd = executePullImageRequest(client, message)) {
-
+ AtomicReference<PullResponseItem> last = new AtomicReference<>();
PullImageResultCallback item = cmd.exec(new
PullImageResultCallback() {
@Override
public void onNext(PullResponseItem item) {
@@ -226,17 +243,21 @@ public class AsyncDockerProducer extends
DefaultAsyncProducer {
LOG.trace("pull image callback {}", item);
+ last.set(item);
}
});
- setResponse(exchange, item);
+ // fails when the pull did not succeed
+ item.awaitCompletion();
+ exchange.getIn().setBody(last.get());
}
}
/**
* Produces a build image request
*/
- private BuildImageCmd executeBuildImageRequest(DockerClient client,
Message message) throws DockerException {
+ private BuildImageCmd executeBuildImageRequest(Exchange exchange,
DockerClient client, Message message)
+ throws DockerException {
LOG.debug("Executing Docker Build Image Request");
@@ -244,12 +265,18 @@ public class AsyncDockerProducer extends
DefaultAsyncProducer {
BuildImageCmd buildImageCmd;
- if (body instanceof InputStream) {
- buildImageCmd = client.buildImageCmd((InputStream) body);
- } else if (body instanceof File) {
- buildImageCmd = client.buildImageCmd((File) body);
+ if (body instanceof File file) {
+ buildImageCmd = client.buildImageCmd(file);
} else {
- throw new DockerException("Unable to location source Image");
+ // the tar archive of the build context, such as a byte[] or the
file of a file consumer
+ InputStream is = body != null
+ ?
exchange.getContext().getTypeConverter().tryConvertTo(InputStream.class,
exchange, body) : null;
+ if (is == null) {
+ throw new DockerException(
+ "imagebuild needs the build context as an InputStream
(a tar archive) or a java.io.File, but the body is "
+ + (body != null ?
body.getClass().getName() : "null"));
+ }
+ buildImageCmd = client.buildImageCmd(is);
}
Boolean noCache =
DockerHelper.getProperty(DockerConstants.DOCKER_NO_CACHE, configuration,
message, Boolean.class);
diff --git
a/components/camel-docker/src/test/java/org/apache/camel/component/docker/consumer/DockerEventsConsumerTest.java
b/components/camel-docker/src/test/java/org/apache/camel/component/docker/consumer/DockerEventsConsumerTest.java
index 7ae455377ac8..25ccf1daad80 100644
---
a/components/camel-docker/src/test/java/org/apache/camel/component/docker/consumer/DockerEventsConsumerTest.java
+++
b/components/camel-docker/src/test/java/org/apache/camel/component/docker/consumer/DockerEventsConsumerTest.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.component.docker.consumer;
+import java.util.concurrent.TimeUnit;
+
import com.github.dockerjava.api.DockerClient;
import com.github.dockerjava.api.command.EventsCmd;
import org.apache.camel.CamelContext;
@@ -27,10 +29,12 @@ import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.test.junit6.CamelTestSupport;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.anyString;
/**
@@ -60,6 +64,17 @@ public class DockerEventsConsumerTest extends
CamelTestSupport {
mock.expectedMinimumMessageCount(1);
}
+ @Test
+ void sinceIsInSeconds() {
+ ArgumentCaptor<String> since = ArgumentCaptor.forClass(String.class);
+ Mockito.verify(eventsCmd).withSince(since.capture());
+
+ // the Docker API takes the since time in seconds, not in milliseconds
+ long now = TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis());
+ long value = Long.parseLong(since.getValue());
+ assertTrue(value <= now && value > now - 60, "since " + value + " is
not near " + now);
+ }
+
@Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
diff --git
a/components/camel-docker/src/test/java/org/apache/camel/component/docker/producer/AsyncDockerProducerTest.java
b/components/camel-docker/src/test/java/org/apache/camel/component/docker/producer/AsyncDockerProducerTest.java
new file mode 100644
index 000000000000..a24b9d0ecef8
--- /dev/null
+++
b/components/camel-docker/src/test/java/org/apache/camel/component/docker/producer/AsyncDockerProducerTest.java
@@ -0,0 +1,284 @@
+/*
+ * 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.docker.producer;
+
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.github.dockerjava.api.DockerClient;
+import com.github.dockerjava.api.async.ResultCallback;
+import com.github.dockerjava.api.command.AttachContainerCmd;
+import com.github.dockerjava.api.command.BuildImageCmd;
+import com.github.dockerjava.api.command.ExecStartCmd;
+import com.github.dockerjava.api.command.LogContainerCmd;
+import com.github.dockerjava.api.command.PullImageCmd;
+import com.github.dockerjava.api.command.PushImageCmd;
+import com.github.dockerjava.api.command.WaitContainerCmd;
+import com.github.dockerjava.api.exception.DockerClientException;
+import com.github.dockerjava.api.model.BuildResponseItem;
+import com.github.dockerjava.api.model.Frame;
+import com.github.dockerjava.api.model.PullResponseItem;
+import com.github.dockerjava.api.model.PushResponseItem;
+import com.github.dockerjava.api.model.StreamType;
+import com.github.dockerjava.api.model.WaitResponse;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.docker.DockerComponent;
+import org.apache.camel.component.docker.DockerConfiguration;
+import org.apache.camel.component.docker.exception.DockerException;
+import org.apache.camel.component.docker.util.DockerTestUtils;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.stubbing.Answer;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+
+/**
+ * The body and the failures of the streaming operations of the async producer.
+ */
+@ExtendWith(MockitoExtension.class)
+public class AsyncDockerProducerTest extends CamelTestSupport {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private final String host = "localhost";
+ private final Integer port = 2375;
+
+ @Mock
+ private DockerClient dockerClient;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ DockerConfiguration dockerConfiguration = new DockerConfiguration();
+
dockerConfiguration.setParameters(DockerTestUtils.getDefaultParameters(host,
port, dockerConfiguration));
+
+ DockerComponent dockerComponent = new
DockerComponent(dockerConfiguration);
+ dockerComponent.setClient(DockerTestUtils.getClientProfile(host, port,
dockerConfiguration), dockerClient);
+ camelContext.addComponent("docker", dockerComponent);
+
+ return camelContext;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ for (String operation : new String[] {
+ "imagebuild", "imagepull", "imagepush",
"containerattach", "containerlog", "containerwait",
+ "execstart" }) {
+ from("direct:" + operation)
+ .to("docker://" + operation + "?host=" + host +
"&port=" + port
+ +
"&containerId=abc&execId=def&repository=busybox&name=busybox");
+ }
+ }
+ };
+ }
+
+ @Test
+ void imageBuildGivesTheImageIdOfATaggedBuild() throws Exception {
+ BuildImageCmd cmd = Mockito.mock(BuildImageCmd.class);
+
Mockito.when(dockerClient.buildImageCmd(any(InputStream.class))).thenReturn(cmd);
+ Mockito.when(cmd.exec(any())).thenAnswer(complete(
+ json("{\"stream\":\"Step 1/1 : FROM busybox\\n\"}",
BuildResponseItem.class),
+ json("{\"stream\":\"Successfully built 4dbb3f2e9e1a\\n\"}",
BuildResponseItem.class),
+ json("{\"stream\":\"Successfully tagged camel:1.0\\n\"}",
BuildResponseItem.class)));
+
+ // a byte[] body is the tar archive of the build context
+ Exchange out = template.send("direct:imagebuild", e ->
e.getIn().setBody(new byte[] { 1, 2, 3 }));
+
+ assertNull(out.getException());
+ assertEquals("4dbb3f2e9e1a", out.getMessage().getBody());
+ Mockito.verify(dockerClient).buildImageCmd(any(InputStream.class));
+ }
+
+ @Test
+ void imageBuildFailsWithoutBuildContext() {
+ Exchange out = template.send("direct:imagebuild", e ->
e.getIn().setBody(null));
+
+ DockerException cause = assertInstanceOf(DockerException.class,
out.getException());
+ assertTrue(cause.getMessage().contains("imagebuild needs the build
context"), cause.getMessage());
+ Mockito.verify(dockerClient,
Mockito.never()).buildImageCmd(any(InputStream.class));
+ }
+
+ @Test
+ void imageBuildFailsWhenTheBuildFails() throws Exception {
+ BuildImageCmd cmd = Mockito.mock(BuildImageCmd.class);
+
Mockito.when(dockerClient.buildImageCmd(any(InputStream.class))).thenReturn(cmd);
+ Mockito.when(cmd.exec(any())).thenAnswer(complete(
+ json("{\"error\":\"The command '/bin/sh -c false' returned a
non-zero code: 1\"}",
+ BuildResponseItem.class)));
+
+ Exchange out = template.send("direct:imagebuild", e ->
e.getIn().setBody("tar".getBytes()));
+
+ DockerClientException cause =
assertInstanceOf(DockerClientException.class, out.getException());
+ assertTrue(cause.getMessage().contains("returned a non-zero code: 1"),
cause.getMessage());
+ }
+
+ @Test
+ void containerLogGivesTheOutput() {
+ LogContainerCmd cmd = Mockito.mock(LogContainerCmd.class);
+
Mockito.when(dockerClient.logContainerCmd(anyString())).thenReturn(cmd);
+ Mockito.when(cmd.exec(any())).thenAnswer(complete(
+ new Frame(StreamType.STDOUT, "hello
".getBytes(StandardCharsets.UTF_8)),
+ new Frame(StreamType.STDERR,
"wørld\n".getBytes(StandardCharsets.UTF_8))));
+
+ Exchange out = template.send("direct:containerlog", e ->
e.getIn().setBody("ignored"));
+
+ assertNull(out.getException());
+ assertEquals("hello wørld\n", out.getMessage().getBody());
+ }
+
+ @Test
+ void containerAttachGivesTheOutput() {
+ AttachContainerCmd cmd = Mockito.mock(AttachContainerCmd.class);
+
Mockito.when(dockerClient.attachContainerCmd(anyString())).thenReturn(cmd);
+ Mockito.when(cmd.exec(any())).thenAnswer(complete(new
Frame(StreamType.RAW, "attached".getBytes())));
+
+ Exchange out = template.send("direct:containerattach", e ->
e.getIn().setBody("ignored"));
+
+ assertNull(out.getException());
+ assertEquals("attached", out.getMessage().getBody());
+ }
+
+ @Test
+ void execStartGivesTheOutput() {
+ ExecStartCmd cmd = Mockito.mock(ExecStartCmd.class);
+ Mockito.when(dockerClient.execStartCmd(anyString())).thenReturn(cmd);
+ Mockito.when(cmd.exec(any())).thenAnswer(complete(new
Frame(StreamType.STDOUT, "total 0\n".getBytes())));
+
+ Exchange out = template.send("direct:execstart", e ->
e.getIn().setBody("ignored"));
+
+ assertNull(out.getException());
+ assertEquals("total 0\n", out.getMessage().getBody());
+ }
+
+ @Test
+ void execStartFailsWhenTheStreamFails() {
+ ExecStartCmd cmd = Mockito.mock(ExecStartCmd.class);
+ Mockito.when(dockerClient.execStartCmd(anyString())).thenReturn(cmd);
+ Mockito.when(cmd.exec(any())).thenAnswer(invocation -> {
+ ResultCallback<Frame> callback = invocation.getArgument(0);
+ callback.onError(new IllegalStateException("connection reset"));
+ return callback;
+ });
+
+ Exchange out = template.send("direct:execstart", e ->
e.getIn().setBody("ignored"));
+
+ IllegalStateException cause =
assertInstanceOf(IllegalStateException.class, out.getException());
+ assertEquals("connection reset", cause.getMessage());
+ }
+
+ @Test
+ void containerWaitGivesTheStatusCode() throws Exception {
+ WaitContainerCmd cmd = Mockito.mock(WaitContainerCmd.class);
+
Mockito.when(dockerClient.waitContainerCmd(anyString())).thenReturn(cmd);
+
Mockito.when(cmd.exec(any())).thenAnswer(complete(json("{\"StatusCode\":3}",
WaitResponse.class)));
+
+ Exchange out = template.send("direct:containerwait", e ->
e.getIn().setBody("ignored"));
+
+ assertNull(out.getException());
+ assertEquals(3, out.getMessage().getBody());
+ }
+
+ @Test
+ void imagePullGivesTheLastItem() throws Exception {
+ PullImageCmd cmd = Mockito.mock(PullImageCmd.class);
+ Mockito.when(dockerClient.pullImageCmd(anyString())).thenReturn(cmd);
+ Mockito.when(cmd.exec(any())).thenAnswer(complete(
+ json("{\"status\":\"Pulling from
library/busybox\",\"id\":\"latest\"}", PullResponseItem.class),
+ json("{\"status\":\"Status: Downloaded newer image for
busybox:latest\"}", PullResponseItem.class)));
+
+ Exchange out = template.send("direct:imagepull", e ->
e.getIn().setBody("ignored"));
+
+ assertNull(out.getException());
+ PullResponseItem item = assertInstanceOf(PullResponseItem.class,
out.getMessage().getBody());
+ assertEquals("Status: Downloaded newer image for busybox:latest",
item.getStatus());
+ }
+
+ @Test
+ void imagePullFailsWhenThePullFails() throws Exception {
+ PullImageCmd cmd = Mockito.mock(PullImageCmd.class);
+ Mockito.when(dockerClient.pullImageCmd(anyString())).thenReturn(cmd);
+ Mockito.when(cmd.exec(any())).thenAnswer(complete(
+ json("{\"status\":\"Pulling fs layer\",\"id\":\"latest\"}",
PullResponseItem.class)));
+
+ Exchange out = template.send("direct:imagepull", e ->
e.getIn().setBody("ignored"));
+
+ DockerClientException cause =
assertInstanceOf(DockerClientException.class, out.getException());
+ assertTrue(cause.getMessage().contains("Could not pull image"),
cause.getMessage());
+ }
+
+ @Test
+ void imagePushGivesTheLastItem() throws Exception {
+ PushImageCmd cmd = Mockito.mock(PushImageCmd.class);
+ Mockito.when(dockerClient.pushImageCmd(anyString())).thenReturn(cmd);
+ Mockito.when(cmd.exec(any())).thenAnswer(complete(
+ json("{\"status\":\"Pushed\",\"id\":\"3d2430473443\"}",
PushResponseItem.class),
+ json("{\"status\":\"latest: digest: sha256:abc size: 527\"}",
PushResponseItem.class)));
+
+ Exchange out = template.send("direct:imagepush", e ->
e.getIn().setBody("ignored"));
+
+ assertNull(out.getException());
+ PushResponseItem item = assertInstanceOf(PushResponseItem.class,
out.getMessage().getBody());
+ assertEquals("latest: digest: sha256:abc size: 527", item.getStatus());
+ }
+
+ @Test
+ void imagePushFailsWhenThePushFails() throws Exception {
+ PushImageCmd cmd = Mockito.mock(PushImageCmd.class);
+ Mockito.when(dockerClient.pushImageCmd(anyString())).thenReturn(cmd);
+ Mockito.when(cmd.exec(any())).thenAnswer(complete(
+ json("{\"error\":\"denied: requested access to the resource is
denied\"}", PushResponseItem.class)));
+
+ Exchange out = template.send("direct:imagepush", e ->
e.getIn().setBody("ignored"));
+
+ DockerException cause = assertInstanceOf(DockerException.class,
out.getException());
+ assertEquals("Could not push image: denied: requested access to the
resource is denied", cause.getMessage());
+ }
+
+ /**
+ * Gives the items to the callback of the command and completes it, as
docker-java does on its own thread.
+ */
+ @SafeVarargs
+ private static <T> Answer<Object> complete(T... items) {
+ return invocation -> {
+ ResultCallback<T> callback = invocation.getArgument(0);
+ for (T item : items) {
+ callback.onNext(item);
+ }
+ callback.onComplete();
+ return callback;
+ };
+ }
+
+ private static <T> T json(String json, Class<T> type) throws Exception {
+ return MAPPER.readValue(json, type);
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 8f5992b77a54..2a70b4f9ea07 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -1275,6 +1275,31 @@ changed fields of an update are now in the new
`CamelDebeziumUpdateDescription`
The `Struct` to `Map` type converter now also converts a `Struct` inside a
list or a map.
+=== camel-docker - the streaming operations give their result, and their
failures fail the exchange
+
+The operations `imagebuild`, `imagepull`, `imagepush`, `containerattach`,
`containerlog`, `containerwait` and
+`execstart` now fail the exchange when they fail. Before, an error such as a
missing build context was only logged and
+the exchange completed with the body unchanged. A failed build or push,
reported by Docker in the output of the
+operation, now also fails the exchange.
+
+The body after these operations changed:
+
+* `imagebuild`: the id of the built image, also for a build with a tag.
Before, it was `null` for a tagged build.
+ The body can now also be a `byte[]` or another body that converts to an
`InputStream` (the tar archive of the build
+ context), besides an `InputStream` or a `java.io.File`.
+* `containerattach`, `containerlog` and `execstart`: the output of stdout and
stderr as a `String`. Before, the body
+ was the docker-java callback, and the output was lost.
+* `containerwait`: the exit code of the container as an `Integer`. Before, it
was the docker-java callback.
+* `imagepull` and `imagepush`: the last `PullResponseItem` or
`PushResponseItem` of the progress. Before, it was the
+ docker-java callback.
+
+The `events` consumer now gives the start of the event stream to Docker in
seconds, as the Docker API expects, so the
+past events of `CamelDockerInitialRange` (in milliseconds) are received.
Before, the time was in milliseconds.
+
+`DockerConstants.DEFAULT_CMD_EXEC_FACTORY` now names the Netty factory, which
is the default of the `cmdExecFactory`
+option. Before, it named the Jersey factory. The default itself did not
change; camel-docker now adds the Netty
+native transports for arm64, so the default works on Apple silicon and Linux
on arm64.
+
=== camel-pubnub - FIRE and SETSTATE with an empty body fail without calling
PubNub
The `FIRE` and `SETSTATE` operations with an empty body now fail with a
`CamelException` (`Can not fire empty
diff --git
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DockerEndpointBuilderFactory.java
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DockerEndpointBuilderFactory.java
index e8c53f18802b..a5670f65be9a 100644
---
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DockerEndpointBuilderFactory.java
+++
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DockerEndpointBuilderFactory.java
@@ -1874,7 +1874,8 @@ public interface DockerEndpointBuilderFactory {
return "CamelDockerNoPrune";
}
/**
- * The initial range.
+ * The time in milliseconds before the start of the events consumer
from
+ * which it also receives the past events.
*
* The option is a: {@code Long} type.
*
