Skip to content

Commit

Permalink
Cleanup after writing responses. Potential fix for #2665
Browse files Browse the repository at this point in the history
  • Loading branch information
@graemerocher
graemerocher committed Feb 12, 2020
1 parent b9370d5 commit c072bc5
Showing 1 changed file with 34 additions and 49 deletions.
83 changes: 34 additions & 49 deletions http-server-netty/src/main/java/io/micronaut/http/server/netty/RoutingInBoundHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,33 +15,6 @@
*/
package io.micronaut.http.server.netty;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;
import java.nio.file.Paths;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import javax.annotation.Nullable;

import io.micronaut.buffer.netty.NettyByteBufferFactory;
import io.micronaut.context.BeanContext;
import io.micronaut.context.exceptions.BeanInstantiationException;
Expand All @@ -56,15 +29,11 @@
import io.micronaut.core.reflect.ClassUtils;
import io.micronaut.core.type.Argument;
import io.micronaut.core.type.ReturnType;
import io.micronaut.http.HttpAttributes;
import io.micronaut.http.HttpHeaders;
import io.micronaut.http.HttpMethod;
import io.micronaut.http.HttpRequest;
import io.micronaut.http.HttpResponse;
import io.micronaut.http.HttpStatus;
import io.micronaut.http.MediaType;
import io.micronaut.http.MutableHttpHeaders;
import io.micronaut.http.MutableHttpResponse;
import io.micronaut.http.*;
import io.micronaut.http.annotation.Produces;
import io.micronaut.http.annotation.Status;
import io.micronaut.http.bind.binders.ContinuationArgumentBinder;
Expand Down Expand Up @@ -101,12 +70,7 @@
import io.micronaut.runtime.http.codec.TextPlainCodec;
import io.micronaut.scheduling.executor.ExecutorSelector;
import io.micronaut.scheduling.executor.ThreadSelection;
import io.micronaut.web.router.BasicObjectRouteMatch;
import io.micronaut.web.router.MethodBasedRouteMatch;
import io.micronaut.web.router.RouteMatch;
import io.micronaut.web.router.Router;
import io.micronaut.web.router.UriRoute;
import io.micronaut.web.router.UriRouteMatch;
import io.micronaut.web.router.*;
import io.micronaut.web.router.exceptions.DuplicateRouteException;
import io.micronaut.web.router.exceptions.UnsatisfiedRouteException;
import io.micronaut.web.router.resource.StaticResourceResolver;
Expand All @@ -121,16 +85,8 @@
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.DecoderResult;
import io.netty.handler.codec.TooLongFrameException;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.DefaultHttpContent;
import io.netty.handler.codec.http.DefaultLastHttpContent;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpContent;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpHeaderValues;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.LastHttpContent;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.multipart.Attribute;
import io.netty.handler.codec.http.multipart.FileUpload;
import io.netty.handler.codec.http.multipart.HttpData;
Expand All @@ -148,6 +104,28 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;
import java.nio.file.Paths;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import static io.micronaut.core.util.KotlinUtils.isKotlinCoroutineSuspended;
import static io.micronaut.inject.util.KotlinExecutableMethodUtils.isKotlinFunctionReturnTypeUnit;

Expand Down Expand Up @@ -1321,7 +1299,9 @@ protected void onComplete(MutableHttpResponse<?> message) {

@Override
protected void doOnError(Throwable t) {
exceptionCaughtInternal(context, t, (NettyHttpRequest) requestReference.get(), false);
final NettyHttpRequest nettyHttpRequest = (NettyHttpRequest) requestReference.get();
nettyHttpRequest.release();
exceptionCaughtInternal(context, t, nettyHttpRequest, false);
}
});
}
Expand Down Expand Up @@ -1354,7 +1334,8 @@ private void writeFinalNettyResponse(MutableHttpResponse<?> message, AtomicRefer
nettyHeaders.add(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
}
// close handled by HttpServerKeepAliveHandler
context.writeAndFlush(nettyResponse);
context.writeAndFlush(nettyResponse)
.addListener(future -> cleanupIfNecessary(context));
context.read();
}
}
Expand Down Expand Up @@ -1687,6 +1668,10 @@ public HttpContent apply(Object message) {
context.read();
});

httpContentPublisher = Flowable.fromPublisher(httpContentPublisher)
.doOnComplete(() -> cleanupIfNecessary(context))
.doOnError(throwable -> cleanupIfNecessary(context));

DelegateStreamedHttpResponse streamedResponse = new DelegateStreamedHttpResponse(nativeResponse, httpContentPublisher);
io.netty.handler.codec.http.HttpHeaders headers = streamedResponse.headers();
headers.add(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
Expand Down

0 comments on commit c072bc5

Please sign in to comment.