root
/
spring-framework
mirror of https://github.com/spring-projects/spring-framework.git
1
0
Fork 0
Code Issues Actions 8 Packages Projects Releases Wiki Activity

referencing fields of a class using this

This commit is contained in:
Marek Hawrylczak 2015-11-07 16:37:40 +01:00
parent 9126fd9dec
commit f1bec5f1e4
6 changed files with 77 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/RequestBodyPublisher.java
View File

@ -55,12 +55,12 @@ class RequestBodyPublisher implements Publisher<ByteBuffer> {
if (s == null) {
throw SpecificationExceptions.spec_2_13_exception();
}
if (subscriber != null) {
if (this.subscriber != null) {
s.onError(new IllegalStateException("Only one subscriber allowed"));
}
subscriber = s;
subscriber.onSubscribe(new RequestBodySubscription());
this.subscriber = s;
this.subscriber.onSubscribe(new RequestBodySubscription());
}
private class RequestBodySubscription
@ -74,7 +74,7 @@ class RequestBodyPublisher implements Publisher<ByteBuffer> {
@Override
public void cancel() {
subscriptionClosed = true;
this.subscriptionClosed = true;
close();
}
@ -82,7 +82,7 @@ class RequestBodyPublisher implements Publisher<ByteBuffer> {
public void request(long n) {
BackpressureUtils.checkRequest(n, subscriber);
if (subscriptionClosed) {
if (this.subscriptionClosed) {
return;
}
@ -96,13 +96,13 @@ class RequestBodyPublisher implements Publisher<ByteBuffer> {
}
private void doOnNext(ByteBuffer buffer) {
draining = false;
this.draining = false;
buffer.flip();
subscriber.onNext(buffer);
}
private void doOnComplete() {
subscriptionClosed = true;
this.subscriptionClosed = true;
try {
subscriber.onComplete();
}
@ -112,7 +112,7 @@ class RequestBodyPublisher implements Publisher<ByteBuffer> {
}
private void doOnError(Throwable t) {
subscriptionClosed = true;
this.subscriptionClosed = true;
try {
subscriber.onError(t);
}
@ -122,19 +122,19 @@ class RequestBodyPublisher implements Publisher<ByteBuffer> {
}
private void close() {
if (pooledBuffer != null) {
safeClose(pooledBuffer);
pooledBuffer = null;
if (this.pooledBuffer != null) {
safeClose(this.pooledBuffer);
this.pooledBuffer = null;
}
if (channel != null) {
safeClose(channel);
channel = null;
if (this.channel != null) {
safeClose(this.channel);
this.channel = null;
}
}
@Override
public void run() {
if (subscriptionClosed || draining) {
if (this.subscriptionClosed || this.draining) {
return;
}
@ -142,12 +142,12 @@ class RequestBodyPublisher implements Publisher<ByteBuffer> {
return;
}
draining = true;
this.draining = true;
if (channel == null) {
channel = exchange.getRequestChannel();
if (this.channel == null) {
this.channel = exchange.getRequestChannel();
if (channel == null) {
if (this.channel == null) {
if (exchange.isRequestComplete()) {
return;
}
@ -157,21 +157,21 @@ class RequestBodyPublisher implements Publisher<ByteBuffer> {
}
}
}
if (pooledBuffer == null) {
pooledBuffer = exchange.getConnection().getByteBufferPool().allocate();
if (this.pooledBuffer == null) {
this.pooledBuffer = exchange.getConnection().getByteBufferPool().allocate();
}
else {
pooledBuffer.getBuffer().clear();
this.pooledBuffer.getBuffer().clear();
}
try {
ByteBuffer buffer = pooledBuffer.getBuffer();
ByteBuffer buffer = this.pooledBuffer.getBuffer();
int count;
do {
count = channel.read(buffer);
count = this.channel.read(buffer);
if (count == 0) {
channel.getReadSetter().set(this);
channel.resumeReads();
this.channel.getReadSetter().set(this);
this.channel.resumeReads();
}
else if (count == -1) {
if (buffer.position() > 0) {
@ -181,11 +181,11 @@ class RequestBodyPublisher implements Publisher<ByteBuffer> {
}
else {
if (buffer.remaining() == 0) {
if (demand == 0) {
channel.suspendReads();
if (this.demand == 0) {
this.channel.suspendReads();
}
doOnNext(buffer);
if (demand > 0) {
if (this.demand > 0) {
scheduleNextMessage();
}
break;
@ -200,12 +200,12 @@ class RequestBodyPublisher implements Publisher<ByteBuffer> {
@Override
public void handleEvent(StreamSourceChannel channel) {
if (subscriptionClosed) {
if (this.subscriptionClosed) {
return;
}
try {
ByteBuffer buffer = pooledBuffer.getBuffer();
ByteBuffer buffer = this.pooledBuffer.getBuffer();
int count;
do {
count = channel.read(buffer);
@ -220,11 +220,11 @@ class RequestBodyPublisher implements Publisher<ByteBuffer> {
}
else {
if (buffer.remaining() == 0) {
if (demand == 0) {
if (this.demand == 0) {
channel.suspendReads();
}
doOnNext(buffer);
if (demand > 0) {
if (this.demand > 0) {
scheduleNextMessage();
}
break;

2
spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/RequestHandlerAdapter.java
View File

@ -50,7 +50,7 @@ class RequestHandlerAdapter implements io.undertow.server.HttpHandler {
new UndertowServerHttpResponse(exchange, responseBodySubscriber);
exchange.dispatch();
httpHandler.handle(request, response).subscribe(new Subscriber<Void>() {
this.httpHandler.handle(request, response).subscribe(new Subscriber<Void>() {
@Override
public void onSubscribe(Subscription subscription) {
subscription.request(Long.MAX_VALUE);

66
spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/ResponseBodySubscriber.java
View File

@ -60,29 +60,29 @@ class ResponseBodySubscriber extends BaseSubscriber<ByteBuffer>
@Override
public void onSubscribe(Subscription s) {
super.onSubscribe(s);
subscription = s;
subscription.request(1);
this.subscription = s;
this.subscription.request(1);
}
@Override
public void onNext(ByteBuffer buffer) {
super.onNext(buffer);
if (responseChannel == null) {
responseChannel = exchange.getResponseChannel();
if (this.responseChannel == null) {
this.responseChannel = this.exchange.getResponseChannel();
}
writing.incrementAndGet();
this.writing.incrementAndGet();
try {
int c;
do {
c = responseChannel.write(buffer);
c = this.responseChannel.write(buffer);
} while (buffer.hasRemaining() && c > 0);
if (buffer.hasRemaining()) {
writing.incrementAndGet();
this.writing.incrementAndGet();
enqueue(buffer);
responseChannel.getWriteSetter().set(this);
responseChannel.resumeWrites();
this.responseChannel.getWriteSetter().set(this);
this.responseChannel.resumeWrites();
}
else {
this.subscription.request(1);
@ -93,8 +93,8 @@ class ResponseBodySubscriber extends BaseSubscriber<ByteBuffer>
onError(ex);
}
finally {
writing.decrementAndGet();
if (closing.get()) {
this.writing.decrementAndGet();
if (this.closing.get()) {
closeIfDone();
}
}
@ -103,12 +103,12 @@ class ResponseBodySubscriber extends BaseSubscriber<ByteBuffer>
private void enqueue(ByteBuffer src) {
do {
PooledByteBuffer pooledBuffer =
exchange.getConnection().getByteBufferPool().allocate();
this.exchange.getConnection().getByteBufferPool().allocate();
ByteBuffer dst = pooledBuffer.getBuffer();
copy(dst, src);
dst.flip();
buffers.add(pooledBuffer);
this.buffers.add(pooledBuffer);
} while (src.remaining() > 0);
}
@ -124,25 +124,25 @@ class ResponseBodySubscriber extends BaseSubscriber<ByteBuffer>
try {
int c;
do {
ByteBuffer buffer = buffers.peek().getBuffer();
ByteBuffer buffer = this.buffers.peek().getBuffer();
do {
c = channel.write(buffer);
} while (buffer.hasRemaining() && c > 0);
if (!buffer.hasRemaining()) {
safeClose(buffers.remove());
safeClose(this.buffers.remove());
}
} while (!buffers.isEmpty() && c > 0);
if (!buffers.isEmpty()) {
} while (!this.buffers.isEmpty() && c > 0);
if (!this.buffers.isEmpty()) {
channel.resumeWrites();
}
else {
writing.decrementAndGet();
this.writing.decrementAndGet();
if (closing.get()) {
if (this.closing.get()) {
closeIfDone();
}
else {
subscription.request(1);
this.subscription.request(1);
}
}
}
@ -154,10 +154,10 @@ class ResponseBodySubscriber extends BaseSubscriber<ByteBuffer>
@Override
public void onError(Throwable t) {
super.onError(t);
if (!exchange.isResponseStarted() &&
exchange.getStatusCode() < INTERNAL_SERVER_ERROR.value()) {
if (!this.exchange.isResponseStarted() &&
this.exchange.getStatusCode() < INTERNAL_SERVER_ERROR.value()) {
exchange.setStatusCode(INTERNAL_SERVER_ERROR.value());
this.exchange.setStatusCode(INTERNAL_SERVER_ERROR.value());
}
logger.error("ResponseBodySubscriber error", t);
}
@ -166,15 +166,15 @@ class ResponseBodySubscriber extends BaseSubscriber<ByteBuffer>
public void onComplete() {
super.onComplete();
if (responseChannel != null) {
closing.set(true);
if (this.responseChannel != null) {
this.closing.set(true);
closeIfDone();
}
}
private void closeIfDone() {
if (writing.get() == 0) {
if (closing.compareAndSet(true, false)) {
if (this.writing.get() == 0) {
if (this.closing.compareAndSet(true, false)) {
closeChannel();
}
}
@ -182,16 +182,16 @@ class ResponseBodySubscriber extends BaseSubscriber<ByteBuffer>
private void closeChannel() {
try {
responseChannel.shutdownWrites();
this.responseChannel.shutdownWrites();
if (!responseChannel.flush()) {
responseChannel.getWriteSetter().set(
if (!this.responseChannel.flush()) {
this.responseChannel.getWriteSetter().set(
flushingChannelListener(
o -> safeClose(responseChannel),
o -> safeClose(this.responseChannel),
closingChannelExceptionHandler()));
responseChannel.resumeWrites();
this.responseChannel.resumeWrites();
}
responseChannel = null;
this.responseChannel = null;
}
catch (IOException ex) {
onError(ex);

6
spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowHttpServer.java
View File

@ -40,7 +40,7 @@ public class UndertowHttpServer extends HttpServerSupport
HttpHandler handler = new RequestHandlerAdapter(getHttpHandler());
undertowServer = Undertow.builder()
this.undertowServer = Undertow.builder()
.addHttpListener(getPort() != -1 ? getPort() : 8080, "localhost")
.setHandler(handler)
.build();
@ -49,7 +49,7 @@ public class UndertowHttpServer extends HttpServerSupport
@Override
public void start() {
if (!running) {
undertowServer.start();
this.undertowServer.start();
running = true;
}
@ -58,7 +58,7 @@ public class UndertowHttpServer extends HttpServerSupport
@Override
public void stop() {
if (running) {
undertowServer.stop();
this.undertowServer.stop();
running = false;
}
}

10
spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowServerHttpRequest.java
View File

@ -54,15 +54,15 @@ class UndertowServerHttpRequest implements ReactiveServerHttpRequest {
@Override
public HttpMethod getMethod() {
return HttpMethod.valueOf(exchange.getRequestMethod().toString());
return HttpMethod.valueOf(this.exchange.getRequestMethod().toString());
}
@Override
public URI getURI() {
try {
StringBuilder uri = new StringBuilder(exchange.getRequestPath());
if (StringUtils.hasLength(exchange.getQueryString())) {
uri.append('?').append(exchange.getQueryString());
StringBuilder uri = new StringBuilder(this.exchange.getRequestPath());
if (StringUtils.hasLength(this.exchange.getQueryString())) {
uri.append('?').append(this.exchange.getQueryString());
}
return new URI(uri.toString());
}
@ -75,7 +75,7 @@ class UndertowServerHttpRequest implements ReactiveServerHttpRequest {
public HttpHeaders getHeaders() {
if (this.headers == null) {
this.headers = new HttpHeaders();
for (HeaderValues headerValues : exchange.getRequestHeaders()) {
for (HeaderValues headerValues : this.exchange.getRequestHeaders()) {
for (String value : headerValues) {
this.headers.add(headerValues.getHeaderName().toString(), value);
}

4
spring-web-reactive/src/main/java/org/springframework/reactive/web/http/undertow/UndertowServerHttpResponse.java
View File

@ -51,7 +51,7 @@ class UndertowServerHttpResponse implements ReactiveServerHttpResponse {
@Override
public void setStatusCode(HttpStatus status) {
exchange.setStatusCode(status.value());
this.exchange.setStatusCode(status.value());
}
@Override
@ -103,7 +103,7 @@ class UndertowServerHttpResponse implements ReactiveServerHttpResponse {
if (!this.headersWritten) {
for (Map.Entry<String, List<String>> entry : this.headers.entrySet()) {
String headerName = entry.getKey();
exchange.getResponseHeaders()
this.exchange.getResponseHeaders()
.addAll(HttpString.tryFromString(headerName), entry.getValue());
}