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

Fix race condition in transition from UNSUBSCRIBED->COMPLETED

Browse Source 
- Ensure completion signal (normal/exception) will be delivered to
the subscriber when transition from UNSUBSCRIBED->COMPLETED

- According to the specification "Publisher.subscribe MUST call onSubscribe
on the provided Subscriber prior to any other signals to that Subscriber" so
ensure onComplete/onError signals will be called AFTER onSubscribe signal.

Issue: SPR-16207
This commit is contained in:
Violeta Georgieva
2017-11-21 12:13:41 +02:00
committed by Rossen Stoyanchev
parent 41b13a4e8a
commit b814875211
2 changed files with 68 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerReadPublisher.java
View File

@@ -52,6 +52,11 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
private volatile long demand;
private volatile boolean publisherCompleted;
@Nullable
private volatile Throwable publisherError;
@SuppressWarnings("rawtypes")
private static final AtomicLongFieldUpdater<AbstractListenerReadPublisher> DEMAND_FIELD_UPDATER =
AtomicLongFieldUpdater.newUpdater(AbstractListenerReadPublisher.class, "demand");
@@ -208,15 +213,54 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
<T> void subscribe(AbstractListenerReadPublisher<T> publisher, Subscriber<? super T> subscriber) {
Assert.notNull(publisher, "Publisher must not be null");
Assert.notNull(subscriber, "Subscriber must not be null");
if (publisher.changeState(this, NO_DEMAND)) {
if (publisher.changeState(this, SUBSCRIBING)) {
Subscription subscription = new ReadSubscription(publisher);
publisher.subscriber = subscriber;
subscriber.onSubscribe(subscription);
publisher.changeState(SUBSCRIBING, NO_DEMAND);
if (publisher.publisherCompleted) {
publisher.onAllDataRead();
}
Throwable publisherError = publisher.publisherError;
if (publisherError != null) {
publisher.onError(publisherError);
}
}
else {
throw new IllegalStateException(toString());
}
}
@Override
<T> void onAllDataRead(AbstractListenerReadPublisher<T> publisher) {
publisher.publisherCompleted = true;
}
@Override
<T> void onError(AbstractListenerReadPublisher<T> publisher, Throwable t) {
publisher.publisherError = t;
}
},
SUBSCRIBING {
<T> void request(AbstractListenerReadPublisher<T> publisher, long n) {
if (Operators.validate(n)) {
Operators.addCap(DEMAND_FIELD_UPDATER, publisher, n);
if (publisher.changeState(this, DEMAND)) {
publisher.checkOnDataAvailable();
}
}
}
@Override
<T> void onAllDataRead(AbstractListenerReadPublisher<T> publisher) {
publisher.publisherCompleted = true;
}
@Override
<T> void onError(AbstractListenerReadPublisher<T> publisher, Throwable t) {
publisher.publisherError = t;
}
},
/**

40
spring-web/src/main/java/org/springframework/http/server/reactive/WriteResultPublisher.java
View File

@@ -119,18 +119,17 @@ class WriteResultPublisher implements Publisher<Void> {
@Override
void subscribe(WriteResultPublisher publisher, Subscriber<? super Void> subscriber) {
Assert.notNull(subscriber, "Subscriber must not be null");
publisher.subscriber = subscriber;
if (publisher.changeState(this, SUBSCRIBED)) {
if (publisher.changeState(this, SUBSCRIBING)) {
Subscription subscription = new ResponseBodyWriteResultSubscription(publisher);
publisher.subscriber = subscriber;
subscriber.onSubscribe(subscription);
publisher.changeState(SUBSCRIBING, SUBSCRIBED);
if (publisher.publisherCompleted) {
publisher.publishComplete();
}
else {
Throwable publisherError = publisher.publisherError;
if (publisherError != null) {
publisher.publishError(publisherError);
}
Throwable publisherError = publisher.publisherError;
if (publisherError != null) {
publisher.publishError(publisherError);
}
}
else {
@@ -147,6 +146,21 @@ class WriteResultPublisher implements Publisher<Void> {
}
},
SUBSCRIBING {
@Override
void request(WriteResultPublisher publisher, long n) {
Operators.validate(n);
}
@Override
void publishComplete(WriteResultPublisher publisher) {
publisher.publisherCompleted = true;
}
@Override
void publishError(WriteResultPublisher publisher, Throwable t) {
publisher.publisherError = t;
}
},
SUBSCRIBED {
@Override
void request(WriteResultPublisher publisher, long n) {
@@ -183,14 +197,6 @@ class WriteResultPublisher implements Publisher<Void> {
void cancel(WriteResultPublisher publisher) {
// ignore
}
@Override
void publishComplete(WriteResultPublisher publisher) {
// ignore
}
@Override
void publishError(WriteResultPublisher publisher, Throwable t) {
// ignore
}
};
void subscribe(WriteResultPublisher publisher, Subscriber<? super Void> subscriber) {
@@ -208,11 +214,11 @@ class WriteResultPublisher implements Publisher<Void> {
}
void publishComplete(WriteResultPublisher publisher) {
throw new IllegalStateException(toString());
// ignore
}
void publishError(WriteResultPublisher publisher, Throwable t) {
throw new IllegalStateException(toString());
// ignore
}
}