Fix memory leak

Release resource when exception occur

Release resource for http1

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

@@ -162,8 +162,13 @@ protected final HttpOutputMessage buildMessage(Object data) throws Throwable {
  data = ((HttpResult<?>) data).getBody();
  }
  HttpOutputMessage outputMessage = encodeHttpOutputMessage(data);
- preOutputMessage(outputMessage);
- responseEncoder.encode(outputMessage.getBody(), data);
+ try {
+ preOutputMessage(outputMessage);
+ responseEncoder.encode(outputMessage.getBody(), data);
+ } catch (Throwable t) {
+ outputMessage.close();
+ throw t;
+ }
  return outputMessage;
  }
 

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

@@ -18,7 +18,7 @@
 
 import java.io.InputStream;
 
-public interface HttpInputMessage {
+public interface HttpInputMessage extends AutoCloseable {
 
  InputStream getBody();
 }

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

@@ -19,7 +19,7 @@
 import java.io.ByteArrayOutputStream;
 import java.io.OutputStream;
 
-public interface HttpOutputMessage {
+public interface HttpOutputMessage extends AutoCloseable {
 
  HttpOutputMessage EMPTY_MESSAGE = new HttpOutputMessage() {
 
@@ -29,6 +29,11 @@ public interface HttpOutputMessage {
  public OutputStream getBody() {
  return INPUT_STREAM;
  }
+
+ @Override
+ public void close() throws Exception {
+ INPUT_STREAM.close();
+ }
  };
 
  OutputStream getBody();

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/DefaultHttp1Request.java

@@ -52,4 +52,9 @@ public String method() {
  public String path() {
  return httpMetadata.path();
  }
+
+ @Override
+ public void close() throws Exception {
+ httpInputMessage.close();
+ }
 }

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/DefaultHttp1Response.java

@@ -42,4 +42,9 @@ public InputStream getBody() {
  public HttpHeaders headers() {
  return httpMetadata.headers();
  }
+
+ @Override
+ public void close() throws Exception {
+ httpInputMessage.close();
+ }
 }

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/Http1InputMessage.java

@@ -32,4 +32,9 @@ public Http1InputMessage(InputStream body) {
  public InputStream getBody() {
  return body;
  }
+
+ @Override
+ public void close() throws Exception {
+ body.close();
+ }
 }

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/Http1OutputMessage.java

@@ -20,6 +20,8 @@
 
 import java.io.OutputStream;
 
+import io.netty.buffer.ByteBufOutputStream;
+
 public class Http1OutputMessage implements HttpOutputMessage {
 
  private final OutputStream outputStream;
@@ -32,4 +34,12 @@ public Http1OutputMessage(OutputStream outputStream) {
  public OutputStream getBody() {
  return outputStream;
  }
+
+ @Override
+ public void close() throws Exception {
+ if (outputStream instanceof ByteBufOutputStream) {
+ ((ByteBufOutputStream) outputStream).buffer().release();
+ }
+ outputStream.close();
+ }
 }

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2InputMessageFrame.java

@@ -62,4 +62,9 @@ public int id() {
  public boolean isEndStream() {
  return endStream;
  }
+
+ @Override
+ public void close() throws Exception {
+ body.close();
+ }
 }

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2OutputMessageFrame.java

@@ -18,6 +18,8 @@
 
 import java.io.OutputStream;
 
+import io.netty.buffer.ByteBufOutputStream;
+
 public class Http2OutputMessageFrame implements Http2OutputMessage {
 
  private final OutputStream body;
@@ -42,6 +44,14 @@ public OutputStream getBody() {
  return body;
  }
 
+ @Override
+ public void close() throws Exception {
+ if (body instanceof ByteBufOutputStream) {
+ ((ByteBufOutputStream) body).buffer().release();
+ }
+ body.close();
+ }
+
  @Override
  public boolean isEndStream() {
  return endStream;

dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http1/DefaultHttp11ServerTransportListener.java

@@ -24,6 +24,7 @@
 import org.apache.dubbo.remoting.http12.HttpHeaderNames;
 import org.apache.dubbo.remoting.http12.HttpInputMessage;
 import org.apache.dubbo.remoting.http12.RequestMetadata;
+import org.apache.dubbo.remoting.http12.exception.DecodeException;
 import org.apache.dubbo.remoting.http12.h1.Http1ServerChannelObserver;
 import org.apache.dubbo.remoting.http12.h1.Http1ServerStreamChannelObserver;
 import org.apache.dubbo.remoting.http12.h1.Http1ServerTransportListener;
@@ -82,6 +83,15 @@ protected HttpMessageListener buildHttpMessageListener() {
  return new DefaultHttpMessageListener(listeningDecoder);
  }
 
+ @Override
+ protected void doOnData(HttpInputMessage message) {
+ try (HttpInputMessage msg = message) {
+ super.doOnData(msg);
+ } catch (Exception e) {
+ throw new DecodeException(e);
+ }
+ }
+
  private ServerCallListener startListener(
  RpcInvocation invocation, MethodDescriptor methodDescriptor, Invoker<?> invoker) {
  switch (methodDescriptor.getRpcType()) {