Added Flux.index(indexMapper)

Author: sinwe

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

@@ -1790,6 +1790,24 @@ class Flux[T] private[publisher](private[publisher] val jFlux: JFlux[T]) extends
     case (jLong: JLong, t: T) => (Long2long(jLong), t)
   }
 
+  /**
+    * Keep information about the order in which source values were received by
+    * indexing them internally with a 0-based incrementing long then combining this
+    * information with the source value into a `I` using the provided [[Function2]] ,
+    * returning a [[Flux[I]]].
+    * <p>
+    * Typical usage would be to produce a [[scala.Tuple2]] similar to [[Flux.index()]], but
+    * 1-based instead of 0-based:
+    * <p>
+    * `index((i, v) => (i+1, v))`
+    *
+    * @param indexMapper the [[Function2]] to use to combine elements and their index.
+    * @return an indexed [[Flux]] with each source value combined with its computed index.
+    */
+  final def index[I](indexMapper: (Long, T) => I) = Flux(jFlux.index[I](new BiFunction[JLong, T, I] {
+    override def apply(t: JLong, u: T) = indexMapper(Long2long(t), u)
+  }))
+
   /**
     * Ignores onNext signals (dropping them) and only reacts on termination.
     *

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

@@ -310,11 +310,19 @@ class FluxTest extends FreeSpec with Matchers with TableDrivenPropertyChecks {
       }
     }
 
-    ".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()
+    ".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()
+      }
+      "with index mapper should return the mapped value" in {
+        val flux = Flux.just("a", "b", "c").index((i, v) => s"$i-$v")
+        StepVerifier.create(flux)
+          .expectNext("0-a", "1-b", "2-c")
+          .verifyComplete()
+      }
     }
 
     ".interval" - {