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 28f5ef2ba89e CAMEL-26233: camel-opensearch - send a built request as
it is, keep the whole JSON search body, take the DeleteIndex name from a String
body, fail a null Index/Update body and read a file as JSON (#29075)
28f5ef2ba89e is described below
commit 28f5ef2ba89e573640903c08a32e563d941bee11
Author: Andrea Cosentino <[email protected]>
AuthorDate: Tue Nov 3 12:11:49 2026 +0100
CAMEL-26233: camel-opensearch - send a built request as it is, keep the
whole JSON search body, take the DeleteIndex name from a String body, fail a
null Index/Update body and read a file as JSON (#29075)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/catalog/components/opensearch.json | 2 +-
.../camel/catalog/docs/opensearch-component.adoc | 31 +-
.../OpensearchActionRequestConverterLoader.java | 8 +
.../camel/component/opensearch/opensearch.json | 2 +-
.../src/main/docs/opensearch-component.adoc | 31 +-
.../component/opensearch/OpensearchEndpoint.java | 30 +-
.../component/opensearch/OpensearchProducer.java | 24 +-
.../OpensearchActionRequestConverter.java | 135 ++++++--
.../opensearch/OpensearchProducerBodyTest.java | 338 +++++++++++++++++++++
.../OpensearchActionRequestConverterTest.java | 133 ++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 19 ++
11 files changed, 682 insertions(+), 71 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/opensearch.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/opensearch.json
index aca514e6dbe7..146664596a51 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/opensearch.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/opensearch.json
@@ -53,7 +53,7 @@
"CamelOpensearchFrom": { "index": 8, "kind": "header", "displayName": "",
"group": "producer", "label": "", "required": false, "javaType": "Integer",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The starting index of the response.", "constantName":
"org.apache.camel.component.opensearch.OpensearchConstants#PARAM_FROM" }
},
"bodyTypes": {
- "producerIn": { "carriers": [ "text", "bytes", "stream", "map", "list",
"pojo" ], "byOption": [ { "option": "operation", "value": "Index", "carriers":
[ "text", "bytes", "stream", "map", "pojo" ] }, { "option": "operation",
"value": "Update", "carriers": [ "text", "bytes", "stream", "map", "pojo" ] },
{ "option": "operation", "value": "Bulk", "carriers": [ "list", "pojo" ] }, {
"option": "operation", "value": "GetById", "carriers": [ "text", "pojo" ] }, {
"option": "operation", "valu [...]
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "map",
"list", "pojo" ], "byOption": [ { "option": "operation", "value": "Index",
"carriers": [ "text", "bytes", "stream", "file", "map", "pojo" ] }, { "option":
"operation", "value": "Update", "carriers": [ "text", "bytes", "stream",
"file", "map", "pojo" ] }, { "option": "operation", "value": "Bulk",
"carriers": [ "list", "pojo" ] }, { "option": "operation", "value": "GetById",
"carriers": [ "text", "pojo" ] }, { "opt [...]
"producerOut": { "carriers": [ "text", "value", "pojo", "list", "iterator"
], "byOption": [ { "option": "operation", "value": "Index", "carriers": [
"text" ], "javaType": "java.lang.String" }, { "option": "operation", "value":
"Update", "carriers": [ "text" ], "javaType": "java.lang.String" }, { "option":
"operation", "value": "Bulk", "carriers": [ "list" ] }, { "option":
"operation", "value": "GetById", "carriers": [ "pojo" ], "javaType":
"org.opensearch.client.opensearch.core.GetRe [...]
},
"properties": {
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/opensearch-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/opensearch-component.adoc
index 5c4f078dbf16..0059906dcada 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/opensearch-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/opensearch-component.adoc
@@ -50,37 +50,44 @@ Set an endpoint URI option or exchange header with a key of
"operation"
and a value set to one of the following. Some operations also require
other parameters or the message body to be set.
+A request of the OpenSearch client in the message body, such as an
`IndexRequest`, a `GetRequest` or a `SearchRequest`,
+sets the operation by its type, whatever the option or header say, and is sent
as it is: its index, id and the rest are
+not changed by the headers or options.
+
[width="100%",cols="10%,10%,80%",options="header",]
|===
|operation |message body |description
-|`Index` |`Map`, `String`, `byte[]`, `Reader`, `InputStream` or
`IndexRequest.Builder` content to index |Adds content to an index and returns
the content's indexId in the body.
+|`Index` |`Map`, `String`, `byte[]`, `Reader`, `InputStream`, a file
(`java.io.File`, `Path` or the file of the file component), `IndexRequest` or
`IndexRequest.Builder` content to index |Adds content to an index and returns
the content's indexId in the body.
+A file is read as the JSON of the document, and a null body fails.
You can set the name of the target index by setting the message header with
the key "indexName".
You can set the indexId by setting the message header with
the key "indexId".
-|`GetById` |`String` or `GetRequest.Builder` index id of content to retrieve
|Retrieves the document corresponding to the given index id and returns a
GetResponse object in the body.
+|`GetById` |`String`, `GetRequest` or `GetRequest.Builder` index id of content
to retrieve |Retrieves the document corresponding to the given index id and
returns a GetResponse object in the body.
You can set the name of the target index by setting the message header with
the key "indexName".
You can set the type of document by setting the message header with
the key "documentClass".
-|`Delete` |`String` or `DeleteRequest.Builder` index id of content to delete
|Deletes the specified indexName and returns a Result object in the body.
+|`Delete` |`String`, `DeleteRequest` or `DeleteRequest.Builder` index id of
content to delete |Deletes the specified indexName and returns a Result object
in the body.
You can set the name of the target index by setting the message header with
the key "indexName".
-|`DeleteIndex` |`String` (its text is not used) or
`DeleteIndexRequest.Builder` |Deletes the index named by the indexName header
or option, not the one in a `String` body, and returns a `Boolean` in the body
that is true when the deletion is acknowledged.
+|`DeleteIndex` |`String` name of the index, `DeleteIndexRequest` or
`DeleteIndexRequest.Builder` |Deletes the index named by the indexName header
or option, or else the one named by a `String` body, and returns a `Boolean` in
the body that is true when the deletion is acknowledged.
You can set the name of the target index by setting the message header with
the key "indexName".
-|`Bulk` |`Iterable` or `BulkRequest.Builder` of any type that is already
accepted (DeleteOperation.Builder for delete operation, UpdateOperation.Builder
for update operation, CreateOperation.Builder for create operation, byte[],
InputStream, String, Reader, Map or any document type for index operation) |
Adds/Updates/Deletes content from/to an index and returns a
List<BulkResponseItem> object in the body
+|`Bulk` |`Iterable`, `BulkRequest` or `BulkRequest.Builder` of any type that
is already accepted (DeleteOperation.Builder for delete operation,
UpdateOperation.Builder for update operation, CreateOperation.Builder for
create operation, byte[], InputStream, String, Reader, Map or any document type
for index operation) | Adds/Updates/Deletes content from/to an index and
returns a List<BulkResponseItem> object in the body
You can set the name of the target index by setting the message header with
the key "indexName".
-|`Search` |`Map`, `String` or `SearchRequest.Builder` |Search the content with
the map of query string.
+|`Search` |`Map`, `String`, `SearchRequest` or `SearchRequest.Builder` |Search
the content with the search request as a JSON `String` or a `Map`: its query,
size, from, sort, aggregations, _source and the rest are used.
+A JSON object with a single member other than query, such as `{"match":
{...}}`, is the query alone.
You can set the name of the target index by setting the message header with
the key "indexName".
You can set the number of hits to return by setting the message header with
the key "size".
You can set the starting document offset by setting the message header with
the key "from".
+The size and from headers or options, when set, override those of the body.
-|`MultiSearch` |`MsearchRequest.Builder` |Multiple search in one
+|`MultiSearch` |`MsearchRequest` or `MsearchRequest.Builder` |Multiple search
in one
-|`MultiGet` |`Iterable<String>` or `MgetRequest.Builder` the id of the
document to retrieve |Multiple get in one
+|`MultiGet` |`Iterable<String>`, `MgetRequest` or `MgetRequest.Builder` the id
of the document to retrieve |Multiple get in one
You can set the name of the target index by setting the message header with
the key "indexName".
@@ -88,7 +95,8 @@ You can set the name of the target index by setting the
message header with the
You must set the name of the target index by setting the message header with
the key "indexName".
-|`Update` |`byte[]`, `InputStream`, `String`, `Reader`, `Map` or any document
type content to update |Updates content to an index and returns the content's
indexId in the body.
+|`Update` |`byte[]`, `InputStream`, `String`, `Reader`, `Map`, a file,
`UpdateRequest`, `UpdateRequest.Builder` or any document type content to update
|Updates content to an index and returns the content's indexId in the body.
+A file is read as JSON, and a null body fails.
You can set the name of the target index by setting the message header with
the key "indexName".
You can set the indexId by setting the message header with
the key "indexId".
@@ -187,7 +195,8 @@ String indexId = template.requestBody("direct:index", map,
String.class);
=== Search Example
Searching on specific field(s) and value use the Operation ´Search´.
-Pass in the query JSON String or the Map
+Pass in the search request as a JSON String or a Map: its size, sort,
aggregations and the other members
+are used with its query.
[tabs]
====
@@ -228,7 +237,7 @@ YAML::
._Java-only: Java test API (ProducerTemplate)_
[source,java]
----
-String query = "{\"query\":{\"match\":{\"doc.content\":\"new release of
ApacheCamel\"}}}";
+String query = "{\"query\":{\"match\":{\"doc.content\":\"new release of
ApacheCamel\"}},\"size\":5}";
HitsMetadata<?> response = template.requestBody("direct:search", query,
HitsMetadata.class);
----
diff --git
a/components/camel-opensearch/src/generated/java/org/apache/camel/component/opensearch/converter/OpensearchActionRequestConverterLoader.java
b/components/camel-opensearch/src/generated/java/org/apache/camel/component/opensearch/converter/OpensearchActionRequestConverterLoader.java
index fdb86d331c01..43489a882424 100644
---
a/components/camel-opensearch/src/generated/java/org/apache/camel/component/opensearch/converter/OpensearchActionRequestConverterLoader.java
+++
b/components/camel-opensearch/src/generated/java/org/apache/camel/component/opensearch/converter/OpensearchActionRequestConverterLoader.java
@@ -84,6 +84,14 @@ public final class OpensearchActionRequestConverterLoader
implements TypeConvert
}
return answer;
});
+ addTypeConverter(registry,
org.opensearch.client.opensearch.core.MsearchRequest.Builder.class,
java.lang.Object.class, false,
+ (type, exchange, value) -> {
+ Object answer =
org.apache.camel.component.opensearch.converter.OpensearchActionRequestConverter.toMsearchRequestBuilder(value,
exchange);
+ if (false && answer == null) {
+ answer = Void.class;
+ }
+ return answer;
+ });
addTypeConverter(registry,
org.opensearch.client.opensearch.core.SearchRequest.Builder.class,
java.lang.Object.class, false,
(type, exchange, value) -> {
Object answer =
org.apache.camel.component.opensearch.converter.OpensearchActionRequestConverter.toSearchRequestBuilder(value,
exchange);
diff --git
a/components/camel-opensearch/src/generated/resources/META-INF/org/apache/camel/component/opensearch/opensearch.json
b/components/camel-opensearch/src/generated/resources/META-INF/org/apache/camel/component/opensearch/opensearch.json
index aca514e6dbe7..146664596a51 100644
---
a/components/camel-opensearch/src/generated/resources/META-INF/org/apache/camel/component/opensearch/opensearch.json
+++
b/components/camel-opensearch/src/generated/resources/META-INF/org/apache/camel/component/opensearch/opensearch.json
@@ -53,7 +53,7 @@
"CamelOpensearchFrom": { "index": 8, "kind": "header", "displayName": "",
"group": "producer", "label": "", "required": false, "javaType": "Integer",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The starting index of the response.", "constantName":
"org.apache.camel.component.opensearch.OpensearchConstants#PARAM_FROM" }
},
"bodyTypes": {
- "producerIn": { "carriers": [ "text", "bytes", "stream", "map", "list",
"pojo" ], "byOption": [ { "option": "operation", "value": "Index", "carriers":
[ "text", "bytes", "stream", "map", "pojo" ] }, { "option": "operation",
"value": "Update", "carriers": [ "text", "bytes", "stream", "map", "pojo" ] },
{ "option": "operation", "value": "Bulk", "carriers": [ "list", "pojo" ] }, {
"option": "operation", "value": "GetById", "carriers": [ "text", "pojo" ] }, {
"option": "operation", "valu [...]
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "map",
"list", "pojo" ], "byOption": [ { "option": "operation", "value": "Index",
"carriers": [ "text", "bytes", "stream", "file", "map", "pojo" ] }, { "option":
"operation", "value": "Update", "carriers": [ "text", "bytes", "stream",
"file", "map", "pojo" ] }, { "option": "operation", "value": "Bulk",
"carriers": [ "list", "pojo" ] }, { "option": "operation", "value": "GetById",
"carriers": [ "text", "pojo" ] }, { "opt [...]
"producerOut": { "carriers": [ "text", "value", "pojo", "list", "iterator"
], "byOption": [ { "option": "operation", "value": "Index", "carriers": [
"text" ], "javaType": "java.lang.String" }, { "option": "operation", "value":
"Update", "carriers": [ "text" ], "javaType": "java.lang.String" }, { "option":
"operation", "value": "Bulk", "carriers": [ "list" ] }, { "option":
"operation", "value": "GetById", "carriers": [ "pojo" ], "javaType":
"org.opensearch.client.opensearch.core.GetRe [...]
},
"properties": {
diff --git
a/components/camel-opensearch/src/main/docs/opensearch-component.adoc
b/components/camel-opensearch/src/main/docs/opensearch-component.adoc
index 5c4f078dbf16..0059906dcada 100644
--- a/components/camel-opensearch/src/main/docs/opensearch-component.adoc
+++ b/components/camel-opensearch/src/main/docs/opensearch-component.adoc
@@ -50,37 +50,44 @@ Set an endpoint URI option or exchange header with a key of
"operation"
and a value set to one of the following. Some operations also require
other parameters or the message body to be set.
+A request of the OpenSearch client in the message body, such as an
`IndexRequest`, a `GetRequest` or a `SearchRequest`,
+sets the operation by its type, whatever the option or header say, and is sent
as it is: its index, id and the rest are
+not changed by the headers or options.
+
[width="100%",cols="10%,10%,80%",options="header",]
|===
|operation |message body |description
-|`Index` |`Map`, `String`, `byte[]`, `Reader`, `InputStream` or
`IndexRequest.Builder` content to index |Adds content to an index and returns
the content's indexId in the body.
+|`Index` |`Map`, `String`, `byte[]`, `Reader`, `InputStream`, a file
(`java.io.File`, `Path` or the file of the file component), `IndexRequest` or
`IndexRequest.Builder` content to index |Adds content to an index and returns
the content's indexId in the body.
+A file is read as the JSON of the document, and a null body fails.
You can set the name of the target index by setting the message header with
the key "indexName".
You can set the indexId by setting the message header with
the key "indexId".
-|`GetById` |`String` or `GetRequest.Builder` index id of content to retrieve
|Retrieves the document corresponding to the given index id and returns a
GetResponse object in the body.
+|`GetById` |`String`, `GetRequest` or `GetRequest.Builder` index id of content
to retrieve |Retrieves the document corresponding to the given index id and
returns a GetResponse object in the body.
You can set the name of the target index by setting the message header with
the key "indexName".
You can set the type of document by setting the message header with
the key "documentClass".
-|`Delete` |`String` or `DeleteRequest.Builder` index id of content to delete
|Deletes the specified indexName and returns a Result object in the body.
+|`Delete` |`String`, `DeleteRequest` or `DeleteRequest.Builder` index id of
content to delete |Deletes the specified indexName and returns a Result object
in the body.
You can set the name of the target index by setting the message header with
the key "indexName".
-|`DeleteIndex` |`String` (its text is not used) or
`DeleteIndexRequest.Builder` |Deletes the index named by the indexName header
or option, not the one in a `String` body, and returns a `Boolean` in the body
that is true when the deletion is acknowledged.
+|`DeleteIndex` |`String` name of the index, `DeleteIndexRequest` or
`DeleteIndexRequest.Builder` |Deletes the index named by the indexName header
or option, or else the one named by a `String` body, and returns a `Boolean` in
the body that is true when the deletion is acknowledged.
You can set the name of the target index by setting the message header with
the key "indexName".
-|`Bulk` |`Iterable` or `BulkRequest.Builder` of any type that is already
accepted (DeleteOperation.Builder for delete operation, UpdateOperation.Builder
for update operation, CreateOperation.Builder for create operation, byte[],
InputStream, String, Reader, Map or any document type for index operation) |
Adds/Updates/Deletes content from/to an index and returns a
List<BulkResponseItem> object in the body
+|`Bulk` |`Iterable`, `BulkRequest` or `BulkRequest.Builder` of any type that
is already accepted (DeleteOperation.Builder for delete operation,
UpdateOperation.Builder for update operation, CreateOperation.Builder for
create operation, byte[], InputStream, String, Reader, Map or any document type
for index operation) | Adds/Updates/Deletes content from/to an index and
returns a List<BulkResponseItem> object in the body
You can set the name of the target index by setting the message header with
the key "indexName".
-|`Search` |`Map`, `String` or `SearchRequest.Builder` |Search the content with
the map of query string.
+|`Search` |`Map`, `String`, `SearchRequest` or `SearchRequest.Builder` |Search
the content with the search request as a JSON `String` or a `Map`: its query,
size, from, sort, aggregations, _source and the rest are used.
+A JSON object with a single member other than query, such as `{"match":
{...}}`, is the query alone.
You can set the name of the target index by setting the message header with
the key "indexName".
You can set the number of hits to return by setting the message header with
the key "size".
You can set the starting document offset by setting the message header with
the key "from".
+The size and from headers or options, when set, override those of the body.
-|`MultiSearch` |`MsearchRequest.Builder` |Multiple search in one
+|`MultiSearch` |`MsearchRequest` or `MsearchRequest.Builder` |Multiple search
in one
-|`MultiGet` |`Iterable<String>` or `MgetRequest.Builder` the id of the
document to retrieve |Multiple get in one
+|`MultiGet` |`Iterable<String>`, `MgetRequest` or `MgetRequest.Builder` the id
of the document to retrieve |Multiple get in one
You can set the name of the target index by setting the message header with
the key "indexName".
@@ -88,7 +95,8 @@ You can set the name of the target index by setting the
message header with the
You must set the name of the target index by setting the message header with
the key "indexName".
-|`Update` |`byte[]`, `InputStream`, `String`, `Reader`, `Map` or any document
type content to update |Updates content to an index and returns the content's
indexId in the body.
+|`Update` |`byte[]`, `InputStream`, `String`, `Reader`, `Map`, a file,
`UpdateRequest`, `UpdateRequest.Builder` or any document type content to update
|Updates content to an index and returns the content's indexId in the body.
+A file is read as JSON, and a null body fails.
You can set the name of the target index by setting the message header with
the key "indexName".
You can set the indexId by setting the message header with
the key "indexId".
@@ -187,7 +195,8 @@ String indexId = template.requestBody("direct:index", map,
String.class);
=== Search Example
Searching on specific field(s) and value use the Operation ´Search´.
-Pass in the query JSON String or the Map
+Pass in the search request as a JSON String or a Map: its size, sort,
aggregations and the other members
+are used with its query.
[tabs]
====
@@ -228,7 +237,7 @@ YAML::
._Java-only: Java test API (ProducerTemplate)_
[source,java]
----
-String query = "{\"query\":{\"match\":{\"doc.content\":\"new release of
ApacheCamel\"}}}";
+String query = "{\"query\":{\"match\":{\"doc.content\":\"new release of
ApacheCamel\"}},\"size\":5}";
HitsMetadata<?> response = template.requestBody("direct:search", query,
HitsMetadata.class);
----
diff --git
a/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchEndpoint.java
b/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchEndpoint.java
index 8b267d3ae984..568d53dfae55 100644
---
a/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchEndpoint.java
+++
b/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchEndpoint.java
@@ -37,16 +37,16 @@ import org.opensearch.client.opensearch.OpenSearchClient;
syntax = "opensearch:clusterName", producerOnly = true,
category = { Category.SEARCH, Category.MONITORING }, headersClass
= OpensearchConstants.class)
@BodyTypes(producerIn = @BodyType(value = {
- Carrier.TEXT, Carrier.BYTES, Carrier.STREAM, Carrier.MAP,
Carrier.LIST, Carrier.POJO },
+ Carrier.TEXT, Carrier.BYTES, Carrier.STREAM, Carrier.FILE,
Carrier.MAP, Carrier.LIST, Carrier.POJO },
byOption = {
@BodyType.When(option = "operation",
value = "Index",
carriers = {
- Carrier.TEXT,
Carrier.BYTES, Carrier.STREAM, Carrier.MAP,
- Carrier.POJO
}),
+ Carrier.TEXT,
Carrier.BYTES, Carrier.STREAM, Carrier.FILE,
+ Carrier.MAP,
Carrier.POJO }),
@BodyType.When(option = "operation",
value = "Update",
carriers = {
- Carrier.TEXT,
Carrier.BYTES, Carrier.STREAM, Carrier.MAP,
- Carrier.POJO
}),
+ Carrier.TEXT,
Carrier.BYTES, Carrier.STREAM, Carrier.FILE,
+ Carrier.MAP,
Carrier.POJO }),
@BodyType.When(option = "operation",
value = "Bulk",
carriers = {
Carrier.LIST, Carrier.POJO }),
@BodyType.When(option = "operation",
value = "GetById",
@@ -63,16 +63,18 @@ import org.opensearch.client.opensearch.OpenSearchClient;
carriers = {
Carrier.TEXT, Carrier.MAP, Carrier.POJO }),
@BodyType.When(option = "operation",
value = "Exists", carriers = Carrier.ANY),
@BodyType.When(option = "operation",
value = "Ping", carriers = Carrier.ANY) },
- description = "By the operation, set by the
operation option or the CamelOpensearchOperation header."
- + " Index and Update take the
document as JSON text, bytes or a stream, a Map"
- + " or a POJO (written as
JSON), or the request builder of the OpenSearch"
- + " client; text that is not
JSON, a List, a java.io.File or no body fail."
+ description = "By the operation, set by the
operation option or the CamelOpensearchOperation header,"
+ + " or by the request of the
OpenSearch client in the body. Index and Update"
+ + " take the document as JSON
text, bytes, a stream or a file of it, or a Map"
+ + " or a POJO (written as
JSON); text that is not JSON, a List or no body fail."
+ " Update takes the partial
document, or the Update API form {\"doc\": {...}}. GetById and Delete take the
id as text, Bulk and MultiGet a List"
- + " of documents or ids,
Search a query as JSON text or a Map (only its query"
- + " member when it has one),
and MultiSearch only an MsearchRequest.Builder;"
- + " each also takes its
request builder. DeleteIndex needs a text body but"
- + " deletes the index of the
CamelOpensearchIndexName header or indexName, not the one"
- + " in the body. Exists and
Ping do not use the body"),
+ + " of documents or ids,
Search the search request as JSON text or a Map, with"
+ + " its query, size, sort,
aggregations and the rest (an object with a single"
+ + " member other than query is
the query alone), and DeleteIndex the name of"
+ + " the index as text, used
when neither the CamelOpensearchIndexName header"
+ + " nor indexName is set. Each
also takes its request of the OpenSearch"
+ + " client, or the builder of
it, which is sent as it is; MultiSearch takes"
+ + " only those. Exists and
Ping do not use the body"),
producerOut = @BodyType(value = { Carrier.TEXT, Carrier.VALUE,
Carrier.POJO, Carrier.LIST, Carrier.ITERATOR },
byOption = {
@BodyType.When(option =
"operation", value = "Index", carriers = Carrier.TEXT,
diff --git
a/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchProducer.java
b/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchProducer.java
index 16f8a2d00b47..872a9055e2e2 100644
---
a/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchProducer.java
+++
b/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchProducer.java
@@ -237,7 +237,7 @@ class OpensearchProducer extends DefaultAsyncProducer {
SearchRequest.Builder searchRequestBuilder =
message.getBody(SearchRequest.Builder.class);
if (searchRequestBuilder == null) {
throw new IllegalArgumentException(
- "Wrong body type. Only Map, String or
SearchRequest.Builder is allowed as a type");
+ "Wrong body type. Only Map, String,
SearchRequest or SearchRequest.Builder is allowed as a type");
}
// is it a scroll request ?
boolean useScroll = message.getHeader(PARAM_SCROLL,
configuration.isUseScroll(), Boolean.class);
@@ -291,7 +291,8 @@ class OpensearchProducer extends DefaultAsyncProducer {
private void processMultiGetAsync(ActionContext ctx, Class<?>
documentClass) throws IOException {
MgetRequest.Builder mgetRequestBuilder =
ctx.getMessage().getBody(MgetRequest.Builder.class);
if (mgetRequestBuilder == null) {
- throw new IllegalArgumentException("Wrong body type. Only
MgetRequest.Builder is allowed as a type");
+ throw new IllegalArgumentException(
+ "Wrong body type. Only Iterable, MgetRequest or
MgetRequest.Builder is allowed as a type");
}
onComplete(
ctx.getClient().mget(mgetRequestBuilder.build(), documentClass)
@@ -305,7 +306,8 @@ class OpensearchProducer extends DefaultAsyncProducer {
private void processMultiSearchAsync(ActionContext ctx, Class<?>
documentClass) throws IOException {
MsearchRequest.Builder msearchRequestBuilder =
ctx.getMessage().getBody(MsearchRequest.Builder.class);
if (msearchRequestBuilder == null) {
- throw new IllegalArgumentException("Wrong body type. Only
MsearchRequest.Builder is allowed as a type");
+ throw new IllegalArgumentException(
+ "Wrong body type. Only MsearchRequest or
MsearchRequest.Builder is allowed as a type");
}
onComplete(
ctx.getClient().msearch(msearchRequestBuilder.build(),
documentClass)
@@ -332,7 +334,7 @@ class OpensearchProducer extends DefaultAsyncProducer {
DeleteIndexRequest.Builder deleteIndexRequestBuilder =
ctx.getMessage().getBody(DeleteIndexRequest.Builder.class);
if (deleteIndexRequestBuilder == null) {
throw new IllegalArgumentException(
- "Wrong body type. Only String or
DeleteIndexRequest.Builder is allowed as a type");
+ "Wrong body type. Only String, DeleteIndexRequest or
DeleteIndexRequest.Builder is allowed as a type");
}
onComplete(
ctx.getClient().indices().delete(deleteIndexRequestBuilder.build())
@@ -347,7 +349,7 @@ class OpensearchProducer extends DefaultAsyncProducer {
DeleteRequest.Builder deleteRequestBuilder =
ctx.getMessage().getBody(DeleteRequest.Builder.class);
if (deleteRequestBuilder == null) {
throw new IllegalArgumentException(
- "Wrong body type. Only String or DeleteRequest.Builder is
allowed as a type");
+ "Wrong body type. Only String, DeleteRequest or
DeleteRequest.Builder is allowed as a type");
}
onComplete(
ctx.getClient().delete(deleteRequestBuilder.build())
@@ -362,7 +364,7 @@ class OpensearchProducer extends DefaultAsyncProducer {
BulkRequest.Builder bulkRequestBuilder =
ctx.getMessage().getBody(BulkRequest.Builder.class);
if (bulkRequestBuilder == null) {
throw new IllegalArgumentException(
- "Wrong body type. Only Iterable or BulkRequest.Builder is
allowed as a type");
+ "Wrong body type. Only Iterable, BulkRequest or
BulkRequest.Builder is allowed as a type");
}
onComplete(
ctx.getClient().bulk(bulkRequestBuilder.build())
@@ -377,7 +379,7 @@ class OpensearchProducer extends DefaultAsyncProducer {
GetRequest.Builder getRequestBuilder =
ctx.getMessage().getBody(GetRequest.Builder.class);
if (getRequestBuilder == null) {
throw new IllegalArgumentException(
- "Wrong body type. Only String or GetRequest.Builder is
allowed as a type");
+ "Wrong body type. Only String, GetRequest or
GetRequest.Builder is allowed as a type");
}
onComplete(
ctx.getClient().get(getRequestBuilder.build(), documentClass),
@@ -388,6 +390,10 @@ class OpensearchProducer extends DefaultAsyncProducer {
* Updates asynchronously a document.
*/
private void processUpdateAsync(ActionContext ctx, Class<?> documentClass)
throws IOException {
+ if (ctx.getMessage().getBody() == null) {
+ throw new IllegalArgumentException(
+ "The body of the Update operation cannot be null: it is
the update of the document, or its request");
+ }
var updateRequestBuilder =
ctx.getMessage().getBody(UpdateRequest.Builder.class);
onComplete(
ctx.getClient().update(updateRequestBuilder.build(),
documentClass)
@@ -399,6 +405,10 @@ class OpensearchProducer extends DefaultAsyncProducer {
* Indexes asynchronously a document.
*/
private void processIndexAsync(ActionContext ctx) throws IOException {
+ if (ctx.getMessage().getBody() == null) {
+ throw new IllegalArgumentException(
+ "The body of the Index operation cannot be null: it is the
document to index, or its request");
+ }
IndexRequest.Builder<?> indexRequestBuilder =
ctx.getMessage().getBody(IndexRequest.Builder.class);
onComplete(
ctx.getClient().index(indexRequestBuilder.build())
diff --git
a/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/converter/OpensearchActionRequestConverter.java
b/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/converter/OpensearchActionRequestConverter.java
index bdb6c711ec9c..7f25a453cb75 100644
---
a/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/converter/OpensearchActionRequestConverter.java
+++
b/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/converter/OpensearchActionRequestConverter.java
@@ -16,10 +16,12 @@
*/
package org.apache.camel.component.opensearch.converter;
+import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.Reader;
import java.io.StringReader;
+import java.nio.file.Path;
import java.util.Map;
import com.fasterxml.jackson.annotation.JsonInclude;
@@ -28,6 +30,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import org.apache.camel.Converter;
import org.apache.camel.Exchange;
+import org.apache.camel.WrappedFile;
import org.apache.camel.component.opensearch.OpensearchConstants;
import org.apache.camel.util.ObjectHelper;
import org.opensearch.client.json.JsonData;
@@ -39,6 +42,7 @@ import org.opensearch.client.opensearch.core.DeleteRequest;
import org.opensearch.client.opensearch.core.GetRequest;
import org.opensearch.client.opensearch.core.IndexRequest;
import org.opensearch.client.opensearch.core.MgetRequest;
+import org.opensearch.client.opensearch.core.MsearchRequest;
import org.opensearch.client.opensearch.core.SearchRequest;
import org.opensearch.client.opensearch.core.UpdateRequest;
import org.opensearch.client.opensearch.core.bulk.BulkOperation;
@@ -95,6 +99,11 @@ public final class OpensearchActionRequestConverter {
}
return builder;
}
+ if (document instanceof IndexRequest<?> indexRequest) {
+ // a built request is sent as it is, as a builder body is: the
producer gets a builder body without
+ // converting it, so the header rules of the builder above do not
apply to it either
+ return indexRequest.toBuilder();
+ }
JacksonJsonpMapper mapper = createMapper();
IndexRequest.Builder<Object> builder = new IndexRequest.Builder<>();
if (document instanceof byte[] byteArray) {
@@ -105,6 +114,10 @@ public final class OpensearchActionRequestConverter {
builder.document(JsonData.of(mapper.objectMapper().reader().readTree(new
StringReader(string)), mapper).toJson());
} else if (document instanceof Reader reader) {
builder.document(JsonData.of(mapper.objectMapper().reader().readTree(reader),
mapper).toJson());
+ } else if (isFile(document)) {
+ try (InputStream inputStream = readFile(document, exchange)) {
+
builder.document(JsonData.of(mapper.objectMapper().reader().readTree(inputStream),
mapper).toJson());
+ }
} else {
builder.document(document);
}
@@ -128,6 +141,10 @@ public final class OpensearchActionRequestConverter {
}
return builder;
}
+ if (document instanceof UpdateRequest<?, ?> updateRequest) {
+ // a built request is sent as it is, as a builder body is
+ return updateRequest.toBuilder();
+ }
JacksonJsonpMapper mapper = createMapper();
UpdateRequest.Builder<?, Object> builder = new
UpdateRequest.Builder<>();
if (document instanceof byte[] byteArray) {
@@ -141,6 +158,10 @@ public final class OpensearchActionRequestConverter {
document =
JsonData.of(mapper.objectMapper().reader().readTree(reader),
mapper).to(JsonNode.class);
} else if (document instanceof Map<?, ?> map) {
document = mapper.objectMapper().convertValue(map, JsonNode.class);
+ } else if (isFile(document)) {
+ try (InputStream inputStream = readFile(document, exchange)) {
+ document =
JsonData.of(mapper.objectMapper().reader().readTree(inputStream),
mapper).to(JsonNode.class);
+ }
}
if (document instanceof JsonNode jsonNode) {
@@ -167,6 +188,10 @@ public final class OpensearchActionRequestConverter {
if (document instanceof GetRequest.Builder getReqBuilder) {
return getReqBuilder;
}
+ if (document instanceof GetRequest getRequest) {
+ // a built request is sent as it is, as a builder body is
+ return getRequest.toBuilder();
+ }
if (document instanceof String string) {
return new GetRequest.Builder()
.index(exchange.getIn().getHeader(OpensearchConstants.PARAM_INDEX_NAME,
String.class))
@@ -180,6 +205,10 @@ public final class OpensearchActionRequestConverter {
if (document instanceof DeleteRequest.Builder deleteReqBuilder) {
return deleteReqBuilder;
}
+ if (document instanceof DeleteRequest deleteRequest) {
+ // a built request is sent as it is, as a builder body is
+ return deleteRequest.toBuilder();
+ }
if (document instanceof String string) {
return new DeleteRequest.Builder()
.index(exchange.getIn().getHeader(OpensearchConstants.PARAM_INDEX_NAME,
String.class))
@@ -193,9 +222,22 @@ public final class OpensearchActionRequestConverter {
if (document instanceof DeleteIndexRequest.Builder
deleteIdxReqBuilder) {
return deleteIdxReqBuilder;
}
- if (document instanceof String) {
- return new DeleteIndexRequest.Builder()
-
.index(exchange.getIn().getHeader(OpensearchConstants.PARAM_INDEX_NAME,
String.class));
+ if (document instanceof DeleteIndexRequest deleteIndexRequest) {
+ // a built request is sent as it is, as a builder body is
+ return deleteIndexRequest.toBuilder();
+ }
+ if (document instanceof String string) {
+ // the index of the header or of the indexName option, else the
one named by the body
+ String indexName =
exchange.getIn().getHeader(OpensearchConstants.PARAM_INDEX_NAME, String.class);
+ if (ObjectHelper.isEmpty(indexName)) {
+ indexName = string;
+ }
+ DeleteIndexRequest.Builder builder = new
DeleteIndexRequest.Builder();
+ if (ObjectHelper.isNotEmpty(indexName)) {
+ builder.index(indexName);
+ }
+ // with no name at all, the request fails when it is built, as the
index is required
+ return builder;
}
return null;
}
@@ -205,6 +247,10 @@ public final class OpensearchActionRequestConverter {
if (documents instanceof MgetRequest.Builder mgetReqBuilder) {
return mgetReqBuilder;
}
+ if (documents instanceof MgetRequest mgetRequest) {
+ // a built request is sent as it is, as a builder body is
+ return mgetRequest.toBuilder();
+ }
if (documents instanceof Iterable<?> documentIterable) {
MgetRequest.Builder builder = new MgetRequest.Builder();
builder.index(exchange.getIn().getHeader(OpensearchConstants.PARAM_INDEX_NAME,
String.class));
@@ -233,33 +279,20 @@ public final class OpensearchActionRequestConverter {
}
return searchReqBuilder;
}
- SearchRequest.Builder builder = new SearchRequest.Builder();
-
- // Only set up the indexName if the message header has the
- // setting
+ if (queryObject instanceof SearchRequest searchRequest) {
+ // a built request is sent as it is, as a builder body is
+ return searchRequest.toBuilder();
+ }
Integer size =
exchange.getIn().getHeader(OpensearchConstants.PARAM_SIZE, Integer.class);
Integer from =
exchange.getIn().getHeader(OpensearchConstants.PARAM_FROM, Integer.class);
- if (ObjectHelper.isNotEmpty(indexName)) {
- builder.index(indexName);
- }
+ JacksonJsonpMapper mapper = createMapper();
+ JsonNode json;
if (queryObject instanceof Map<?, ?> mapQuery) {
- // Remove 'query' prefix from the query object for backward
- // compatibility with Elasticsearch
- if (mapQuery.containsKey(OPENSEARCH_QUERY_DSL_PREFIX)) {
- mapQuery = (Map<?, ?>)
mapQuery.get(OPENSEARCH_QUERY_DSL_PREFIX);
- }
- queryObject = mapQuery;
+ json = mapper.objectMapper().valueToTree(mapQuery);
} else if (queryObject instanceof String queryString) {
- JacksonJsonpMapper mapper = createMapper();
- JsonNode jsonTextObject =
mapper.objectMapper().readValue(queryString, JsonNode.class);
- JsonNode parentJsonNode =
jsonTextObject.get(OPENSEARCH_QUERY_DSL_PREFIX);
- if (parentJsonNode != null) {
- queryString = parentJsonNode.toString();
- }
- mapper.objectMapper().reader().readTree(new
StringReader(queryString));
- queryObject =
JsonData.of(mapper.objectMapper().reader().readTree(new
StringReader(queryString)), mapper).toJson();
+ json = mapper.objectMapper().readValue(queryString,
JsonNode.class);
} else {
// Cannot convert the queryObject into SearchRequest
LOG.warn(
@@ -267,6 +300,22 @@ public final class OpensearchActionRequestConverter {
queryObject == null ? "null" :
queryObject.getClass().getName());
return null;
}
+
+ SearchRequest.Builder builder;
+ if (json != null && json.isObject() && json.size() == 1 &&
!json.has(OPENSEARCH_QUERY_DSL_PREFIX)) {
+ // a query alone, such as {"match": {...}}: an object with a
single member, other than query
+ builder = new SearchRequest.Builder().query(JsonData.of(json,
mapper).to(Query.class));
+ } else {
+ // the body of a search request: its query, size, from, sort,
aggregations, _source and the rest
+ builder = JsonData.of(json,
mapper).to(SearchRequest.class).toBuilder();
+ }
+
+ // Only set up the indexName if the message header has the
+ // setting
+ if (ObjectHelper.isNotEmpty(indexName)) {
+ builder.index(indexName);
+ }
+ // the size and from of the headers or of the options, when set,
override those of the body
if (size != null) {
builder.size(size);
}
@@ -274,16 +323,30 @@ public final class OpensearchActionRequestConverter {
builder.from(from);
}
- builder.query(JsonData.of(queryObject,
createMapper()).to(Query.class));
-
return builder;
}
+ @Converter
+ public static MsearchRequest.Builder toMsearchRequestBuilder(Object
queryObject, Exchange exchange) {
+ if (queryObject instanceof MsearchRequest.Builder msearchReqBuilder) {
+ return msearchReqBuilder;
+ }
+ if (queryObject instanceof MsearchRequest msearchRequest) {
+ // a built request is sent as it is, as a builder body is
+ return msearchRequest.toBuilder();
+ }
+ return null;
+ }
+
@Converter
public static BulkRequest.Builder toBulkRequestBuilder(Object documents,
Exchange exchange) throws IOException {
if (documents instanceof BulkRequest.Builder bulkReqBuilder) {
return bulkReqBuilder;
}
+ if (documents instanceof BulkRequest bulkRequest) {
+ // a built request is sent as it is, as a builder body is
+ return bulkRequest.toBuilder();
+ }
if (documents instanceof Iterable<?> documentIterable) {
BulkRequest.Builder builder = new BulkRequest.Builder();
builder.index(exchange.getIn().getHeader(OpensearchConstants.PARAM_INDEX_NAME,
String.class));
@@ -311,6 +374,26 @@ public final class OpensearchActionRequestConverter {
return null;
}
+ /**
+ * Whether the document is a file, a java.io.File, a Path or the file of
the file component, whose content is the
+ * JSON of the document.
+ */
+ private static boolean isFile(Object document) {
+ return document instanceof File || document instanceof Path ||
document instanceof WrappedFile;
+ }
+
+ /**
+ * Opens the content of the file, to read as JSON.
+ */
+ private static InputStream readFile(Object file, Exchange exchange) {
+ InputStream inputStream =
exchange.getContext().getTypeConverter().convertTo(InputStream.class, exchange,
file);
+ if (inputStream == null) {
+ throw new IllegalArgumentException(
+ "Cannot read the file of type " +
file.getClass().getName() + " as the JSON of the document");
+ }
+ return inputStream;
+ }
+
private static JacksonJsonpMapper createMapper() {
ObjectMapper objectMapper = new ObjectMapper()
.configure(SerializationFeature.INDENT_OUTPUT, false)
diff --git
a/components/camel-opensearch/src/test/java/org/apache/camel/component/opensearch/OpensearchProducerBodyTest.java
b/components/camel-opensearch/src/test/java/org/apache/camel/component/opensearch/OpensearchProducerBodyTest.java
new file mode 100644
index 000000000000..899533d77b01
--- /dev/null
+++
b/components/camel-opensearch/src/test/java/org/apache/camel/component/opensearch/OpensearchProducerBodyTest.java
@@ -0,0 +1,338 @@
+/*
+ * 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.opensearch;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.file.FileConsumer;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.opensearch.client.opensearch._types.Result;
+import org.opensearch.client.opensearch.core.BulkRequest;
+import org.opensearch.client.opensearch.core.DeleteRequest;
+import org.opensearch.client.opensearch.core.GetRequest;
+import org.opensearch.client.opensearch.core.GetResponse;
+import org.opensearch.client.opensearch.core.IndexRequest;
+import org.opensearch.client.opensearch.core.MgetRequest;
+import org.opensearch.client.opensearch.core.MsearchRequest;
+import org.opensearch.client.opensearch.core.SearchRequest;
+import org.opensearch.client.opensearch.core.UpdateRequest;
+import org.opensearch.client.opensearch.core.search.HitsMetadata;
+import org.opensearch.client.opensearch.indices.DeleteIndexRequest;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The request the producer sends for the bodies it takes, checked against a
node that records the requests and answers
+ * them as OpenSearch does.
+ */
+class OpensearchProducerBodyTest {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private static final String SHARDS =
"\"_shards\":{\"total\":1,\"successful\":1,\"failed\":0}";
+ private static final String WRITTEN =
"{\"_index\":\"idx\",\"_id\":\"1\",\"_version\":1,\"result\":\"%s\"," + SHARDS
+ +
",\"_seq_no\":0,\"_primary_term\":1}";
+ private static final String FOUND
+ =
"{\"_index\":\"idx\",\"_id\":\"1\",\"_version\":1,\"_seq_no\":0,\"_primary_term\":1,\"found\":true,"
+ + "\"_source\":{\"name\":\"camel\"}}";
+ private static final String HITS = "{\"took\":1,\"timed_out\":false," +
SHARDS
+ +
",\"hits\":{\"total\":{\"value\":0,\"relation\":\"eq\"},\"hits\":[]}}";
+
+ private final List<Recorded> requests = new CopyOnWriteArrayList<>();
+ private HttpServer server;
+ private CamelContext context;
+ private ProducerTemplate template;
+ private String uri;
+
+ @TempDir
+ Path dir;
+
+ @BeforeEach
+ void start() throws Exception {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/", this::answer);
+ server.start();
+ uri = "opensearch:test-cluster?hostAddresses=127.0.0.1:" +
server.getAddress().getPort();
+ context = new DefaultCamelContext();
+ context.start();
+ template = context.createProducerTemplate();
+ }
+
+ @AfterEach
+ void stop() {
+ context.stop();
+ server.stop(0);
+ }
+
+ @Test
+ void aBuiltIndexRequestIsSentAsItIs() throws Exception {
+ IndexRequest<Object> request = IndexRequest.of(b ->
b.index("idx").id("1").document(Map.of("name", "camel")));
+
+ assertEquals("1", template.requestBody(uri + "&indexName=other",
request));
+
+ Recorded sent = single();
+ assertEquals("PUT", sent.method());
+ assertEquals("/idx/_doc/1", sent.path());
+ assertEquals(json("{\"name\":\"camel\"}"), json(sent.body()));
+ }
+
+ @Test
+ void aBuiltUpdateRequestIsSentAsItIs() throws Exception {
+ UpdateRequest<Object, Object> request
+ = UpdateRequest.of(b ->
b.index("idx").id("1").doc(Map.of("name", "camel")));
+
+ assertEquals("1", template.requestBody(uri + "&indexName=other",
request));
+
+ Recorded sent = single();
+ assertEquals("/idx/_update/1", sent.path());
+ assertEquals(json("{\"doc\":{\"name\":\"camel\"}}"),
json(sent.body()));
+ }
+
+ @Test
+ void aBuiltGetRequestIsSentAsItIs() {
+ GetResponse<?> response
+ = template.requestBody(uri, GetRequest.of(b ->
b.index("idx").id("1")), GetResponse.class);
+
+ assertTrue(response.found());
+ assertEquals("GET", single().method());
+ assertEquals("/idx/_doc/1", single().path());
+ }
+
+ @Test
+ void aBuiltDeleteRequestIsSentAsItIs() {
+ Object result = template.requestBody(uri, DeleteRequest.of(b ->
b.index("idx").id("1")));
+
+ assertEquals(Result.Deleted, result);
+ assertEquals("DELETE", single().method());
+ assertEquals("/idx/_doc/1", single().path());
+ }
+
+ @Test
+ void aBuiltSearchRequestIsSentAsItIs() throws Exception {
+ SearchRequest request = SearchRequest.of(b ->
b.index("idx").size(3).query(q -> q.matchAll(m -> m)));
+
+ assertInstanceOf(HitsMetadata.class, template.requestBody(uri +
"&indexName=other", request));
+
+ Recorded sent = single();
+ assertEquals("/idx/_search", sent.path());
+ assertEquals(json("{\"query\":{\"match_all\":{}},\"size\":3}"),
json(sent.body()));
+ }
+
+ @Test
+ void aBuiltMultiSearchRequestIsSentAsItIs() {
+ MsearchRequest request = MsearchRequest.of(b -> b.index("idx")
+ .searches(s -> s.header(h -> h).body(q -> q.query(m ->
m.matchAll(a -> a)))));
+
+ List<?> responses = template.requestBody(uri, request, List.class);
+
+ assertEquals(1, responses.size());
+ assertEquals("/idx/_msearch", single().path());
+ assertTrue(single().body().contains("match_all"), single().body());
+ }
+
+ @Test
+ void aBuiltMultiGetRequestIsSentAsItIs() throws Exception {
+ List<?> docs = template.requestBody(uri, MgetRequest.of(b ->
b.index("idx").ids("1")), List.class);
+
+ assertEquals(1, docs.size());
+ assertEquals("/idx/_mget", single().path());
+ assertEquals(json("{\"ids\":[\"1\"]}"), json(single().body()));
+ }
+
+ @Test
+ void aBuiltBulkRequestIsSentAsItIs() throws Exception {
+ BulkRequest request = BulkRequest.of(b -> b.operations(
+ o -> o.index(i ->
i.index("idx").id("1").document(Map.of("name", "camel")))));
+
+ List<?> items = template.requestBody(uri, request, List.class);
+
+ assertEquals(1, items.size());
+ assertEquals("/_bulk", single().path());
+ String[] lines = single().body().split("\n");
+ assertEquals(json("{\"index\":{\"_id\":\"1\",\"_index\":\"idx\"}}"),
json(lines[0]));
+ assertEquals(json("{\"name\":\"camel\"}"), json(lines[1]));
+ }
+
+ @Test
+ void aBuiltDeleteIndexRequestIsSentAsItIs() {
+ Object acknowledged = template.requestBody(uri,
DeleteIndexRequest.of(b -> b.index("idx")));
+
+ assertEquals(Boolean.TRUE, acknowledged);
+ assertEquals("DELETE", single().method());
+ assertEquals("/idx", single().path());
+ }
+
+ @Test
+ void deleteIndexDeletesTheIndexNamedByTheBodyWithoutHeaderOrOption() {
+ Object acknowledged = template.requestBody(uri +
"&operation=DeleteIndex", "idx");
+
+ assertEquals(Boolean.TRUE, acknowledged);
+ assertEquals("/idx", single().path());
+ }
+
+ @Test
+ void deleteIndexPrefersTheIndexNameOption() {
+ Object acknowledged = template.requestBody(uri +
"&operation=DeleteIndex&indexName=idx", "other");
+
+ assertEquals(Boolean.TRUE, acknowledged);
+ assertEquals("/idx", single().path());
+ }
+
+ @Test
+ void indexFailsClearlyWithoutBody() {
+ Exchange exchange = template.request(uri +
"&operation=Index&indexName=idx", e -> e.getMessage().setBody(null));
+
+ IllegalArgumentException e =
assertInstanceOf(IllegalArgumentException.class, exchange.getException());
+ assertEquals("The body of the Index operation cannot be null: it is
the document to index, or its request",
+ e.getMessage());
+ assertTrue(requests.isEmpty(), "a request was sent");
+ }
+
+ @Test
+ void updateFailsClearlyWithoutBody() {
+ Exchange exchange = template.request(uri +
"&operation=Update&indexName=idx", e -> {
+ e.getMessage().setHeader(OpensearchConstants.PARAM_INDEX_ID, "1");
+ e.getMessage().setBody(null);
+ });
+
+ IllegalArgumentException e =
assertInstanceOf(IllegalArgumentException.class, exchange.getException());
+ assertEquals("The body of the Update operation cannot be null: it is
the update of the document, or its request",
+ e.getMessage());
+ assertTrue(requests.isEmpty(), "a request was sent");
+ }
+
+ @Test
+ void indexReadsAFileAsTheJsonOfTheDocument() throws Exception {
+ File file = write("doc.json", "{\"name\":\"camel\"}").toFile();
+
+ assertEquals("1", template.requestBody(uri +
"&operation=Index&indexName=idx", file));
+
+ assertEquals(json("{\"name\":\"camel\"}"), json(single().body()));
+ }
+
+ @Test
+ void indexReadsAPathAsTheJsonOfTheDocument() throws Exception {
+ Path path = write("doc.json", "{\"name\":\"camel\"}");
+
+ assertEquals("1", template.requestBody(uri +
"&operation=Index&indexName=idx", path));
+
+ assertEquals(json("{\"name\":\"camel\"}"), json(single().body()));
+ }
+
+ @Test
+ void indexReadsTheFileOfTheFileComponentAsTheJsonOfTheDocument() throws
Exception {
+ File file = write("doc.json", "{\"name\":\"camel\"}").toFile();
+ Object genericFile = FileConsumer.asGenericFile(dir.toString(), file,
null, false);
+
+ assertEquals("1", template.requestBody(uri +
"&operation=Index&indexName=idx", genericFile));
+
+ assertEquals(json("{\"name\":\"camel\"}"), json(single().body()));
+ }
+
+ @Test
+ void updateReadsAFileAsJson() throws Exception {
+ File file = write("update.json",
"{\"doc\":{\"name\":\"camel\"}}").toFile();
+
+ assertEquals("1", template.requestBodyAndHeader(uri +
"&operation=Update&indexName=idx", file,
+ OpensearchConstants.PARAM_INDEX_ID, "1"));
+
+ assertEquals("/idx/_update/1", single().path());
+ assertEquals(json("{\"doc\":{\"name\":\"camel\"}}"),
json(single().body()));
+ }
+
+ @Test
+ void getByIdStillTakesTheIdAsText() {
+ GetResponse<?> response = template.requestBody(uri +
"&operation=GetById&indexName=idx", "1", GetResponse.class);
+
+ assertTrue(response.found());
+ assertEquals("/idx/_doc/1", single().path());
+ }
+
+ private Path write(String name, String content) throws IOException {
+ return Files.writeString(dir.resolve(name), content);
+ }
+
+ private Recorded single() {
+ assertEquals(1, requests.size(), "requests sent: " + requests);
+ return requests.get(0);
+ }
+
+ private static JsonNode json(String text) throws IOException {
+ return MAPPER.readTree(text);
+ }
+
+ private void answer(HttpExchange exchange) throws IOException {
+ String body = new String(exchange.getRequestBody().readAllBytes(),
StandardCharsets.UTF_8);
+ String method = exchange.getRequestMethod();
+ String path = exchange.getRequestURI().getPath();
+ requests.add(new Recorded(method, path, body));
+ byte[] response = response(method,
path).getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type", "application/json");
+ exchange.sendResponseHeaders(200, response.length);
+ try (OutputStream out = exchange.getResponseBody()) {
+ out.write(response);
+ }
+ }
+
+ private static String response(String method, String path) {
+ if (path.endsWith("/_search")) {
+ return HITS;
+ } else if (path.endsWith("/_msearch")) {
+ return "{\"took\":1,\"responses\":[" + HITS.replaceFirst("\\}$",
",\"status\":200}") + "]}";
+ } else if (path.endsWith("/_mget")) {
+ return "{\"docs\":[" + FOUND + "]}";
+ } else if (path.endsWith("/_bulk")) {
+ return "{\"took\":1,\"errors\":false,\"items\":[{\"index\":"
+ + String.format(WRITTEN, "created").replaceFirst("\\}$",
",\"status\":201}") + "}]}";
+ } else if (path.contains("/_update/")) {
+ return String.format(WRITTEN, "updated");
+ } else if (path.contains("/_doc")) {
+ return switch (method) {
+ case "GET" -> FOUND;
+ case "DELETE" -> String.format(WRITTEN, "deleted");
+ default -> String.format(WRITTEN, "created");
+ };
+ }
+ // the deletion of an index
+ return "{\"acknowledged\":true}";
+ }
+
+ private record Recorded(String method, String path, String body) {
+ }
+}
diff --git
a/components/camel-opensearch/src/test/java/org/apache/camel/component/opensearch/converter/OpensearchActionRequestConverterTest.java
b/components/camel-opensearch/src/test/java/org/apache/camel/component/opensearch/converter/OpensearchActionRequestConverterTest.java
index 87dede478c7d..223cc6085d8d 100644
---
a/components/camel-opensearch/src/test/java/org/apache/camel/component/opensearch/converter/OpensearchActionRequestConverterTest.java
+++
b/components/camel-opensearch/src/test/java/org/apache/camel/component/opensearch/converter/OpensearchActionRequestConverterTest.java
@@ -16,8 +16,11 @@
*/
package org.apache.camel.component.opensearch.converter;
+import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
+import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.component.opensearch.OpensearchConstants;
@@ -26,10 +29,17 @@ import org.apache.camel.support.DefaultExchange;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.opensearch.client.opensearch._types.query_dsl.Query;
import org.opensearch.client.opensearch.core.IndexRequest;
+import org.opensearch.client.opensearch.core.SearchRequest;
import org.opensearch.client.opensearch.core.UpdateRequest;
+import org.opensearch.client.opensearch.indices.DeleteIndexRequest;
+import org.opensearch.client.util.MissingRequiredPropertyException;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class OpensearchActionRequestConverterTest {
@@ -83,4 +93,127 @@ public class OpensearchActionRequestConverterTest {
// no CamelOpensearchIndexId header -> the caller's id must be
preserved, not overwritten with null
assertEquals("original", result.build().id());
}
+
+ @Test
+ void aJsonSearchBodyKeepsItsSizeSortAggregationsAndSource() throws
Exception {
+ Exchange exchange = new DefaultExchange(context);
+ String body = """
+ {"query": {"match": {"name": "camel"}}, "size": 5, "from": 2,
+ "sort": [{"name": "asc"}], "aggs": {"names": {"terms":
{"field": "name"}}},
+ "_source": ["name"], "track_total_hits": true}
+ """;
+
+ SearchRequest request =
OpensearchActionRequestConverter.toSearchRequestBuilder(body, exchange).build();
+
+ assertEquals(Query.Kind.Match, request.query()._kind());
+ assertEquals(5, request.size());
+ assertEquals(2, request.from());
+ assertEquals(1, request.sort().size());
+ assertEquals("name", request.sort().get(0).field().field());
+ assertTrue(request.aggregations().containsKey("names"));
+ assertNotNull(request.source());
+ assertNotNull(request.trackTotalHits());
+ }
+
+ @Test
+ void aJsonSearchBodyWithoutQueryIsASearchRequest() throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+
+ SearchRequest request =
OpensearchActionRequestConverter.toSearchRequestBuilder(
+ "{\"size\": 0, \"aggs\": {\"names\": {\"terms\": {\"field\":
\"name\"}}}}", exchange).build();
+
+ assertEquals(0, request.size());
+ assertTrue(request.aggregations().containsKey("names"));
+ }
+
+ @Test
+ void aJsonQueryAloneIsStillTheQuery() throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(OpensearchConstants.PARAM_INDEX_NAME,
"idx");
+
+ SearchRequest request = OpensearchActionRequestConverter
+ .toSearchRequestBuilder("{\"match\": {\"name\": \"camel\"}}",
exchange).build();
+
+ assertEquals(Query.Kind.Match, request.query()._kind());
+ assertEquals("name", request.query().match().field());
+ assertEquals(List.of("idx"), request.index());
+ }
+
+ @Test
+ void aMapSearchBodyKeepsItsSizeAndSort() throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+ Map<String, Object> body = new LinkedHashMap<>();
+ body.put("query", Map.of("match", Map.of("name", "camel")));
+ body.put("size", 5);
+ body.put("sort", List.of(Map.of("name", "asc")));
+
+ SearchRequest request =
OpensearchActionRequestConverter.toSearchRequestBuilder(body, exchange).build();
+
+ assertEquals(Query.Kind.Match, request.query()._kind());
+ assertEquals(5, request.size());
+ assertEquals(1, request.sort().size());
+ }
+
+ @Test
+ void aMapQueryAloneIsStillTheQuery() throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+
+ SearchRequest request = OpensearchActionRequestConverter
+ .toSearchRequestBuilder(Map.of("match", Map.of("name",
"camel")), exchange).build();
+
+ assertEquals(Query.Kind.Match, request.query()._kind());
+ }
+
+ @Test
+ void aMapSearchBodyWhoseQueryIsNotAMap() throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+ Map<String, Object> body = Map.of("query", new
ObjectMapper().readTree("{\"match_all\": {}}"));
+
+ SearchRequest request =
OpensearchActionRequestConverter.toSearchRequestBuilder(body, exchange).build();
+
+ assertEquals(Query.Kind.MatchAll, request.query()._kind());
+ }
+
+ @Test
+ void theSizeAndFromHeadersOverrideThoseOfTheSearchBody() throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(OpensearchConstants.PARAM_SIZE, 10);
+ exchange.getIn().setHeader(OpensearchConstants.PARAM_FROM, 20);
+
+ SearchRequest request =
OpensearchActionRequestConverter.toSearchRequestBuilder(
+ "{\"query\": {\"match_all\": {}}, \"size\": 5, \"from\": 2}",
exchange).build();
+
+ assertEquals(10, request.size());
+ assertEquals(20, request.from());
+ }
+
+ @Test
+ void deleteIndexTakesTheIndexNameOfTheBodyWithoutHeader() {
+ Exchange exchange = new DefaultExchange(context);
+
+ DeleteIndexRequest request
+ =
OpensearchActionRequestConverter.toDeleteIndexRequestBuilder("twitter",
exchange).build();
+
+ assertEquals(List.of("twitter"), request.index());
+ }
+
+ @Test
+ void deleteIndexPrefersTheIndexNameHeader() {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(OpensearchConstants.PARAM_INDEX_NAME,
"logs");
+
+ DeleteIndexRequest request
+ =
OpensearchActionRequestConverter.toDeleteIndexRequestBuilder("twitter",
exchange).build();
+
+ assertEquals(List.of("logs"), request.index());
+ }
+
+ @Test
+ void deleteIndexWithoutAnyIndexNameFailsBeforeSending() {
+ Exchange exchange = new DefaultExchange(context);
+
+ DeleteIndexRequest.Builder builder =
OpensearchActionRequestConverter.toDeleteIndexRequestBuilder(" ", exchange);
+
+ assertThrows(MissingRequiredPropertyException.class, builder::build);
+ }
}
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 95a84ce3a204..488977773fba 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
@@ -4829,6 +4829,25 @@ success and stored nothing.
The catalog now says that `deleteByQuery` defaults to `false`, as it always
did at runtime. A `null` value in a `Map`
body or in a `CamelSolrField.*` header is left out of the document, instead of
failing with a `NullPointerException`.
+=== camel-opensearch - built requests, search bodies, DeleteIndex and file
bodies
+
+A request of the OpenSearch client in the message body (an `IndexRequest`,
`UpdateRequest`, `GetRequest`,
+`DeleteRequest`, `SearchRequest`, `MsearchRequest`, `MgetRequest`,
`BulkRequest` or `DeleteIndexRequest`) is now sent
+as it is, like its builder. Before, a built `IndexRequest` was written as the
document itself, with a generated id, a
+built `UpdateRequest` failed without the id header, and the other built
requests failed with "Wrong body type".
+
+The `Search` operation now uses the whole search request of a JSON text or
`Map` body: its `size`, `from`, `sort`,
+aggregations, `_source` and the other members are no longer dropped, so the
hits returned can change. A JSON object
+with a single member other than `query`, such as `{"match": {...}}`, is still
taken as the query alone. The `size` and
+`from` headers or options, when set, still override those of the body.
+
+The `DeleteIndex` operation now deletes the index named by a `String` body
when neither the
+`CamelOpensearchIndexName` header nor the `indexName` option is set. Before,
the call failed.
+
+The `Index` and `Update` operations now fail with an
`IllegalArgumentException` when the body is null, instead of
+failing with a Jackson error. A `java.io.File`, a `java.nio.file.Path` or the
file of the file
+component is now read as the JSON of the document, instead of being written as
a JSON string of its path.
+
=== camel-ssh - the shell channel and sleepForShellPrompt
An `ssh` endpoint with `channelType=shell` and no `shellPrompt` now fails when
it starts. Previously each command
