fix reactor#1529 Prevent StepVerifier collect hang when consuming all…

… signals

Fix has also been tested against test [RecordTest](https://github.com/george-hawkins/publish-cancel-test/blob/master/src/test/java/com/example/RecordTest.java) provided by the issue author.

Everything worked fine if publisher ended with `onComplete` signal.
Without `onComplete` ,  `completeLatch` was never decremented and as a result 
verify step blocked forever.

reactor-test/src/main/java/reactor/test/DefaultStepVerifierBuilder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2011-2018 Pivotal Software Inc, All Rights Reserved.
+ * Copyright (c) 2011-2019 Pivotal Software Inc, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -1307,7 +1307,7 @@ boolean onCollect(Signal<T> actualSignal) {
 				this.completeLatch.countDown();
 				return true;
 			}
-			return true;
+			return false;
 		}
 
 		@SuppressWarnings("unchecked")

reactor-test/src/test/java/reactor/test/StepVerifierTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2011-2017 Pivotal Software Inc, All Rights Reserved.
+ * Copyright (c) 2011-2019 Pivotal Software Inc, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -1220,6 +1220,50 @@ public void testThenConsumeWhileFails() {
 		        .withMessageContaining("expectNext(9)");
 	}
 
+	@Test
+	public void testExpectRecordedMatches() {
+		List<Integer> expected = Arrays.asList(1,2);
+
+		StepVerifier.create(Flux.just(1,2))
+		            .recordWith(ArrayList::new)
+		            .thenConsumeWhile(i -> i < 2)
+		            .expectRecordedMatches(expected::equals)
+		            .thenCancel()
+		            .verify();
+	}
+
+	@Test
+	public void testExpectRecordedMatchesTwice() {
+		List<Integer> expected1 = Arrays.asList(1,2);
+		List<Integer> expected2 = Arrays.asList(3,4);
+
+		StepVerifier.create(Flux.just(1,2,3,4))
+		            .recordWith(ArrayList::new)
+		            .thenConsumeWhile(i -> i < 2)
+		            .expectRecordedMatches(expected1::equals)
+		            .recordWith(ArrayList::new)
+		            .thenConsumeWhile(i -> i < 4)
+		            .expectRecordedMatches(expected2::equals)
+		            .thenCancel()
+		            .verify();
+	}
+
+	@Test
+	public void testExpectRecordedMatchesWithoutComplete() {
+		List<Integer> expected = Arrays.asList(1,2);
+
+		TestPublisher<Integer> publisher = TestPublisher.createCold();
+		publisher.next(1);
+		publisher.next(2);
+
+		StepVerifier.create(publisher)
+		            .recordWith(ArrayList::new)
+		            .thenConsumeWhile(i -> i < 2)
+		            .expectRecordedMatches(expected::equals)
+		            .thenCancel()
+		            .verify();
+	}
+
 	@Test
 	public void testWithDescription() {
 		assertThatExceptionOfType(AssertionError.class)