Allow unparsed binary data to be used for ingestion

java-client/src/main/java/co/elastic/clients/elasticsearch/core/CreateRequest.java

@@ -512,5 +512,5 @@ public static <TDocument> JsonpDeserializer<CreateRequest<TDocument>> createCrea
 				}
 				return params;
 
-			}, SimpleEndpoint.emptyMap(), true, CreateResponse._DESERIALIZER);
+			}, SimpleEndpoint.emptyMap(), r -> r.document(), CreateResponse._DESERIALIZER);
 }

java-client/src/main/java/co/elastic/clients/elasticsearch/core/IndexRequest.java

@@ -663,5 +663,5 @@ public static <TDocument> JsonpDeserializer<IndexRequest<TDocument>> createIndex
 				}
 				return params;
 
-			}, SimpleEndpoint.emptyMap(), true, IndexResponse._DESERIALIZER);
+			}, SimpleEndpoint.emptyMap(), r -> r.document(), IndexResponse._DESERIALIZER);
 }

java-client/src/main/java/co/elastic/clients/json/JsonpUtils.java

@@ -150,6 +150,60 @@ public static void skipValue(JsonParser parser, Event event) {
         }
     }
 
+    /**
+     * Copy the JSON value at the current parser location to a JSON generator.
+     */
+    public static void copy(JsonParser parser, JsonGenerator generator, JsonParser.Event event) {
+
+        switch (event) {
+            case START_OBJECT:
+                generator.writeStartObject();
+                while ((event = parser.next()) != Event.END_OBJECT) {
+                    expectEvent(parser, Event.KEY_NAME, event);
+                    generator.writeKey(parser.getString());
+                    copy(parser, generator, parser.next());
+                }
+                generator.writeEnd();
+                break;
+
+            case START_ARRAY:
+                generator.writeStartArray();
+                generator.writeStartObject();
+                while ((event = parser.next()) != Event.END_ARRAY) {
+                    copy(parser, generator, event);
+                }
+                generator.writeEnd();
+                break;
+
+            case VALUE_STRING:
+                generator.write(parser.getString());
+                break;
+
+            case VALUE_FALSE:
+                generator.write(false);
+                break;
+
+            case VALUE_TRUE:
+                generator.write(true);
+                break;
+
+            case VALUE_NULL:
+                generator.writeNull();
+                break;
+
+            case VALUE_NUMBER:
+                if (parser.isIntegralNumber()) {
+                    generator.write(parser.getLong());
+                } else {
+                    generator.write(parser.getBigDecimal());
+                }
+                break;
+
+            default:
+                throw new UnexpectedJsonEventException(parser, event);
+        }
+    }
+
     public static <T> void serialize(T value, JsonGenerator generator, @Nullable JsonpSerializer<T> serializer, JsonpMapper mapper) {
         if (serializer != null) {
             serializer.serialize(value, generator, mapper);

java-client/src/main/java/co/elastic/clients/transport/Endpoint.java

@@ -70,7 +70,23 @@ default Map<String, String> headers(RequestT request) {
     return Collections.emptyMap();
   }
 
-  boolean hasRequestBody();
+  /**
+   * Get the body for a request. The caller must handle several cases depending on the interface implemented by the result:
+   * <li>
+   *     {@code null} means the request has no body.
+   * </li>
+   * <li>
+   *     {@link co.elastic.clients.json.NdJsonpSerializable} must be serialized as nd-json.
+   * </li>
+   * <li>
+   *     {@link co.elastic.clients.util.BinaryData} must be serialized as is.
+   * </li>
+   * <li>
+   *     All other objects must be serialized as JSON using a {@link co.elastic.clients.json.JsonpMapper}
+   * </li>
+   */
+  @Nullable
+  Object body(RequestT request);
 
   /**
    * Is this status code to be considered as an error?
@@ -90,7 +106,7 @@ default BinaryEndpoint<RequestT> withBinaryResponse() {
           this::requestUrl,
           this::queryParameters,
           this::headers,
-          this.hasRequestBody(),
+          this::body,
           null
       );
   }

java-client/src/main/java/co/elastic/clients/transport/endpoints/BinaryEndpoint.java

@@ -24,6 +24,19 @@
 
 public class BinaryEndpoint<RequestT> extends EndpointBase<RequestT, BinaryResponse> {
 
+    public BinaryEndpoint(
+        String id,
+        Function<RequestT, String> method,
+        Function<RequestT, String> requestUrl,
+        Function<RequestT,
+            Map<String, String>> queryParameters,
+        Function<RequestT, Map<String, String>> headers,
+        Function<RequestT, Object> body,
+        Object ignored // same number of arguments as SimpleEndpoint
+    ) {
+        super(id, method, requestUrl, queryParameters, headers, body);
+    }
+
     public BinaryEndpoint(
         String id,
         Function<RequestT, String> method,
@@ -34,7 +47,7 @@ public BinaryEndpoint(
         boolean hasRequestBody,
         Object ignored // same number of arguments as SimpleEndpoint
     ) {
-        super(id, method, requestUrl, queryParameters, headers, hasRequestBody);
+        super(id, method, requestUrl, queryParameters, headers, hasRequestBody ? returnSelf() : returnNull());
     }
 
     @Override

java-client/src/main/java/co/elastic/clients/transport/endpoints/BooleanEndpoint.java

@@ -34,7 +34,7 @@ public BooleanEndpoint(
         boolean hasRequestBody,
         Object ignored // same number of arguments as SimpleEndpoint
     ) {
-        super(id, method, requestUrl, queryParameters, headers, hasRequestBody);
+        super(id, method, requestUrl, queryParameters, headers, hasRequestBody ? returnSelf() : returnNull());
     }
 
     @Override

java-client/src/main/java/co/elastic/clients/transport/endpoints/DelegatingJsonEndpoint.java

@@ -58,9 +58,10 @@ public Map<String, String> headers(Req request) {
         return endpoint.headers(request);
     }
 
+    @Nullable
     @Override
-    public boolean hasRequestBody() {
-        return endpoint.hasRequestBody();
+    public Object body(Req request) {
+        return endpoint.body(request);
     }
 
     @Override

java-client/src/main/java/co/elastic/clients/transport/endpoints/EndpointBase.java

@@ -24,13 +24,16 @@
 import co.elastic.clients.transport.Endpoint;
 import org.apache.http.client.utils.URLEncodedUtils;
 
+import javax.annotation.Nullable;
 import java.util.Collections;
 import java.util.Map;
 import java.util.function.Function;
 
 public class EndpointBase<RequestT, ResponseT> implements Endpoint<RequestT, ResponseT, ErrorResponse> {
 
     private static final Function<?, Map<String, String>> EMPTY_MAP = x -> Collections.emptyMap();
+    private static final Function<?, Object> RETURN_NULL = x -> null;
+    private static final Function<?, ?> RETURN_SELF = x -> x;
 
     /**
      * Returns a function that always returns an empty String to String map. Useful to avoid creating lots of
@@ -41,27 +44,44 @@ public static <T> Function<T, Map<String, String>> emptyMap() {
         return (Function<T, Map<String, String>>) EMPTY_MAP;
     }
 
+    /**
+     * Returns a function that always returns {@code null}.
+     */
+    @SuppressWarnings("unchecked")
+    static <T, U> Function<T, U> returnNull() {
+        return (Function<T, U>) RETURN_NULL;
+    }
+
+    /**
+     * Returns a function that always returns its parameter. It's similar to {@code Function.identity()} with the difference
+     * that the input and output generic parameters are different, making it suitable for use in a wider range of use cases.
+     */
+    @SuppressWarnings("unchecked")
+    static <T, U> Function<T, U> returnSelf() {
+        return (Function<T, U>) RETURN_SELF;
+    }
+
     protected final String id;
     protected final Function<RequestT, String> method;
     protected final Function<RequestT, String> requestUrl;
     protected final Function<RequestT, Map<String, String>> queryParameters;
     protected final Function<RequestT, Map<String, String>> headers;
-    protected final boolean hasRequestBody;
+    protected final Function<RequestT, Object> body;
 
     public EndpointBase(
         String id,
         Function<RequestT, String> method,
         Function<RequestT, String> requestUrl,
         Function<RequestT, Map<String, String>> queryParameters,
         Function<RequestT, Map<String, String>> headers,
-        boolean hasRequestBody
+        Function<RequestT, Object> body
     ) {
         this.id = id;
         this.method = method;
         this.requestUrl = requestUrl;
         this.queryParameters = queryParameters;
         this.headers = headers;
-        this.hasRequestBody = hasRequestBody;
+        this.body = body;
     }
 
     @Override
@@ -89,9 +109,10 @@ public Map<String, String> headers(RequestT request) {
         return this.headers.apply(request);
     }
 
+    @Nullable
     @Override
-    public boolean hasRequestBody() {
-        return this.hasRequestBody;
+    public Object body(RequestT request) {
+        return this.body.apply(request);
     }
 
     // ES-specific
@@ -114,7 +135,7 @@ public <NewResponseT> SimpleEndpoint<RequestT, NewResponseT> withResponseDeseria
             requestUrl,
             queryParameters,
             headers,
-            hasRequestBody,
+            body,
             newResponseParser
         );
     }

java-client/src/main/java/co/elastic/clients/transport/endpoints/SimpleEndpoint.java

@@ -24,24 +24,12 @@
 import co.elastic.clients.transport.JsonEndpoint;
 import org.apache.http.client.utils.URLEncodedUtils;
 
-import java.util.Collections;
 import java.util.Map;
 import java.util.function.Function;
 
 public class SimpleEndpoint<RequestT, ResponseT> extends EndpointBase<RequestT, ResponseT>
     implements JsonEndpoint<RequestT, ResponseT, ErrorResponse> {
 
-    private static final Function<?, Map<String, String>> EMPTY_MAP = x -> Collections.emptyMap();
-
-    /**
-     * Returns a function that always returns an empty String to String map. Useful to avoid creating lots of
-     * duplicate lambdas in endpoints that don't have headers or parameters.
-     */
-    @SuppressWarnings("unchecked")
-    public static <T> Function<T, Map<String, String>> emptyMap() {
-        return (Function<T, Map<String, String>>) EMPTY_MAP;
-    }
-
     private final JsonpDeserializer<ResponseT> responseParser;
 
     public SimpleEndpoint(
@@ -50,13 +38,33 @@ public SimpleEndpoint(
         Function<RequestT, String> requestUrl,
         Function<RequestT, Map<String, String>> queryParameters,
         Function<RequestT, Map<String, String>> headers,
-        boolean hasRequestBody,
+        Function<RequestT, Object> body,
         JsonpDeserializer<ResponseT> responseParser
     ) {
-        super(id, method, requestUrl, queryParameters, headers, hasRequestBody);
+        super(id, method, requestUrl, queryParameters, headers, body);
         this.responseParser = responseParser;
     }
 
+    public SimpleEndpoint(
+        String id,
+        Function<RequestT, String> method,
+        Function<RequestT, String> requestUrl,
+        Function<RequestT, Map<String, String>> queryParameters,
+        Function<RequestT, Map<String, String>> headers,
+        boolean hasResponseBody,
+        JsonpDeserializer<ResponseT> responseParser
+    ) {
+        this(
+            id,
+            method,
+            requestUrl,
+            queryParameters,
+            headers,
+            hasResponseBody ? returnSelf() : returnNull(),
+            responseParser
+        );
+    }
+
     @Override
     public JsonpDeserializer<ResponseT> responseDeserializer() {
         return this.responseParser;
@@ -76,7 +84,7 @@ public <NewResponseT> SimpleEndpoint<RequestT, NewResponseT> withResponseDeseria
             requestUrl,
             queryParameters,
             headers,
-            hasRequestBody,
+            body,
             newResponseParser
         );
     }

java-client/src/main/java/co/elastic/clients/transport/rest_client/RestClientTransport.java

@@ -59,6 +59,7 @@
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
@@ -215,21 +216,41 @@ private <RequestT> org.elasticsearch.client.Request prepareLowLevelRequest(
 
         clientReq.addParameters(params);
 
-        if (endpoint.hasRequestBody()) {
-            // Request has a body and must implement JsonpSerializable or NdJsonpSerializable
-
-            if (request instanceof NdJsonpSerializable) {
+        Object body = endpoint.body(request);
+        if (body != null) {
+            // Request has a body
+            if (body instanceof NdJsonpSerializable) {
                 List<ByteBuffer> lines = new ArrayList<>();
-                collectNdJsonLines(lines, (NdJsonpSerializable)request);
+                collectNdJsonLines(lines, (NdJsonpSerializable) request);
                 clientReq.setEntity(new MultiBufferEntity(lines, JsonContentType));
+
+            } else if (body instanceof BinaryData) {
+                BinaryData data = (BinaryData)body;
+
+                // ES expects the Accept and Content-Type headers to be consistent.
+                ContentType contentType;
+                String dataContentType = data.contentType();
+                if (co.elastic.clients.util.ContentType.APPLICATION_JSON.equals(dataContentType)) {
+                    // Fast path
+                    contentType = JsonContentType;
+                } else {
+                    contentType = ContentType.parse(dataContentType);
+                }
+
+                clientReq.setEntity(new MultiBufferEntity(
+                    Collections.singletonList(data.asByteBuffer()),
+                    contentType
+                ));
+
             } else {
                 ByteArrayOutputStream baos = new ByteArrayOutputStream();
                 JsonGenerator generator = mapper.jsonProvider().createGenerator(baos);
-                mapper.serialize(request, generator);
+                mapper.serialize(body, generator);
                 generator.close();
                 clientReq.setEntity(new ByteArrayEntity(baos.toByteArray(), JsonContentType));
             }
         }
+
         // Request parameter intercepted by LLRC
         clientReq.addParameter("ignore", "400,401,403,404,405");
         return clientReq;