Visitar URL original
Make sure to always close input streams bound to responses (#1970) · apache/maven-resolver@dbaf309 · GitHub
Skip to content

Commit dbaf309

Browse files
kwinkwin
authored
Make sure to always close input streams bound to responses (#1970)
Ensure all connections are closed when transporter is closed. Add PUT IT where response contains body. This closes #1964
1 parent 45482a8 commit dbaf309

7 files changed

Lines changed: 96 additions & 25 deletions

File tree

‎maven-resolver-test-http/pom.xml‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -131,6 +131,10 @@
131131
<groupId>com.google.code.gson</groupId>
132132
<artifactId>gson</artifactId>
133133
</dependency>
134+
<dependency>
135+
<groupId>org.awaitility</groupId>
136+
<artifactId>awaitility</artifactId>
137+
</dependency>
134138
</dependencies>
135139

136140
</project>

‎maven-resolver-test-http/src/main/java/org/eclipse/aether/internal/test/util/http/HttpServer.java‎

Lines changed: 39 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import java.nio.file.StandardOpenOption;
2828
import java.util.ArrayList;
2929
import java.util.Base64;
30+
import java.util.Collection;
3031
import java.util.Collections;
3132
import java.util.List;
3233
import java.util.Map;
@@ -60,6 +61,8 @@
6061
import org.eclipse.jetty.http3.server.HTTP3ServerQuicConfiguration;
6162
import org.eclipse.jetty.io.ByteBufferPool;
6263
import org.eclipse.jetty.io.Content;
64+
import org.eclipse.jetty.io.DatagramChannelEndPoint;
65+
import org.eclipse.jetty.io.EndPoint;
6366
import org.eclipse.jetty.quic.quiche.server.QuicheServerConnector;
6467
import org.eclipse.jetty.quic.quiche.server.QuicheServerQuicConfiguration;
6568
import org.eclipse.jetty.server.Handler;
@@ -189,6 +192,8 @@ public enum ChecksumHeader {
189192

190193
private final List<LogEntry> logEntries = Collections.synchronizedList(new ArrayList<>());
191194

195+
private String responseBodyForPut;
196+
192197
public String getHost() {
193198
return "localhost";
194199
}
@@ -323,6 +328,11 @@ public HttpServer setRangeSupport(boolean rangeSupport) {
323328
return this;
324329
}
325330

331+
public HttpServer setResponseBodyForPut(String body) {
332+
this.responseBodyForPut = body;
333+
return this;
334+
}
335+
326336
public HttpServer setWebDav(boolean webDav) {
327337
this.webDav = webDav;
328338
return this;
@@ -397,6 +407,27 @@ public void stop() throws Exception {
397407
}
398408
}
399409

410+
public int getNumConnectedEndPoints() {
411+
if (server.isStopped()) {
412+
throw new IllegalStateException("Server is stopped");
413+
}
414+
Collection<EndPoint> connectedEndPoints = new ArrayList<>();
415+
if (httpConnector != null) {
416+
connectedEndPoints.addAll(httpConnector.getConnectedEndPoints());
417+
}
418+
if (httpsConnector != null) {
419+
connectedEndPoints.addAll(httpsConnector.getConnectedEndPoints());
420+
}
421+
if (http3Connector != null) {
422+
// filter out the always present DatagramChannelEndPoint, which is not a real live connection
423+
// (https://github.com/jetty/jetty.project/issues/15436)
424+
http3Connector.getConnectedEndPoints().stream()
425+
.filter(endPoint -> !(endPoint instanceof DatagramChannelEndPoint))
426+
.forEach(connectedEndPoints::add);
427+
}
428+
return connectedEndPoints.size();
429+
}
430+
400431
private class CompressionEnforcingHandler extends CompressionHandler {
401432
// duplicate of CompressionHandler.pathConfigs which is private
402433
private final PathMappings<CompressionConfig> pathConfigs = new PathMappings<>();
@@ -662,7 +693,14 @@ public boolean handle(Request req, Response response, Callback callback) throws
662693
file.delete();
663694
throw e;
664695
}
665-
response.setStatus(HttpServletResponse.SC_NO_CONTENT);
696+
// optionally add some response body to test that the client can handle it, even though Maven
697+
// Repository Protocol doesn't mention it
698+
if (responseBodyForPut != null) {
699+
writeResponseBodyMessage(req, response, responseBodyForPut);
700+
response.setStatus(HttpServletResponse.SC_CREATED);
701+
} else {
702+
response.setStatus(HttpServletResponse.SC_NO_CONTENT);
703+
}
666704
} else {
667705
response.setStatus(HttpServletResponse.SC_FORBIDDEN);
668706
}

‎maven-resolver-test-http/src/main/java/org/eclipse/aether/internal/test/util/http/HttpTransporterTest.java‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,10 +39,12 @@
3939
import java.util.Enumeration;
4040
import java.util.HashMap;
4141
import java.util.Map;
42+
import java.util.concurrent.TimeUnit;
4243
import java.util.concurrent.atomic.AtomicReference;
4344
import java.util.function.Supplier;
4445
import java.util.stream.Stream;
4546

47+
import org.awaitility.Awaitility;
4648
import org.eclipse.aether.ConfigurationProperties;
4749
import org.eclipse.aether.DefaultRepositoryCache;
4850
import org.eclipse.aether.DefaultRepositorySystemSession;
@@ -305,6 +307,8 @@ protected void tearDown() throws Exception {
305307
closer = null;
306308
}
307309
if (httpServer != null) {
310+
// check for leaked connections (e.g., due to not closing response body streams)
311+
Awaitility.await().atMost(5, TimeUnit.SECONDS).until(() -> httpServer.getNumConnectedEndPoints() == 0);
308312
httpServer.stop();
309313
httpServer = null;
310314
}
@@ -1499,6 +1503,20 @@ protected void testPut_AuthCache_Preemptive() throws Exception {
14991503
assertEquals(1, httpServer.getLogEntries().size()); // put w/ auth
15001504
}
15011505

1506+
@Test
1507+
protected void testPut_WithResponseBody() throws Exception {
1508+
httpServer.setAuthentication("testuser", "testpass");
1509+
httpServer.setResponseBodyForPut("Some dummy response body");
1510+
auth = new AuthenticationBuilder()
1511+
.addUsername("testuser")
1512+
.addPassword("testpass")
1513+
.build();
1514+
newTransporter(httpServer.getHttpUrl());
1515+
PutTask task = new PutTask(URI.create("repo/file.txt")).setDataString("upload");
1516+
transporter.put(task);
1517+
// this leads to stuck threads in some transporters if the response body is not consumed
1518+
}
1519+
15021520
@Test
15031521
@Timeout(20)
15041522
protected void testConcurrency() throws Exception {

‎maven-resolver-transport-apache/src/test/java/org/eclipse/aether/transport/apache/ApacheTransporterTest.java‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
import org.eclipse.aether.spi.connector.transport.GetTask;
3434
import org.eclipse.aether.spi.connector.transport.PutTask;
3535
import org.eclipse.aether.spi.io.PathProcessorSupport;
36+
import org.junit.jupiter.api.AfterEach;
3637
import org.junit.jupiter.api.Test;
3738

3839
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -68,6 +69,18 @@ protected boolean supportsHttp2() {
6869
return false;
6970
}
7071

72+
@AfterEach
73+
@Override
74+
protected void tearDown() throws Exception {
75+
// make sure to also release any connection in the global state (otherwise the check for connection leaks will
76+
// fail)
77+
GlobalState globalState = GlobalState.get(session);
78+
if (globalState != null) {
79+
globalState.close();
80+
}
81+
super.tearDown();
82+
}
83+
7184
@Test
7285
void testGet_WebDav() throws Exception {
7386
httpServer.setWebDav(true);

‎maven-resolver-transport-jdk-parent/maven-resolver-transport-jdk11/src/main/java/org/eclipse/aether/transport/jdk/JdkTransporter.java‎

Lines changed: 10 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -291,16 +291,11 @@ protected void implGet(GetTask task) throws Exception {
291291
resume = false;
292292
continue;
293293
}
294-
try {
295-
JdkRFC9457Reporter.INSTANCE.generateException(response, (statusCode, reasonPhrase) -> {
296-
throw new HttpTransporterException(statusCode);
297-
});
298-
} finally {
299-
closeBody(response);
300-
}
294+
JdkRFC9457Reporter.INSTANCE.generateException(response, (statusCode, reasonPhrase) -> {
295+
throw new HttpTransporterException(statusCode);
296+
});
301297
}
302298
} catch (ConnectException e) {
303-
closeBody(response);
304299
throw enhance(e);
305300
}
306301
break;
@@ -437,17 +432,14 @@ protected void implPut(PutTask task) throws Exception {
437432
task.getDataLength()));
438433
}
439434
prepare(request);
435+
HttpResponse<InputStream> response = null;
440436
try {
441-
HttpResponse<InputStream> response = send(request.build(), HttpResponse.BodyHandlers.ofInputStream());
437+
response = send(request.build(), HttpResponse.BodyHandlers.ofInputStream());
442438
task.getListener().transportPropertiesAvailable(createTransportProperties(response));
443439
if (response.statusCode() >= MULTIPLE_CHOICES) {
444-
try {
445-
JdkRFC9457Reporter.INSTANCE.generateException(response, (statusCode, reasonPhrase) -> {
446-
throw new HttpTransporterException(statusCode);
447-
});
448-
} finally {
449-
closeBody(response);
450-
}
440+
JdkRFC9457Reporter.INSTANCE.generateException(response, (statusCode, reasonPhrase) -> {
441+
throw new HttpTransporterException(statusCode);
442+
});
451443
}
452444
} catch (ConnectException e) {
453445
throw enhance(e);
@@ -458,6 +450,8 @@ protected void implPut(PutTask task) throws Exception {
458450
throw (TransferCancelledException) rootCause;
459451
}
460452
throw e;
453+
} finally {
454+
closeBody(response);
461455
}
462456
}
463457

