Added Flux.index()

Author: sinwe

src/main/scala/reactor/core/scala/publisher/Flux.scala

@@ -1779,6 +1779,17 @@ class Flux[T] private[publisher](private[publisher] val jFlux: JFlux[T]) extends
   //  TODO: How to test???
   final def hide() = Flux(jFlux.hide())
 
+  /**
+    * Keep information about the order in which source values were received by
+    * indexing them with a 0-based incrementing long, returning a [[Flux]]
+    * of [[scala.Tuple2[index, value]]
+    *
+    * @return an indexed [[Flux]] with each source value combined with its 0-based index.
+    */
+  final def index(): Flux[(Long, T)] = Flux(jFlux.index()).map(tupleTwo2ScalaTuple2) map {
+    case (jLong: JLong, t: T) => (Long2long(jLong), t)
+  }
+
   /**
     * Ignores onNext signals (dropping them) and only reacts on termination.
     *

src/test/scala/reactor/core/scala/publisher/FluxTest.scala

@@ -309,6 +309,14 @@ class FluxTest extends FreeSpec with Matchers with TableDrivenPropertyChecks {
           .verifyComplete()
       }
     }
+
+    ".index should return tuple with the index" in {
+      val flux = Flux.just("a", "b", "c").index()
+      StepVerifier.create(flux)
+        .expectNext((0l, "a"), (1l, "b"), (2l, "c"))
+        .verifyComplete()
+    }
+
     ".interval" - {
       "without delay should produce flux of Long starting from 0 every provided timespan immediately" in {
         StepVerifier.withVirtualTime(() => Flux.interval(1 second).take(5))