Merge branch '1019-codec-memory-leak' into 3.x

Signed-off-by: Ben Hale <bhale@pivotal.io>

Author: nebhale

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/doppler/MultipartCodec.java

@@ -16,20 +16,26 @@
 
 package org.cloudfoundry.reactor.doppler;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import io.netty.buffer.Unpooled;
 import io.netty.handler.codec.DelimiterBasedFrameDecoder;
 import io.netty.handler.codec.http.HttpHeaderNames;
 import reactor.core.publisher.Flux;
 import reactor.netty.ByteBufFlux;
 import reactor.netty.http.client.HttpClientResponse;
 
+import java.io.IOException;
 import java.io.InputStream;
 import java.nio.charset.Charset;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
 final class MultipartCodec {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(MultipartCodec.class);
+
     private static final Pattern BOUNDARY_PATTERN = Pattern.compile("multipart/.+; boundary=(.*)");
 
     private static final int MAX_PAYLOAD_SIZE = 1024 * 1024;
@@ -49,7 +55,8 @@ static DelimiterBasedFrameDecoder createDecoder(HttpClientResponse response) {
 
     static Flux<InputStream> decode(ByteBufFlux body) {
         return body.asInputStream()
-            .skip(1);
+            .skip(1)
+            .doOnDiscard(InputStream.class, MultipartCodec::close);
     }
 
     private static String extractMultipartBoundary(HttpClientResponse response) {
@@ -63,4 +70,12 @@ private static String extractMultipartBoundary(HttpClientResponse response) {
         }
     }
 
+    private static void close(InputStream in) {
+        try {
+            in.close();
+        } catch (IOException e) {
+            LOGGER.warn("Could not close input stream. This will cause a direct memory leak.", e);
+        }
+    }
+
 }

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/util/Operator.java

@@ -164,7 +164,7 @@ public <T> Flux<T> parseBodyToFlux(Function<HttpClientResponseWithBody, Publishe
                 ByteBufFlux body = connection.inbound().receive();
 
                 return processResponse(response, body).flatMapMany(responseTransformer)
-                    .doOnTerminate(connection::dispose);
+                    .doFinally(signalType -> connection.dispose());
             });
         }
 
@@ -257,7 +257,7 @@ private Publisher<InputStream> handleWebsocketCommunication(WebsocketInbound inb
             return inbound.aggregateFrames()
                 .receive()
                 .asInputStream()
-                .doOnTerminate(outbound::sendClose);
+                .doFinally(signalType -> outbound.sendClose());
         }
 
     }