Repository: reactor/reactor-core
Issue: repeatWhenEmpty hangs on cancel when discard hook is present

## Expected Behavior
It should be safe to cancel the sequence.

## Actual Behavior
Cancelling a sequence which is in a `repeatWhenEmpty()` loop will cause issues when a discard hook has been registered. In that case `Operators.onDiscardMultiple()` is invoked with the iterator used to keep track of the current iteration and the hook would be called for every remaining number up to `Long.MAX_VALUE`. 

## Steps to Reproduce
This test case will succeed when removing the `doOnDiscard()` operator.

```java
@Test
public void shouldCompleteEventually()
{
    StepVerifier.create(Mono.empty()
                            .repeatWhenEmpty(Repeat.once())
                            .doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release))
                .thenAwait()
                .thenCancel()
                .verify(Duration.ofSeconds(5));
}
```

## Possible Solution
Probably we should treat the remaining iterations in a way that they can be safely discarded without bothering any application level hooks. Also, I guess `FluxStream` should not be considered "knownToBeFinite" if its size is `Long.MAX_VALUE`.

## Your Environment
* Reactor version(s) used: 3.3.6.RELEASE


Changed files:
reactor-core/src/test/java/reactor/core/publisher/MonoRepeatWhenEmptyTest.java, reactor-core/src/main/java/reactor/core/publisher/Mono.java

diff --git a/reactor-core/src/main/java/reactor/core/publisher/Mono.java b/reactor-core/src/main/java/reactor/core/publisher/Mono.java
index 6cd3bd838..14e668db1 100644
--- a/reactor-core/src/main/java/reactor/core/publisher/Mono.java
+++ b/reactor-core/src/main/java/reactor/core/publisher/Mono.java
@@ -27,6 +27,7 @@ import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
 import java.util.function.BiPredicate;
@@ -133,9 +134,9 @@ public abstract class Mono<T> implements Publisher<T> {
 	 *             }
 	 *         }
 	 *     };
-	 *     
+	 *
 	 *     client.addListener(listener);
-	 *     
+	 *
 	 *     sink.onDispose(() -&gt; client.removeListener(listener));
 	 * });
 	 * </code></pre>
@@ -2675,7 +2676,7 @@ public abstract class Mono<T> implements Publisher<T> {
 
 	/**
 	 * Hides the identity of this {@link Mono} instance.
-	 * 
+	 *
 	 * <p>The main purpose of this operator is to prevent certain identity-based
 	 * optimizations from happening, mostly for diagnostic purposes.
 	 *
@@ -2684,7 +2685,7 @@ public abstract class Mono<T> implements Publisher<T> {
 	public final Mono<T> hide() {
 	    return onAssembly(new MonoHide<>(this));
 	}
-	
+
 	/**
 	 * Ignores onNext signal (dropping it) and only propagates termination events.
 	 *
@@ -3416,23 +3417,16 @@ public abstract class Mono<T> implements Publisher<T> {
 	 * as long as the companion {@link Publisher} produces an onNext signal and the maximum number of repeats isn't exceeded.
 	 */
 	public final Mono<T> repeatWhenEmpty(int maxRepeat, Function<Flux<Long>, ? extends Publisher<?>> repeatFactory) {
-		return Mono.defer(() -> {
-			Flux<Long> iterations;
-
-			if(maxRepeat == Integer.MAX_VALUE) {
-				iterations = Flux.fromStream(LongStream.range(0, Long.MAX_VALUE).boxed());
+		return Mono.defer(() -> this.repeatWhen(o -> {
+			if (maxRepeat == Integer.MAX_VALUE) {
+				return repeatFactory.apply(o.index().map(Tuple2::getT1));
 			}
 			else {
-				iterations = Flux
-					.range(0, maxRepeat)
-					.map(Integer::longValue)
-					.concatWith(Flux.error(new IllegalStateException("Exceeded maximum number of repeats"), true));
+				return repeatFactory.apply(o.index().map(Tuple2::getT1)
+						.take(maxRepeat)
+						.concatWith(Flux.error(new IllegalStateException("Exceeded maximum number of repeats"), true)));
 			}
-
-			return this.repeatWhen(o -> repeatFactory.apply(o
-						.zipWith(iterations, 1, (c, i) -> i)))
-			           .next();
-		});
+		}).next());
 	}
 
 
diff --git a/reactor-core/src/test/java/reactor/core/publisher/MonoRepeatWhenEmptyTest.java b/reactor-core/src/test/java/reactor/core/publisher/MonoRepeatWhenEmptyTest.java
index 3bff20b94..e000821f5 100644
--- a/reactor-core/src/test/java/reactor/core/publisher/MonoRepeatWhenEmptyTest.java
+++ b/reactor-core/src/test/java/reactor/core/publisher/MonoRepeatWhenEmptyTest.java
@@ -18,8 +18,10 @@ package reactor.core.publisher;
 
 import org.junit.Assert;
 import org.junit.Test;
+import reactor.test.StepVerifier;
 import reactor.test.subscriber.AssertSubscriber;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -91,4 +93,13 @@ public class MonoRepeatWhenEmptyTest {
         Assert.assertEquals(Arrays.asList(0L, 1L), iterations);
     }
 
+    @Test(timeout = 1000L)
+    public void gh2196_discardHandlerHang() {
+        StepVerifier.create(Mono.empty()
+                .repeatWhenEmpty(f -> f.next())
+                .doOnDiscard(Object.class, System.out::println))
+                .thenAwait()
+                .thenCancel()
+                .verify();
+    }
 }