Support more content types for Triple protocol

dubbo-dependencies-bom/pom.xml

@@ -207,6 +207,7 @@
     <palantirJavaFormat.version>2.38.0</palantirJavaFormat.version>
     <revision>3.3.0-beta.2-SNAPSHOT</revision>
     <open_feign_version>3.1.5</open_feign_version>
+    <commons_fileupload.version>1.4</commons_fileupload.version>
   </properties>
 
   <dependencyManagement>
@@ -636,6 +637,11 @@
         <artifactId>tomcat-embed-core</artifactId>
         <version>${tomcat_embed_version}</version>
       </dependency>
+      <dependency>
+        <groupId>commons-fileupload</groupId>
+        <artifactId>commons-fileupload</artifactId>
+        <version>${commons_fileupload.version}</version>
+      </dependency>
       <dependency>
         <groupId>io.etcd</groupId>
         <artifactId>jetcd-core</artifactId>

dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/utils/UrlUtils.java

@@ -134,6 +134,8 @@ public static Collection<String> allSerializations(URL url) {
                 .filter(StringUtils::isNotBlank)
                 .ifPresent(serializations::add);
         serializations.add(DefaultSerializationSelector.getDefaultRemotingSerialization());
+        System.out.println("Trying to find serializations from URL:" + url);
+        System.out.println("Found serialization(s):" + serializations);
         return Collections.unmodifiableSet(serializations);
     }
 

dubbo-remoting/dubbo-remoting-http12/pom.xml

@@ -60,5 +60,20 @@
       <artifactId>netty-codec-http2</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>commons-fileupload</groupId>
+      <artifactId>commons-fileupload</artifactId>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>javax.servlet</groupId>
+      <artifactId>javax.servlet-api</artifactId>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.jboss.resteasy</groupId>
+      <artifactId>resteasy-jaxb-provider</artifactId>
+      <scope>provided</scope>
+    </dependency>
   </dependencies>
 </project>

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/AbstractServerHttpChannelObserver.java

@@ -16,9 +16,15 @@
  */
 package org.apache.dubbo.remoting.http12;
 
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.remoting.http12.exception.DecodeException;
 import org.apache.dubbo.remoting.http12.exception.EncodeException;
 import org.apache.dubbo.remoting.http12.exception.HttpStatusException;
 import org.apache.dubbo.remoting.http12.message.HttpMessageCodec;
