Added SFlux.name

This will also fix bug #18

Author: sinwe

src/main/scala/reactor/core/scala/Scannable.scala

@@ -91,8 +91,16 @@ trait Scannable {
 }
 
 object Scannable {
-  def from(any: AnyRef): Scannable = new Scannable {
-    override def jScannable: JScannable = JScannable.from(any)
+//  def from(any: AnyRef): Scannable = new Scannable {
+//    override def jScannable: JScannable = JScannable.from(any)
+//  }
+
+  def from(any: Option[AnyRef]): Scannable = {
+    any match {
+      case None => JScannable.from(None.orNull)
+      case Some(s: Scannable) => s.jScannable
+      case _ => JScannable.from(new Object())
+    }
   }
 
   implicit def JScannable2Scannable(js: JScannable): Scannable = new Scannable {

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

@@ -7,9 +7,10 @@ import java.util.function.{BiFunction, Function, Supplier}
 import java.util.{Collection => JCollection, List => JList, Map => JMap}
 
 import org.reactivestreams.{Publisher, Subscriber, Subscription}
-import reactor.core.Disposable
+import reactor.core.{Disposable, Scannable => JScannable}
 import reactor.core.publisher.FluxSink.OverflowStrategy
 import reactor.core.publisher.{FluxSink, Signal, SignalType, SynchronousSink, Flux => JFlux, GroupedFlux => JGroupedFlux}
+import reactor.core.scala.Scannable
 import reactor.core.scala.publisher.PimpMyPublisher._
 import reactor.core.scheduler.{Scheduler, Schedulers}
 import reactor.util.concurrent.Queues.{SMALL_BUFFER_SIZE, XS_BUFFER_SIZE}
@@ -223,6 +224,8 @@ trait SFlux[T] extends SFluxLike[T, SFlux] with MapablePublisher[T] {
 
   final def mergeWith(other: Publisher[_ <: T]): SFlux[T] = coreFlux.mergeWith(other)
 
+  final def name(name: String): SFlux[T] = coreFlux.name(name)
+
   final def nonEmpty: SMono[Boolean] = hasElements
 
   final def onErrorMap(mapper: Throwable => _ <: Throwable): SFlux[T] = coreFlux.onErrorMap(mapper)
@@ -348,6 +351,10 @@ object SFlux {
     new ReactiveSFlux[O](JFlux.zip[I, O](combinator, prefetch, sources: _*))
 }
 
-private[publisher] class ReactiveSFlux[T](publisher: Publisher[T]) extends SFlux[T] {
-  override private[publisher] def coreFlux: JFlux[T] = JFlux.from(publisher)
+private[publisher] class ReactiveSFlux[T](publisher: Publisher[T]) extends SFlux[T] with Scannable {
+  override private[publisher] val coreFlux: JFlux[T] = JFlux.from(publisher)
+
+  override def scanUnsafe(key: JScannable.Attr[_]): Option[AnyRef] = Option(jScannable.scanUnsafe(key))
+
+  override val jScannable: JScannable = JScannable.from(coreFlux)
 }

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

@@ -10,7 +10,8 @@ import java.util.function.Predicate
 import org.reactivestreams.Subscription
 import org.scalatest.prop.TableDrivenPropertyChecks
 import org.scalatest.{FreeSpec, Matchers}
-import reactor.core.publisher.{Flux, _}
+import reactor.core.publisher._
+import reactor.core.scala.Scannable
 import reactor.core.scheduler.Schedulers
 import reactor.test.StepVerifier
 import reactor.test.scheduler.VirtualTimeScheduler
@@ -1363,6 +1364,13 @@ class SFluxTest extends FreeSpec with Matchers with TableDrivenPropertyChecks {
       }
     }
 
+    ".name should call the underlying Flux.name method" in {
+      val name = "one two three four"
+      val flux = SFlux.just(1, 2, 3, 4).name(name)
+      val scannable: Scannable = Scannable.from(Option(flux))
+      scannable.name shouldBe name
+    }
+
     ".nonEmpty should return true if this flux has at least one element" in {
       StepVerifier.create(SFlux.just(1, 2, 3).nonEmpty)
         .expectNext(true)