From d0836773bafcc870e8f66dfacbad0a4c6c8314c8 Mon Sep 17 00:00:00 2001
From: Mark Paluch <mpaluch@pivotal.io>
Date: Thu, 10 Nov 2016 17:07:10 +0100
Subject: [PATCH] DATACMNS-836 - Apply deferred reactive type conversion as
 RxJava's PublisherAdapter keep state between multiple subscribers.

---
 .../util/ReactiveWrapperConverters.java       | 90 +++++++++++--------
 1 file changed, 51 insertions(+), 39 deletions(-)

diff --git a/src/main/java/org/springframework/data/repository/util/ReactiveWrapperConverters.java b/src/main/java/org/springframework/data/repository/util/ReactiveWrapperConverters.java
index 6e5ed26aa..d2a90a915 100644
--- a/src/main/java/org/springframework/data/repository/util/ReactiveWrapperConverters.java
+++ b/src/main/java/org/springframework/data/repository/util/ReactiveWrapperConverters.java
@@ -358,6 +358,10 @@ public class ReactiveWrapperConverters {
 		}
 	}
 
+	// -------------------------------------------------------------------------
+	// ReactiveStreams converters
+	// -------------------------------------------------------------------------
+
 	/**
 	 * A {@link Converter} to convert a {@link Publisher} to {@link Flux}.
 	 *
@@ -390,6 +394,10 @@ public class ReactiveWrapperConverters {
 		}
 	}
 
+	// -------------------------------------------------------------------------
+	// RxJava 1 converters
+	// -------------------------------------------------------------------------
+
 	/**
 	 * A {@link Converter} to convert a {@link Publisher} to {@link Single}.
 	 *
@@ -450,7 +458,7 @@ public class ReactiveWrapperConverters {
 
 		@Override
 		public Publisher<?> convert(Single<?> source) {
-			return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Single.class).toPublisher(source);
+			return Flux.defer(() -> REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Single.class).toPublisher(source));
 		}
 	}
 
@@ -466,7 +474,7 @@ public class ReactiveWrapperConverters {
 
 		@Override
 		public Mono<?> convert(Single<?> source) {
-			return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Single.class).toMono(source);
+			return Mono.defer(() -> REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Single.class).toMono(source));
 		}
 	}
 
@@ -482,7 +490,7 @@ public class ReactiveWrapperConverters {
 
 		@Override
 		public Flux<?> convert(Single<?> source) {
-			return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Single.class).toFlux(source);
+			return Flux.defer(() -> REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Single.class).toFlux(source));
 		}
 	}
 
@@ -498,7 +506,7 @@ public class ReactiveWrapperConverters {
 
 		@Override
 		public Publisher<?> convert(Completable source) {
-			return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Completable.class).toFlux(source);
+			return Flux.defer(() -> REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Completable.class).toFlux(source));
 		}
 	}
 
@@ -530,7 +538,7 @@ public class ReactiveWrapperConverters {
 
 		@Override
 		public Publisher<?> convert(Observable<?> source) {
-			return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Observable.class).toFlux(source);
+			return Flux.defer(() -> REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Observable.class).toFlux(source));
 		}
 	}
 
@@ -546,7 +554,7 @@ public class ReactiveWrapperConverters {
 
 		@Override
 		public Mono<?> convert(Observable<?> source) {
-			return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Observable.class).toMono(source);
+			return Mono.defer(() -> REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Observable.class).toMono(source));
 		}
 	}
 
@@ -562,10 +570,46 @@ public class ReactiveWrapperConverters {
 
 		@Override
 		public Flux<?> convert(Observable<?> source) {
-			return REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Observable.class).toFlux(source);
+			return Flux.defer(() -> REACTIVE_ADAPTER_REGISTRY.getAdapterFrom(Observable.class).toFlux(source));
 		}
 	}
 
+	/**
+	 * A {@link Converter} to convert a {@link Observable} to {@link Single}.
+	 *
+	 * @author Mark Paluch
+	 * @author 2.0
+	 */
+	public enum RxJava1ObservableToSingleConverter implements Converter<Observable<?>, Single<?>> {
+
+		INSTANCE;
+
+		@Override
+		public Single<?> convert(Observable<?> source) {
+			return source.toSingle();
+		}
+	}
+
+	/**
+	 * A {@link Converter} to convert a {@link Single} to {@link Single}.
+	 *
+	 * @author Mark Paluch
+	 * @author 2.0
+	 */
+	public enum RxJava1SingleToObservableConverter implements Converter<Single<?>, Observable<?>> {
+
+		INSTANCE;
+
+		@Override
+		public Observable<?> convert(Single<?> source) {
+			return source.toObservable();
+		}
+	}
+
+	// -------------------------------------------------------------------------
+	// RxJava 2 converters
+	// -------------------------------------------------------------------------
+
 	/**
 	 * A {@link Converter} to convert a {@link Publisher} to {@link io.reactivex.Single}.
 	 *
@@ -841,38 +885,6 @@ public class ReactiveWrapperConverters {
 		}
 	}
 
-	/**
-	 * A {@link Converter} to convert a {@link Observable} to {@link Single}.
-	 *
-	 * @author Mark Paluch
-	 * @author 2.0
-	 */
-	public enum RxJava1ObservableToSingleConverter implements Converter<Observable<?>, Single<?>> {
-
-		INSTANCE;
-
-		@Override
-		public Single<?> convert(Observable<?> source) {
-			return source.toSingle();
-		}
-	}
-
-	/**
-	 * A {@link Converter} to convert a {@link Single} to {@link Single}.
-	 *
-	 * @author Mark Paluch
-	 * @author 2.0
-	 */
-	public enum RxJava1SingleToObservableConverter implements Converter<Single<?>, Observable<?>> {
-
-		INSTANCE;
-
-		@Override
-		public Observable<?> convert(Single<?> source) {
-			return source.toObservable();
-		}
-	}
-
 	/**
 	 * A {@link Converter} to convert a {@link Observable} to {@link Single}.
 	 *