Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Early registration of ReadListener

Browse Source 
This is similar to the WriteListener changes on the ServerHttpResponse
where we are more naturally exposed to a delayed write. Nevertheless
we could also have a delayed read and should be consistent. The early
initialization of the RequestBodyPublisher also simplifies the
internal implementation a bit.
This commit is contained in:
Rossen Stoyanchev
2016-12-14 16:12:08 -05:00
parent 3370f41c61
commit 4738a61e98
2 changed files with 34 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
spring-web/src/main/java/org/springframework/http/server/reactive/ServletServerHttpRequest.java
View File

@@ -54,19 +54,13 @@ public class ServletServerHttpRequest extends AbstractServerHttpRequest {
private final HttpServletRequest request;
private final DataBufferFactory bufferFactory;
private final int bufferSize;
private final Object bodyPublisherLock = new Object();
private volatile RequestBodyPublisher bodyPublisher;
private final RequestBodyPublisher bodyPublisher;
private final Object cookieLock = new Object();
public ServletServerHttpRequest(HttpServletRequest request, AsyncContext asyncContext,
DataBufferFactory bufferFactory, int bufferSize) {
DataBufferFactory bufferFactory, int bufferSize) throws IOException {
super(initUri(request), initHeaders(request));
@@ -74,12 +68,16 @@ public class ServletServerHttpRequest extends AbstractServerHttpRequest {
Assert.isTrue(bufferSize > 0, "'bufferSize' must be higher than 0");
this.request = request;
this.bufferFactory = bufferFactory;
this.bufferSize = bufferSize;
asyncContext.addListener(new RequestAsyncListener());
// Tomcat expects ReadListener registration on initial thread
ServletInputStream inputStream = request.getInputStream();
this.bodyPublisher = new RequestBodyPublisher(inputStream, bufferFactory, bufferSize);
this.bodyPublisher.registerReadListener();
}
private static URI initUri(HttpServletRequest request) {
Assert.notNull(request, "'request' must not be null");
try {
@@ -168,24 +166,7 @@ public class ServletServerHttpRequest extends AbstractServerHttpRequest {
@Override
public Flux<DataBuffer> getBody() {
try {
RequestBodyPublisher publisher = this.bodyPublisher;
if (publisher == null) {
synchronized (this.bodyPublisherLock) {
publisher = this.bodyPublisher;
if (publisher == null) {
ServletInputStream inputStream = this.request.getInputStream();
publisher = new RequestBodyPublisher(inputStream, this.bufferFactory, this.bufferSize);
publisher.registerReadListener();
this.bodyPublisher = publisher;
}
}
}
return Flux.from(publisher);
}
catch (IOException ex) {
return Flux.error(ex);
}
return Flux.from(this.bodyPublisher);
}
@@ -198,33 +179,22 @@ public class ServletServerHttpRequest extends AbstractServerHttpRequest {
public void onTimeout(AsyncEvent event) {
Throwable ex = event.getThrowable();
ex = ex != null ? ex : new IllegalStateException("Async operation timeout.");
handleError(ex);
bodyPublisher.onError(ex);
}
@Override
public void onError(AsyncEvent event) {
handleError(event.getThrowable());
}
private void handleError(Throwable ex) {
if (bodyPublisher != null) {
bodyPublisher.onError(ex);
}
bodyPublisher.onError(event.getThrowable());
}
@Override
public void onComplete(AsyncEvent event) {
if (bodyPublisher != null) {
bodyPublisher.onAllDataRead();
}
bodyPublisher.onAllDataRead();
}
}
private static class RequestBodyPublisher extends AbstractListenerReadPublisher<DataBuffer> {
private final RequestBodyPublisherReadListener readListener =
new RequestBodyPublisherReadListener();
private final ServletInputStream inputStream;
private final DataBufferFactory bufferFactory;
@@ -241,7 +211,7 @@ public class ServletServerHttpRequest extends AbstractServerHttpRequest {
}
public void registerReadListener() throws IOException {
this.inputStream.setReadListener(this.readListener);
this.inputStream.setReadListener(new RequestBodyPublisherReadListener());
}
@Override
@@ -268,7 +238,6 @@ public class ServletServerHttpRequest extends AbstractServerHttpRequest {
return null;
}
private class RequestBodyPublisherReadListener implements ReadListener {
@Override