+import org.apache.dubbo.remoting.http12.message.HttpMessageCodecFactory;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+
+import java.util.List;
 
 public abstract class AbstractServerHttpChannelObserver implements CustomizableHttpChannelObserver<Object> {
 
@@ -32,18 +38,18 @@ public abstract class AbstractServerHttpChannelObserver implements CustomizableH
 
     private boolean headerSent;
 
-    private HttpMessageCodec httpMessageCodec;
+    private HttpMessageCodec responseEncoder;
 
     public AbstractServerHttpChannelObserver(HttpChannel httpChannel) {
         this.httpChannel = httpChannel;
     }
 
-    public void setHttpMessageCodec(HttpMessageCodec httpMessageCodec) {
-        this.httpMessageCodec = httpMessageCodec;
+    public void setResponseEncoder(HttpMessageCodec responseEncoder) {
+        this.responseEncoder = responseEncoder;
     }
 
-    protected HttpMessageCodec getHttpMessageCodec() {
-        return httpMessageCodec;
+    public HttpMessageCodec getResponseEncoder() {
+        return responseEncoder;
     }
 
     @Override
@@ -77,7 +83,8 @@ public void onNext(Object data) {
             }
             HttpOutputMessage outputMessage = encodeHttpOutputMessage(data);
             preOutputMessage(outputMessage);
-            this.httpMessageCodec.encode(outputMessage.getBody(), data);
+            this.responseEncoder.encode(outputMessage.getBody(), data);
+            // TODO: write时写入的序列化信息是客户端支持的吗？
             getHttpChannel().writeMessage(outputMessage);
             postOutputMessage(outputMessage);
         } catch (Throwable e) {
@@ -114,7 +121,7 @@ public void onError(Throwable throwable) {
             errorResponse.setMessage(throwable.getMessage());
             this.errorResponseCustomizer.accept(errorResponse, throwable);
             HttpOutputMessage httpOutputMessage = encodeHttpOutputMessage(errorResponse);
-            this.httpMessageCodec.encode(httpOutputMessage.getBody(), errorResponse);
+            this.responseEncoder.encode(httpOutputMessage.getBody(), errorResponse);
             getHttpChannel().writeMessage(httpOutputMessage);
         } catch (Throwable ex) {
             throwable = new EncodeException(ex);
@@ -140,7 +147,7 @@ private void doSendHeaders(String statusCode) {
                 .headers()
                 .set(
                         HttpHeaderNames.CONTENT_TYPE.getName(),
-                        httpMessageCodec.contentType().getName());
+                        responseEncoder.responseContentType().getName());
         this.headersCustomizer.accept(httpMetadata.headers());
         getHttpChannel().writeHeader(httpMetadata);
         this.headerSent = true;
@@ -154,4 +161,20 @@ protected void doOnCompleted(Throwable throwable) {
         this.trailersCustomizer.accept(httpMetadata.headers(), throwable);
         getHttpChannel().writeHeader(httpMetadata);
     }
+
+    public void findAndSetEncoder(URL url, HttpHeaders headers, FrameworkModel frameworkModel) {
+        List<HttpMessageCodecFactory> factories =
+                frameworkModel.getExtensionLoader(HttpMessageCodecFactory.class).getActivateExtensions();
+        this.responseEncoder = findCodecFactoryForEncode(headers, factories).createCodec(url, frameworkModel, headers);
+    }
+
+    public HttpMessageCodecFactory findCodecFactoryForEncode(
+            HttpHeaders headers, List<HttpMessageCodecFactory> candidates) {
+        for (HttpMessageCodecFactory factory : candidates) {
+            if (factory.supportEncode(headers)) {
+                return factory;
+            }
+        }
+        throw new DecodeException("Can not recognize accept header:" + headers.toString());
+    }
 }

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HttpHeaderNames.java

@@ -25,7 +25,9 @@ public enum HttpHeaderNames {
 
     TRANSFER_ENCODING("transfer-encoding"),
 
-    TE("te");
+    TE("te"),
+
+    ACCEPT("accept");
 
     private final String name;
 

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HttpHeaders.java

@@ -212,6 +212,14 @@ public Map<String, String> toSingleValueMap() {
         return result;
     }
 
+    public String getContentType() {
+        return getFirst(HttpHeaderNames.CONTENT_TYPE.getName());
+    }
+
+    public String getAcceptEncoding() {
+        return getFirst(HttpHeaderNames.ACCEPT.getName());
+    }
+
     @Override
     public boolean equals(Object other) {
         return (this == other || this.targetMap.equals(other));

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/HttpMessageCodec.java

@@ -21,6 +21,8 @@
 import org.apache.dubbo.remoting.http12.exception.DecodeException;
 import org.apache.dubbo.remoting.http12.exception.EncodeException;
 
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 
@@ -48,8 +50,22 @@ default Object[] decode(InputStream inputStream, Class<?>[] targetTypes) throws
 
     MediaType contentType();
 
+    default MediaType responseContentType() {
+        return contentType();
+    }
+
     default boolean support(String contentType) {
         MediaType mediaType = this.contentType();
         return mediaType.getName().startsWith(contentType);
     }
+
+    default ByteArrayOutputStream toByteArrayStream(InputStream in) throws IOException {
+        ByteArrayOutputStream result = new ByteArrayOutputStream();
+        byte[] buffer = new byte[1024];
+        int length;
+        while ((length = in.read(buffer)) != -1) {
+            result.write(buffer, 0, length);
+        }
+        return result;
+    }
 }

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/HttpMessageCodecFactory.java

@@ -19,6 +19,8 @@
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.extension.ExtensionScope;
 import org.apache.dubbo.common.extension.SPI;
+import org.apache.dubbo.remoting.http12.HttpHeaderNames;
+import org.apache.dubbo.remoting.http12.HttpHeaders;
 import org.apache.dubbo.rpc.model.FrameworkModel;
 
 /**
@@ -27,12 +29,19 @@
 @SPI(scope = ExtensionScope.FRAMEWORK)
 public interface HttpMessageCodecFactory {
 
-    HttpMessageCodec createCodec(URL url, FrameworkModel frameworkModel);
+    HttpMessageCodec createCodec(URL url, FrameworkModel frameworkModel, HttpHeaders headers);
 
     MediaType contentType();
 
-    default boolean support(String contentType) {
+    default boolean supportDecode(HttpHeaders headers) {
+        String contentType = headers.getFirst(HttpHeaderNames.CONTENT_TYPE.getName());
         MediaType mediaType = this.contentType();
-        return mediaType.getName().startsWith(contentType);
+        return contentType.startsWith(mediaType.getName());
+    }
+
+    default boolean supportEncode(HttpHeaders headers) {
+        String acceptEncoding = headers.getFirst(HttpHeaderNames.ACCEPT.getName());
+        return acceptEncoding != null
+                && acceptEncoding.contains(this.contentType().getName());
     }
 }

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/MediaType.java

@@ -28,6 +28,15 @@ public class MediaType {
 
     public static final MediaType TEXT_EVENT_STREAM_VALUE = new MediaType("text", "event-stream");
 
+    public static final MediaType MULTIPART_FORM_DATA = new MediaType("multipart", "form-data");
+
+    public static final MediaType APPLICATION_X_WWW_FROM_URLENCODED =
+            new MediaType("application", "x-www-form-urlencoded");
+
+    public static final MediaType APPLICATION_XML = new MediaType("application", "xml");
+
+    public static final MediaType TEXT_PLAIN = new MediaType("text", "plain");
+
     private final String name;
 
     private final String type;

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/JsonCodec.java → dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/codec/JsonCodec.java

@@ -14,12 +14,14 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.dubbo.remoting.http12.message;
+package org.apache.dubbo.remoting.http12.message.codec;
 
 import org.apache.dubbo.common.extension.Activate;
 import org.apache.dubbo.common.utils.JsonUtils;
 import org.apache.dubbo.remoting.http12.exception.DecodeException;
 import org.apache.dubbo.remoting.http12.exception.EncodeException;
+import org.apache.dubbo.remoting.http12.message.HttpMessageCodec;
+import org.apache.dubbo.remoting.http12.message.MediaType;
 
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -35,8 +37,6 @@
 @Activate
 public class JsonCodec implements HttpMessageCodec {
 
-    public static final HttpMessageCodec INSTANCE = new JsonCodec();
-
     @Override
     public MediaType contentType() {
         return MediaType.APPLICATION_JSON_VALUE;

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/JsonCodecFactory.java → dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/codec/JsonCodecFactory.java

@@ -14,10 +14,14 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.dubbo.remoting.http12.message;
+package org.apache.dubbo.remoting.http12.message.codec;
 
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.extension.Activate;
+import org.apache.dubbo.remoting.http12.HttpHeaders;
+import org.apache.dubbo.remoting.http12.message.HttpMessageCodec;
+import org.apache.dubbo.remoting.http12.message.HttpMessageCodecFactory;
+import org.apache.dubbo.remoting.http12.message.MediaType;
 import org.apache.dubbo.rpc.model.FrameworkModel;
 
 @Activate
@@ -26,7 +30,7 @@ public class JsonCodecFactory implements HttpMessageCodecFactory {
     public static final String NAME = "json";
 
     @Override
-    public HttpMessageCodec createCodec(URL url, FrameworkModel frameworkModel) {
+    public HttpMessageCodec createCodec(URL url, FrameworkModel frameworkModel, HttpHeaders headers) {
         return new JsonCodec();
     }
 

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/JsonPbCodec.java → dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/codec/JsonPbCodec.java

@@ -14,13 +14,15 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.dubbo.remoting.http12.message;
+package org.apache.dubbo.remoting.http12.message.codec;
 
 import org.apache.dubbo.common.extension.Activate;
 import org.apache.dubbo.common.utils.ClassUtils;
 import org.apache.dubbo.common.utils.MethodUtils;
 import org.apache.dubbo.remoting.http12.exception.DecodeException;
 import org.apache.dubbo.remoting.http12.exception.EncodeException;
+import org.apache.dubbo.remoting.http12.message.HttpMessageCodec;
+import org.apache.dubbo.remoting.http12.message.MediaType;
 
 import java.io.IOException;
 import java.io.InputStream;

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/JsonPbCodecFactory.java → dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/codec/JsonPbCodecFactory.java

@@ -14,22 +14,26 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.dubbo.remoting.http12.message;
+package org.apache.dubbo.remoting.http12.message.codec;
 
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.extension.Activate;
 import org.apache.dubbo.common.utils.ClassUtils;
+import org.apache.dubbo.remoting.http12.HttpHeaders;
+import org.apache.dubbo.remoting.http12.message.HttpMessageCodec;
+import org.apache.dubbo.remoting.http12.message.HttpMessageCodecFactory;
+import org.apache.dubbo.remoting.http12.message.MediaType;
 import org.apache.dubbo.rpc.model.FrameworkModel;
 
 @Activate(order = -100)
 public class JsonPbCodecFactory implements HttpMessageCodecFactory {
 
     @Override
-    public HttpMessageCodec createCodec(URL url, FrameworkModel frameworkModel) {
+    public HttpMessageCodec createCodec(URL url, FrameworkModel frameworkModel, HttpHeaders headers) {
         HttpMessageCodec codec = frameworkModel
                 .getExtensionLoader(HttpMessageCodecFactory.class)
                 .getExtension(JsonCodecFactory.NAME)
-                .createCodec(url, frameworkModel);
+                .createCodec(url, frameworkModel, headers);
         JsonPbCodec jsonPbCodec = new JsonPbCodec();
         jsonPbCodec.setJsonCodec(codec);
         return jsonPbCodec;
@@ -41,8 +45,8 @@ public MediaType contentType() {
     }
 
     @Override
-    public boolean support(String contentType) {
-        return HttpMessageCodecFactory.super.support(contentType)
+    public boolean supportDecode(HttpHeaders headers) {
+        return HttpMessageCodecFactory.super.supportDecode(headers)
                 && ClassUtils.isPresent(
                         "com.google.protobuf.Message", getClass().getClassLoader());
     }