Adding support for more than one protocol for transport

Signed-off-by: Vacha Shah <[email protected]>

server/src/main/java/org/opensearch/transport/BaseInboundMessage.java

@@ -0,0 +1,37 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.transport;
+
+import org.opensearch.common.annotation.ExperimentalApi;
+
+/**
+ * Base class for inbound data as a message.
+ * Different implementations are used for different protocols.
+ *
+ * @opensearch.internal
+ */
+@ExperimentalApi
+public interface BaseInboundMessage {
+
+    /**
+     * The protocol used to encode this message
+     */
+    static String NATIVE_PROTOCOL = "native";
+    static String PROTOBUF_PROTOCOL = "protobuf";
+
+    /**
+     * @return the protocol used to encode this message
+     */
+    public String getProtocol();
+
+    /**
+     * Set the protocol used to encode this message
+     */
+    public void setProtocol();
+}

server/src/main/java/org/opensearch/transport/InboundMessage.java

@@ -47,14 +47,15 @@
  * @opensearch.api
  */
 @PublicApi(since = "1.0.0")
-public class InboundMessage implements Releasable {
+public class InboundMessage implements Releasable, BaseInboundMessage {
 
     private final Header header;
     private final ReleasableBytesReference content;
     private final Exception exception;
     private final boolean isPing;
     private Releasable breakerRelease;
     private StreamInput streamInput;
+    private String protocol;
 
     public InboundMessage(Header header, ReleasableBytesReference content, Releasable breakerRelease) {
         this.header = header;
@@ -133,4 +134,14 @@ public void close() {
     public String toString() {
         return "InboundMessage{" + header + "}";
     }
+
+    @Override
+    public String getProtocol() {
+        return NATIVE_PROTOCOL;
+    }
+
+    @Override
+    public void setProtocol() {
+        this.protocol = NATIVE_PROTOCOL;
+    }
 }

server/src/main/java/org/opensearch/transport/InboundPipeline.java

@@ -62,10 +62,11 @@ public class InboundPipeline implements Releasable {
     private final StatsTracker statsTracker;
     private final InboundDecoder decoder;
     private final InboundAggregator aggregator;
-    private final BiConsumer<TcpChannel, InboundMessage> messageHandler;
+    private final BiConsumer<TcpChannel, BaseInboundMessage> messageHandler;
     private Exception uncaughtException;
     private final ArrayDeque<ReleasableBytesReference> pending = new ArrayDeque<>(2);
     private boolean isClosed = false;
+    private Version version;
 
     public InboundPipeline(
         Version version,
@@ -74,14 +75,15 @@ public InboundPipeline(
         LongSupplier relativeTimeInMillis,
         Supplier<CircuitBreaker> circuitBreaker,
         Function<String, RequestHandlerRegistry<TransportRequest>> registryFunction,
-        BiConsumer<TcpChannel, InboundMessage> messageHandler
+        BiConsumer<TcpChannel, BaseInboundMessage> messageHandler
     ) {
         this(
             statsTracker,
             relativeTimeInMillis,
             new InboundDecoder(version, recycler),
             new InboundAggregator(circuitBreaker, registryFunction),
-            messageHandler
+            messageHandler,
+            version
         );
     }
 
@@ -90,13 +92,15 @@ public InboundPipeline(
         LongSupplier relativeTimeInMillis,
         InboundDecoder decoder,
         InboundAggregator aggregator,
-        BiConsumer<TcpChannel, InboundMessage> messageHandler
+        BiConsumer<TcpChannel, BaseInboundMessage> messageHandler,
+        Version version
     ) {
         this.relativeTimeInMillis = relativeTimeInMillis;
         this.statsTracker = statsTracker;
         this.decoder = decoder;
         this.aggregator = aggregator;
         this.messageHandler = messageHandler;
+        this.version = version;
     }
 
     @Override
@@ -124,37 +128,42 @@ public void doHandleBytes(TcpChannel channel, ReleasableBytesReference reference
         statsTracker.markBytesRead(reference.length());
         pending.add(reference.retain());
 
-        final ArrayList<Object> fragments = fragmentList.get();
-        boolean continueHandling = true;
-
-        while (continueHandling && isClosed == false) {
-            boolean continueDecoding = true;
-            while (continueDecoding && pending.isEmpty() == false) {
-                try (ReleasableBytesReference toDecode = getPendingBytes()) {
-                    final int bytesDecoded = decoder.decode(toDecode, fragments::add);
-                    if (bytesDecoded != 0) {
-                        releasePendingBytes(bytesDecoded);
-                        if (fragments.isEmpty() == false && endOfMessage(fragments.get(fragments.size() - 1))) {
+        String incomingMessageProtocol = TcpTransport.determineTransportProtocol(reference);
+        if (incomingMessageProtocol.equals(BaseInboundMessage.PROTOBUF_PROTOCOL) && this.version.onOrAfter(Version.V_3_0_0)) {
+            // protobuf messages logic can be added here
+        } else {
+            final ArrayList<Object> fragments = fragmentList.get();
+            boolean continueHandling = true;
+
+            while (continueHandling && isClosed == false) {
+                boolean continueDecoding = true;
+                while (continueDecoding && pending.isEmpty() == false) {
+                    try (ReleasableBytesReference toDecode = getPendingBytes()) {
+                        final int bytesDecoded = decoder.decode(toDecode, fragments::add);
+                        if (bytesDecoded != 0) {
+                            releasePendingBytes(bytesDecoded);
+                            if (fragments.isEmpty() == false && endOfMessage(fragments.get(fragments.size() - 1))) {
+                                continueDecoding = false;
+                            }
+                        } else {
                             continueDecoding = false;
                         }
-                    } else {
-                        continueDecoding = false;
                     }
                 }
-            }
 
-            if (fragments.isEmpty()) {
-                continueHandling = false;
-            } else {
-                try {
-                    forwardFragments(channel, fragments);
-                } finally {
-                    for (Object fragment : fragments) {
-                        if (fragment instanceof ReleasableBytesReference) {
-                            ((ReleasableBytesReference) fragment).close();
+                if (fragments.isEmpty()) {
+                    continueHandling = false;
+                } else {
+                    try {
+                        forwardFragments(channel, fragments);
+                    } finally {
+                        for (Object fragment : fragments) {
+                            if (fragment instanceof ReleasableBytesReference) {
+                                ((ReleasableBytesReference) fragment).close();
+                            }
                         }
+                        fragments.clear();
                     }
-                    fragments.clear();
                 }
             }
         }