Renamed 'rx' package to 'reactive'

spring-web-reactive/src/main/java/org/springframework/reactive/io/ByteArrayPublisherInputStream.java

@@ -1,4 +1,4 @@
-package org.springframework.rx.io;/*
+package org.springframework.reactive.io;/*
  * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -20,7 +20,7 @@ import java.io.InputStream;
 
 import org.reactivestreams.Publisher;
 
-import org.springframework.rx.util.BlockingSignalQueue;
+import org.springframework.reactive.util.BlockingSignalQueue;
 import org.springframework.util.Assert;
 
 /**

spring-web-reactive/src/main/java/org/springframework/reactive/io/ByteArrayPublisherOutputStream.java

@@ -1,4 +1,4 @@
-package org.springframework.rx.io;
+package org.springframework.reactive.io;
 
 import java.io.IOException;
 import java.io.OutputStream;
@@ -6,7 +6,7 @@ import java.util.Arrays;
 
 import org.reactivestreams.Publisher;
 
-import org.springframework.rx.util.BlockingSignalQueue;
+import org.springframework.reactive.util.BlockingSignalQueue;
 
 /**
  * {@code OutputStream} implementation that stores all written bytes, to be retrieved

spring-web-reactive/src/main/java/org/springframework/reactive/util/BlockingSignalQueue.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.rx.util;
+package org.springframework.reactive.util;
 
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;

spring-web-reactive/src/main/java/org/springframework/reactive/util/OnComplete.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.rx.util;
+package org.springframework.reactive.util;
 
 /**
  * @author Arjen Poutsma

spring-web-reactive/src/main/java/org/springframework/reactive/util/OnError.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.rx.util;
+package org.springframework.reactive.util;
 
 import org.springframework.util.Assert;
 

spring-web-reactive/src/main/java/org/springframework/reactive/util/OnNext.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.rx.util;
+package org.springframework.reactive.util;
 
 import org.springframework.util.Assert;
 

spring-web-reactive/src/main/java/org/springframework/reactive/util/Signal.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.rx.util;
+package org.springframework.reactive.util;
 
 /**
  * @author Arjen Poutsma

spring-web-reactive/src/main/java/org/springframework/reactive/web/HttpHandler.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.rx.web.servlet;
+package org.springframework.reactive.web;
 
 import org.reactivestreams.Publisher;
 

spring-web-reactive/src/main/java/org/springframework/reactive/web/servlet/AsyncContextSynchronizer.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.rx.web.servlet;
+package org.springframework.reactive.web.servlet;
 
 import java.io.IOException;
 import java.util.concurrent.atomic.AtomicInteger;

spring-web-reactive/src/main/java/org/springframework/reactive/web/servlet/HttpHandlerServlet.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.rx.web.servlet;
+package org.springframework.reactive.web.servlet;
 
 import java.io.IOException;
 import javax.servlet.AsyncContext;
@@ -26,6 +26,8 @@ import javax.servlet.http.HttpServletResponse;
 
 import org.reactivestreams.Publisher;
 
+import org.springframework.reactive.web.HttpHandler;
+
 /**
  * @author Arjen Poutsma
  */

spring-web-reactive/src/main/java/org/springframework/reactive/web/servlet/RequestBodyPublisher.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.rx.web.servlet;
+package org.springframework.reactive.web.servlet;
 
 import java.io.IOException;
 import java.nio.charset.Charset;

spring-web-reactive/src/main/java/org/springframework/reactive/web/servlet/ResponseBodySubscriber.java

@@ -14,10 +14,9 @@
  * limitations under the License.
  */
 
-package org.springframework.rx.web.servlet;
+package org.springframework.reactive.web.servlet;
 
 import java.io.IOException;
-import java.util.concurrent.atomic.AtomicBoolean;
 import javax.servlet.ServletOutputStream;
 import javax.servlet.WriteListener;
 
@@ -41,7 +40,7 @@ public class ResponseBodySubscriber implements WriteListener, Subscriber<byte[]>
 
 	private byte[] buffer;
 
-	private AtomicBoolean complete = new AtomicBoolean(false);
+	private volatile boolean subscriberComplete = false;
 
 	public ResponseBodySubscriber(AsyncContextSynchronizer synchronizer) {
 		this.synchronizer = synchronizer;
@@ -72,7 +71,9 @@ public class ResponseBodySubscriber implements WriteListener, Subscriber<byte[]>
 	public void onComplete() {
 		logger.debug("Complete buffer: " + (buffer == null));
 
-		if (complete.compareAndSet(false, true) && buffer == null) {
+		this.subscriberComplete = true;
+
+		if (buffer == null) {
 			this.synchronizer.writeComplete();
 		}
 	}
@@ -84,20 +85,22 @@ public class ResponseBodySubscriber implements WriteListener, Subscriber<byte[]>
 		boolean ready = output.isReady();
 		logger.debug("Output: " + ready + " buffer: " + (buffer == null));
 
-		if (this.buffer != null && ready) {
-			output.write(this.buffer);
-			this.buffer = null;
+		if (ready) {
+			if (this.buffer != null) {
+				output.write(this.buffer);
+				this.buffer = null;
 
-			if (!complete.get()) {
-				this.subscription.request(1);
+				if (!subscriberComplete) {
+					this.subscription.request(1);
+				}
+				else {
+					this.synchronizer.writeComplete();
+				}
 			}
 			else {
-				this.synchronizer.writeComplete();
+				this.subscription.request(1);
 			}
 		}
-		else if (this.buffer == null && ready) {
-			this.subscription.request(1);
-		}
 	}
 
 	@Override
@@ -105,7 +108,4 @@ public class ResponseBodySubscriber implements WriteListener, Subscriber<byte[]>
 		logger.error("ResponseBodySubscriber error", t);
 	}
 
-
-	private void complete() {
-	}
 }