‎maven-resolver-transport-jetty/src/main/java/org/eclipse/aether/transport/jetty/JettyTransporter.java‎

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -373,8 +373,7 @@ protected void implPut(PutTask task) throws Exception {
373373
});
374374
AtomicBoolean started = new AtomicBoolean(false);
375375
Response response;
376-
InputStreamResponseListener listener = new InputStreamResponseListener();
377-
try {
376+
try (InputStreamResponseListener listener = new InputStreamResponseListener()) {
378377
request.onRequestCommit(r -> {
379378
if (task.getDataLength() == 0) {
380379
if (started.compareAndSet(false, true)) {
@@ -404,6 +403,11 @@ protected void implPut(PutTask task) throws Exception {
404403
.send(listener);
405404
response = listener.get(requestTimeout, TimeUnit.MILLISECONDS);
406405
task.getListener().transportPropertiesAvailable(createTransportProperties(request, rawResponseHeaders));
406+
if (response.getStatus() >= MULTIPLE_CHOICES) {
407+
JettyRFC9457Reporter.INSTANCE.generateException(listener, (statusCode, reasonPhrase) -> {
408+
throw new HttpTransporterException(statusCode);
409+
});
410+
}
407411
} catch (ExecutionException e) {
408412
Throwable t = e.getCause();
409413
if (t instanceof IOException ioex) {
@@ -418,12 +422,6 @@ protected void implPut(PutTask task) throws Exception {
418422
throw new RuntimeException(t);
419423
}
420424
}
421-
422-
if (response.getStatus() >= MULTIPLE_CHOICES) {
423-
JettyRFC9457Reporter.INSTANCE.generateException(listener, (statusCode, reasonPhrase) -> {
424-
throw new HttpTransporterException(statusCode);
425-
});
426-
}
427425
}
428426

429427
@Override

‎pom.xml‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -338,6 +338,12 @@
338338
<artifactId>jmh-generator-annprocess</artifactId>
339339
<version>${jmhVersion}</version>
340340
</dependency>
341+
<dependency>
342+
<groupId>org.awaitility</groupId>
343+
<artifactId>awaitility</artifactId>
344+
<!-- release 4.3.1 was not successful (https://github.com/awaitility/awaitility/issues/306) -->
345+
<version>4.3.0</version>
346+
</dependency>
341347
</dependencies>
342348
</dependencyManagement>
343349

0 commit comments

Comments
 (0)