DATAMONGO-1707 - Upgrade to Reactor 3.1 M2.

Adopt to API change from Publisher.subscribe() to Publisher.toProcessor(). Adopt to changed reactor-test groupId. Provide mocks for calls that allowed previously null Publishers.

spring-data-mongodb/pom.xml

@@ -111,7 +111,7 @@
 		</dependency>
 
 		<dependency>
-		 	<groupId>io.projectreactor.addons</groupId>
+			<groupId>io.projectreactor</groupId>
 			<artifactId>reactor-test</artifactId>
 			<version>${reactor}</version>
 			<optional>true</optional>

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/ReactiveMongoParameterAccessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -15,16 +15,16 @@
  */
 package org.springframework.data.mongodb.repository.query;
 
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.MonoProcessor;
+
 import java.util.ArrayList;
 import java.util.List;
 
 import org.springframework.data.repository.util.ReactiveWrapperConverters;
 import org.springframework.data.repository.util.ReactiveWrappers;
 
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.core.publisher.MonoProcessor;
-
 /**
  * Reactive {@link org.springframework.data.repository.query.ParametersParameterAccessor} implementation that subscribes
  * to reactive parameter wrapper types upon creation. This class performs synchronization when acessing parameters.
@@ -55,9 +55,9 @@ class ReactiveMongoParameterAccessor extends MongoParametersParameterAccessor {
 			}
 
 			if (ReactiveWrappers.isSingleValueType(value.getClass())) {
-				subscriptions.add(ReactiveWrapperConverters.toWrapper(value, Mono.class).subscribe());
+				subscriptions.add(ReactiveWrapperConverters.toWrapper(value, Mono.class).toProcessor());
 			} else {
-				subscriptions.add(ReactiveWrapperConverters.toWrapper(value, Flux.class).collectList().subscribe());
+				subscriptions.add(ReactiveWrapperConverters.toWrapper(value, Flux.class).collectList().toProcessor());
 			}
 		}
 	}

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateUnitTests.java

@@ -21,6 +21,8 @@ import static org.mockito.Mockito.*;
 import static org.mockito.Mockito.any;
 import static org.springframework.data.mongodb.core.aggregation.Aggregation.*;
 
+import reactor.core.publisher.Mono;
+
 import org.bson.Document;
 import org.junit.Before;
 import org.junit.Ignore;
@@ -137,6 +139,8 @@ public class ReactiveMongoTemplateUnitTests {
 	@Test // DATAMONGO-1518
 	public void findAndModfiyShoudUseCollationWhenPresent() {
 
+		when(collection.findOneAndUpdate(any(), any(), any())).thenReturn(Mono.empty());
+
 		template.findAndModify(new BasicQuery("{}").collation(Collation.of("fr")), new Update(), AutogenerateableId.class)
 				.subscribe();
 
@@ -149,6 +153,8 @@ public class ReactiveMongoTemplateUnitTests {
 	@Test // DATAMONGO-1518
 	public void findAndRemoveShouldUseCollationWhenPresent() {
 
+		when(collection.findOneAndDelete(any(), any())).thenReturn(Mono.empty());
+
 		template.findAndRemove(new BasicQuery("{}").collation(Collation.of("fr")), AutogenerateableId.class).subscribe();
 
 		ArgumentCaptor<FindOneAndDeleteOptions> options = ArgumentCaptor.forClass(FindOneAndDeleteOptions.class);
@@ -174,6 +180,8 @@ public class ReactiveMongoTemplateUnitTests {
 	@Test // DATAMONGO-1518
 	public void updateOneShouldUseCollationWhenPresent() {
 
+		when(collection.updateOne(any(), any(), any())).thenReturn(Mono.empty());
+
 		template.updateFirst(new BasicQuery("{}").collation(Collation.of("fr")), new Update().set("foo", "bar"),
 				AutogenerateableId.class).subscribe();
 
@@ -186,6 +194,8 @@ public class ReactiveMongoTemplateUnitTests {
 	@Test // DATAMONGO-1518
 	public void updateManyShouldUseCollationWhenPresent() {
 
+		when(collection.updateMany(any(), any(), any())).thenReturn(Mono.empty());
+
 		template.updateMulti(new BasicQuery("{}").collation(Collation.of("fr")), new Update().set("foo", "bar"),
 				AutogenerateableId.class).subscribe();
 
@@ -199,6 +209,8 @@ public class ReactiveMongoTemplateUnitTests {
 	@Test // DATAMONGO-1518
 	public void replaceOneShouldUseCollationWhenPresent() {
 
+		when(collection.replaceOne(any(), any(), any())).thenReturn(Mono.empty());
+
 		template.updateFirst(new BasicQuery("{}").collation(Collation.of("fr")), new Update(), AutogenerateableId.class)
 				.subscribe();