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

Polishing

Browse Source 
See gh-26434
This commit is contained in:
Rossen Stoyanchev
2021-02-11 21:26:37 +00:00
parent 3e502d4739
commit 1c6bab23ea
4 changed files with 34 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerServerHttpResponse.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-2021 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -38,12 +38,12 @@ public abstract class AbstractListenerServerHttpResponse extends AbstractServerH
private final AtomicBoolean writeCalled = new AtomicBoolean();
public AbstractListenerServerHttpResponse(DataBufferFactory dataBufferFactory) {
super(dataBufferFactory);
public AbstractListenerServerHttpResponse(DataBufferFactory bufferFactory) {
super(bufferFactory);
}
public AbstractListenerServerHttpResponse(DataBufferFactory dataBufferFactory, HttpHeaders headers) {
super(dataBufferFactory, headers);
public AbstractListenerServerHttpResponse(DataBufferFactory bufferFactory, HttpHeaders headers) {
super(bufferFactory, headers);
}
@@ -56,15 +56,15 @@ public abstract class AbstractListenerServerHttpResponse extends AbstractServerH
protected final Mono<Void> writeAndFlushWithInternal(
Publisher<? extends Publisher<? extends DataBuffer>> body) {
if (this.writeCalled.compareAndSet(false, true)) {
Processor<? super Publisher<? extends DataBuffer>, Void> processor = createBodyFlushProcessor();
return Mono.from(subscriber -> {
body.subscribe(processor);
processor.subscribe(subscriber);
});
if (!this.writeCalled.compareAndSet(false, true)) {
return Mono.error(new IllegalStateException(
"writeWith() or writeAndFlushWith() has already been called"));
}
return Mono.error(new IllegalStateException(
"writeWith() or writeAndFlushWith() has already been called"));
Processor<? super Publisher<? extends DataBuffer>, Void> processor = createBodyFlushProcessor();
return Mono.from(subscriber -> {
body.subscribe(processor);
processor.subscribe(subscriber);
});
}
/**

6
spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteFlushProcessor.java
View File

@@ -58,7 +58,7 @@ public abstract class AbstractListenerWriteFlushProcessor<T> implements Processo
@Nullable
private Subscription subscription;
private volatile boolean subscriberCompleted;
private volatile boolean sourceCompleted;
private final WriteResultPublisher resultPublisher;
@@ -293,7 +293,7 @@ public abstract class AbstractListenerWriteFlushProcessor<T> implements Processo
return;
}
if (processor.changeState(this, REQUESTED)) {
if (processor.subscriberCompleted) {
if (processor.sourceCompleted) {
handleSubscriberCompleted(processor);
}
else {
@@ -304,7 +304,7 @@ public abstract class AbstractListenerWriteFlushProcessor<T> implements Processo
}
@Override
public <T> void onComplete(AbstractListenerWriteFlushProcessor<T> processor) {
processor.subscriberCompleted = true;
processor.sourceCompleted = true;
// A competing write might have completed very quickly
if (processor.state.get().equals(State.REQUESTED)) {
handleSubscriberCompleted(processor);

10
spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2021 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -64,7 +64,7 @@ public abstract class AbstractListenerWriteProcessor<T> implements Processor<T,
private volatile T currentData;
/* Indicates "onComplete" was received during the (last) write. */
private volatile boolean subscriberCompleted;
private volatile boolean sourceCompleted;
/**
* Indicates we're waiting for one last isReady-onWritePossible cycle
@@ -374,7 +374,7 @@ public abstract class AbstractListenerWriteProcessor<T> implements Processor<T,
if (processor.write(data)) {
if (processor.changeState(WRITING, REQUESTED)) {
processor.currentData = null;
if (processor.subscriberCompleted) {
if (processor.sourceCompleted) {
processor.readyToCompleteAfterLastWrite = true;
processor.changeStateToReceived(REQUESTED);
}
@@ -397,7 +397,7 @@ public abstract class AbstractListenerWriteProcessor<T> implements Processor<T,
@Override
public <T> void onComplete(AbstractListenerWriteProcessor<T> processor) {
processor.subscriberCompleted = true;
processor.sourceCompleted = true;
// A competing write might have completed very quickly
if (processor.state.get().equals(State.REQUESTED)) {
processor.changeStateToComplete(State.REQUESTED);
@@ -408,7 +408,7 @@ public abstract class AbstractListenerWriteProcessor<T> implements Processor<T,
WRITING {
@Override
public <T> void onComplete(AbstractListenerWriteProcessor<T> processor) {
processor.subscriberCompleted = true;
processor.sourceCompleted = true;
// A competing write might have completed very quickly
if (processor.state.get().equals(State.REQUESTED)) {
processor.changeStateToComplete(State.REQUESTED);

17
spring-web/src/main/java/org/springframework/http/server/reactive/ServletHttpHandlerAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2021 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -239,15 +239,24 @@ public class ServletHttpHandlerAdapter implements Servlet {
}
/**
* AsyncListener to complete the {@link AsyncContext} in case of error or
* timeout notifications from the container
* <p>Additional {@link AsyncListener}s are registered in
* {@link ServletServerHttpRequest} to signal onError/onComplete to the
* request body Subscriber, and in {@link ServletServerHttpResponse} to
* cancel the write Publisher and signal onError/onComplete downstream to
* the writing result Subscriber.
*/
private static class HandlerResultAsyncListener implements AsyncListener {
private final AtomicBoolean isCompleted;
private final String logPrefix;
public HandlerResultAsyncListener(AtomicBoolean isCompleted, ServletServerHttpRequest httpRequest) {
public HandlerResultAsyncListener(AtomicBoolean isCompleted, ServletServerHttpRequest request) {
this.isCompleted = isCompleted;
this.logPrefix = httpRequest.getLogPrefix();
this.logPrefix = request.getLogPrefix();
}
@Override
@@ -277,7 +286,7 @@ public class ServletHttpHandlerAdapter implements Servlet {
}
private class HandlerResultSubscriber implements Subscriber<Void> {
private static class HandlerResultSubscriber implements Subscriber<Void> {
private final AsyncContext asyncContext;