Tidy up Netty4 HTTP response interfaces

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4ChunkedHttpResponse.java

@@ -12,13 +12,14 @@
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.netty.handler.codec.http.HttpVersion;
 
+import org.elasticsearch.http.HttpResponse;
 import org.elasticsearch.rest.ChunkedRestResponseBody;
 import org.elasticsearch.rest.RestStatus;
 
 /**
  * A http response that will be transferred via chunked encoding when handled by {@link Netty4HttpPipeliningHandler}.
  */
-public final class Netty4ChunkedHttpResponse extends DefaultHttpResponse implements Netty4RestResponse {
+public final class Netty4ChunkedHttpResponse extends DefaultHttpResponse implements Netty4HttpResponse, HttpResponse {
 
     private final int sequence;
 
@@ -38,4 +39,14 @@ public ChunkedRestResponseBody body() {
     public int getSequence() {
         return sequence;
     }
+
+    @Override
+    public void addHeader(String name, String value) {
+        headers().add(name, value);
+    }
+
+    @Override
+    public boolean containsHeader(String name) {
+        return headers().contains(name);
+    }
 }

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4FullHttpResponse.java

@@ -13,10 +13,11 @@
 import io.netty.handler.codec.http.HttpVersion;
 
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.http.HttpResponse;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.transport.netty4.Netty4Utils;
 
-public final class Netty4FullHttpResponse extends DefaultFullHttpResponse implements Netty4RestResponse {
+public final class Netty4FullHttpResponse extends DefaultFullHttpResponse implements Netty4HttpResponse, HttpResponse {
 
     private final int sequence;
 
@@ -29,4 +30,14 @@ public final class Netty4FullHttpResponse extends DefaultFullHttpResponse implem
     public int getSequence() {
         return sequence;
     }
+
+    @Override
+    public void addHeader(String name, String value) {
+        headers().add(name, value);
+    }
+
+    @Override
+    public boolean containsHeader(String name) {
+        return headers().contains(name);
+    }
 }

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpPipeliningHandler.java

@@ -55,7 +55,7 @@ public class Netty4HttpPipeliningHandler extends ChannelDuplexHandler {
     private final Logger logger;
 
     private final int maxEventsHeld;
-    private final PriorityQueue<Tuple<? extends Netty4RestResponse, ChannelPromise>> outboundHoldingQueue;
+    private final PriorityQueue<Tuple<? extends Netty4HttpResponse, ChannelPromise>> outboundHoldingQueue;
 
     private record ChunkedWrite(PromiseCombiner combiner, ChannelPromise onDone, ChunkedRestResponseBody responseBody) {}
 
@@ -136,10 +136,10 @@ protected void handlePipelinedRequest(ChannelHandlerContext ctx, Netty4HttpReque
 
     @Override
     public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) throws IOException {
-        assert msg instanceof Netty4RestResponse : "Invalid message type: " + msg.getClass();
+        assert msg instanceof Netty4HttpResponse : "Invalid message type: " + msg.getClass();
         boolean success = false;
         try {
-            final Netty4RestResponse restResponse = (Netty4RestResponse) msg;
+            final Netty4HttpResponse restResponse = (Netty4HttpResponse) msg;
             if (restResponse.getSequence() != writeSequence) {
                 assert restResponse.getSequence() > writeSequence
                     : "response sequence [" + restResponse.getSequence() + "] we below write sequence [" + writeSequence + "]";
@@ -174,7 +174,7 @@ public void write(final ChannelHandlerContext ctx, final Object msg, final Chann
 
     private void doWriteQueued(ChannelHandlerContext ctx) throws IOException {
         while (outboundHoldingQueue.isEmpty() == false && outboundHoldingQueue.peek().v1().getSequence() == writeSequence) {
-            final Tuple<? extends Netty4RestResponse, ChannelPromise> top = outboundHoldingQueue.poll();
+            final Tuple<? extends Netty4HttpResponse, ChannelPromise> top = outboundHoldingQueue.poll();
             assert top != null : "we know the outbound holding queue to not be empty at this point";
             doWrite(ctx, top.v1(), top.v2());
         }
@@ -191,7 +191,7 @@ private void doWriteQueued(ChannelHandlerContext ctx) throws IOException {
         SPLIT_THRESHOLD = (int) (NettyAllocator.suggestedMaxAllocationSize() * 0.99);
     }
 
-    private void doWrite(ChannelHandlerContext ctx, Netty4RestResponse readyResponse, ChannelPromise promise) throws IOException {
+    private void doWrite(ChannelHandlerContext ctx, Netty4HttpResponse readyResponse, ChannelPromise promise) throws IOException {
         assert currentChunkedWrite == null : "unexpected existing write [" + currentChunkedWrite + "]";
         assert readyResponse != null : "cannot write null response";
         assert readyResponse.getSequence() == writeSequence;
@@ -347,11 +347,11 @@ public void close(ChannelHandlerContext ctx, ChannelPromise promise) {
             safeFailPromise(currentChunkedWrite.onDone, new ClosedChannelException());
             currentChunkedWrite = null;
         }
-        List<Tuple<? extends Netty4RestResponse, ChannelPromise>> inflightResponses = removeAllInflightResponses();
+        List<Tuple<? extends Netty4HttpResponse, ChannelPromise>> inflightResponses = removeAllInflightResponses();
 
         if (inflightResponses.isEmpty() == false) {
             ClosedChannelException closedChannelException = new ClosedChannelException();
-            for (Tuple<? extends Netty4RestResponse, ChannelPromise> inflightResponse : inflightResponses) {
+            for (Tuple<? extends Netty4HttpResponse, ChannelPromise> inflightResponse : inflightResponses) {
                 safeFailPromise(inflightResponse.v2(), closedChannelException);
             }
         }
@@ -396,8 +396,8 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
         }
     }
 
-    private List<Tuple<? extends Netty4RestResponse, ChannelPromise>> removeAllInflightResponses() {
-        ArrayList<Tuple<? extends Netty4RestResponse, ChannelPromise>> responses = new ArrayList<>(outboundHoldingQueue);
+    private List<Tuple<? extends Netty4HttpResponse, ChannelPromise>> removeAllInflightResponses() {
+        ArrayList<Tuple<? extends Netty4HttpResponse, ChannelPromise>> responses = new ArrayList<>(outboundHoldingQueue);
         outboundHoldingQueue.clear();
         return responses;
     }

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpResponse.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.http.netty4;
+
+/**
+ * Super-interface for responses handled by the Netty4 HTTP transport.
+ */
+public sealed interface Netty4HttpResponse permits Netty4FullHttpResponse, Netty4ChunkedHttpResponse {
+    /**
+     * @return The sequence number for the request which corresponds with this response, for making sure that we send responses to pipelined
+     * requests in the correct order.
+     */
+    int getSequence();
+}