fix #1056 doOnEach correctly passes itself to onSubscribe

This commit ensures that the FluxDoOnEach DoOnEachSubscriber correctly
passes itself to actual.onSubscribe rather than its own upstream
Subscription, thus ensuring that no short-circuiting fusion is
established between the downstream and upstream.

reactor-core/src/main/java/reactor/core/publisher/FluxDoOnEach.java

@@ -80,7 +80,7 @@ public void cancel() {
 		@Override
 		public void onSubscribe(Subscription s) {
 			this.s = s;
-			actual.onSubscribe(s);
+			actual.onSubscribe(this);
 		}
 
 		@Override

reactor-core/src/test/java/reactor/core/publisher/FluxDoOnEachTest.java

@@ -19,6 +19,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.atomic.LongAdder;
 
@@ -29,6 +30,7 @@
 import reactor.core.CoreSubscriber;
 import reactor.core.Exceptions;
 import reactor.core.Scannable;
+import reactor.core.scheduler.Schedulers;
 import reactor.test.StepVerifier;
 import reactor.test.StepVerifierOptions;
 import reactor.test.subscriber.AssertSubscriber;
@@ -76,6 +78,26 @@ else if (s.isOnComplete()) {
 		Assert.assertEquals(2, state.intValue());
 	}
 
+
+	//see https://github.com/reactor/reactor-core/issues/1056
+	@Test
+	public void fusion() {
+		AtomicInteger invocationCount = new AtomicInteger();
+
+		Flux<String> sourceWithFusionAsync = Flux.just("foo")
+		                                         .publishOn(Schedulers.elastic())
+		                                         .flatMap(v -> Flux.just("flatMap_" + v)
+		                                                           .doOnEach(sig -> invocationCount.incrementAndGet())
+		                                         );
+
+		StepVerifier.create(sourceWithFusionAsync)
+		            .expectNoFusionSupport()
+		            .expectNext("flatMap_foo")
+		            .verifyComplete();
+
+		assertThat(invocationCount).as("doOnEach invoked").hasValue(2);
+	}
+
 	@Test
 	public void error() {
 		AssertSubscriber<Integer> ts = AssertSubscriber.create();