From 327c4204096c9f2e7f3f17801f11393fb9d5e9e6 Mon Sep 17 00:00:00 2001
From: Sebastien Deleuze <sdeleuze@pivotal.io>
Date: Thu, 7 Jan 2016 14:51:43 +0100
Subject: [PATCH] Remove mandatory dependency on Reactor Stream with JarJar

---
 spring-web-reactive/build.gradle              | 48 ++++++++++++++++---
 .../util/ByteBufferPublisherInputStream.java  |  6 ++-
 .../http/server/reactive/XmlHandler.java      |  3 +-
 3 files changed, 46 insertions(+), 11 deletions(-)

diff --git a/spring-web-reactive/build.gradle b/spring-web-reactive/build.gradle
index 117cab0e2a..704f8abdfa 100644
--- a/spring-web-reactive/build.gradle
+++ b/spring-web-reactive/build.gradle
@@ -12,8 +12,45 @@ apply plugin: 'propdeps'
 apply plugin: 'propdeps-idea'
 apply plugin: 'propdeps-maven'
 
+ext {
+	springVersion = '4.2.3.RELEASE'
+	reactorVersion = '2.5.0.BUILD-SNAPSHOT'
+	tomcatVersion = '8.0.28'
+	jettyVersion = '9.3.5.v20151012'
+}
+
+configurations {
+	jarjar
+	reactorstream
+}
+
+task reactorstreamRepackJar(type: Jar) { repackJar ->
+	repackJar.baseName = "spring-reactive-reactorstream-repack"
+	repackJar.version = reactorVersion
+
+	doLast() {
+		project.ant {
+			taskdef name: "jarjar", classname: "com.tonicsystems.jarjar.JarJarTask",
+				classpath: configurations.jarjar.asPath
+			jarjar(destfile: repackJar.archivePath) {
+				configurations.reactorstream.each { originalJar ->
+					zipfileset(src: originalJar)
+				}
+				// repackage reactor. => org.springframework.reactor
+				rule(pattern: "reactor.rx.**", result: "org.springframework.reactor.rx.@1")
+			}
+		}
+	}
+}
+
+
 jar {
 	baseName = 'spring-reactive'
+
+	dependsOn reactorstreamRepackJar
+		from(zipTree(reactorstreamRepackJar.archivePath)) {
+			include "reactor/rx/subscriber/BlockingQueueSubscriber.java"
+		}
 }
 
 group = 'org.springframework.reactive'
@@ -30,19 +67,14 @@ configurations.all {
 	resolutionStrategy.cacheChangingModulesFor 0, 'seconds'
 }
 
-ext {
-	springVersion = '4.2.3.RELEASE'
-	reactorVersion = '2.5.0.BUILD-SNAPSHOT'
-	tomcatVersion = '8.0.28'
-	jettyVersion = '9.3.5.v20151012'
-}
-
 dependencies {
 	compile "org.springframework:spring-core:${springVersion}"
 	compile "org.springframework:spring-web:${springVersion}"
 	compile "org.reactivestreams:reactive-streams:1.0.0"
 	compile "io.projectreactor:reactor-core:${reactorVersion}"
 	compile "commons-logging:commons-logging:1.2"
+	reactorstream("io.projectreactor:reactor-stream:${reactorVersion}@jar")
+	compile(files(reactorstreamRepackJar))
 
 	optional 'io.reactivex:rxjava:1.1.0'
 	optional "io.reactivex:rxnetty-http:0.5.0-SNAPSHOT"
@@ -57,6 +89,8 @@ dependencies {
 
 	provided "javax.servlet:javax.servlet-api:3.1.0"
 
+	jarjar("com.googlecode.jarjar:jarjar:1.3")
+
 	testCompile "junit:junit:4.12"
 	testCompile "org.springframework:spring-test:${springVersion}"
 	testCompile "org.slf4j:slf4j-jcl:1.7.12"
diff --git a/spring-web-reactive/src/main/java/org/springframework/util/ByteBufferPublisherInputStream.java b/spring-web-reactive/src/main/java/org/springframework/util/ByteBufferPublisherInputStream.java
index 10aa4c23d2..90d6b422ff 100644
--- a/spring-web-reactive/src/main/java/org/springframework/util/ByteBufferPublisherInputStream.java
+++ b/spring-web-reactive/src/main/java/org/springframework/util/ByteBufferPublisherInputStream.java
@@ -19,11 +19,13 @@ package org.springframework.util;
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.ByteBuffer;
+import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscription;
-import reactor.rx.Streams;
+
+import org.springframework.reactor.rx.subscriber.BlockingQueueSubscriber;
 
 /**
  * {@code InputStream} implementation based on a byte array {@link Publisher}.
@@ -60,7 +62,7 @@ public class ByteBufferPublisherInputStream extends InputStream {
 	public ByteBufferPublisherInputStream(Publisher<ByteBuffer> publisher, int requestSize) {
 		Assert.notNull(publisher, "'publisher' must not be null");
 
-		this.queue = Streams.from(publisher).toBlockingQueue(requestSize);
+		this.queue = new BlockingQueueSubscriber<>(publisher, null, new ArrayBlockingQueue<>(requestSize), false, requestSize);
 	}
 
 
diff --git a/spring-web-reactive/src/test/java/org/springframework/http/server/reactive/XmlHandler.java b/spring-web-reactive/src/test/java/org/springframework/http/server/reactive/XmlHandler.java
index dd1a4278de..39e7be5c9a 100644
--- a/spring-web-reactive/src/test/java/org/springframework/http/server/reactive/XmlHandler.java
+++ b/spring-web-reactive/src/test/java/org/springframework/http/server/reactive/XmlHandler.java
@@ -24,7 +24,6 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import reactor.Mono;
 import reactor.io.buffer.Buffer;
-import reactor.rx.Streams;
 
 import org.springframework.http.MediaType;
 import org.springframework.util.BufferOutputStream;
@@ -73,7 +72,7 @@ public class XmlHandler implements HttpHandler {
 			bos.close();
 			buffer.flip();
 
-			return response.setBody(Streams.just(buffer.byteBuffer()));
+			return response.setBody(Mono.just(buffer.byteBuffer()));
 		}
 		catch (Exception ex) {
 			logger.error(ex, ex);