Breaking changes:

- Renamed ConnectableFlux => ConnectableSFlux

Added SFlux.publish

Author: sinwe

src/main/scala/reactor/core/scala/publisher/ConnectableFlux.scala renamed to src/main/scala/reactor/core/scala/publisher/ConnectableSFlux.scala

@@ -1,26 +1,26 @@
 package reactor.core.scala.publisher
 
-import reactor.core.publisher.{ConnectableFlux => JConnectableFlux}
 import reactor.core.Disposable
+import reactor.core.publisher.ConnectableFlux
 import reactor.core.scheduler.Scheduler
 
 import scala.concurrent.duration.Duration
 
-class ConnectableFlux[T]private (private val jConnectableFlux: JConnectableFlux[T]) extends SFlux[T] {
+class ConnectableSFlux[T]private(private val connectableFlux: ConnectableFlux[T]) extends SFlux[T] {
 
   /**
-    * Connects this [[ConnectableFlux]] to the upstream source when the first [[org.reactivestreams.Subscriber]]
+    * Connects this [[ConnectableSFlux]] to the upstream source when the first [[org.reactivestreams.Subscriber]]
     * subscribes.
     *
     * <p>
     * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/v3.0.6.RELEASE/src/docs/marble/autoconnect.png" alt="">
     *
     * @return a [[SFlux]] that connects to the upstream source when the first [[org.reactivestreams.Subscriber]] subscribes
     */
-  final def autoConnect(): SFlux[T] = SFlux.fromPublisher(jConnectableFlux.autoConnect())
+  final def autoConnect(): SFlux[T] = SFlux.fromPublisher(connectableFlux.autoConnect())
 
   /**
-    * Connects this [[ConnectableFlux]] to the upstream source when the specified amount of
+    * Connects this [[ConnectableSFlux]] to the upstream source when the specified amount of
     * [[org.reactivestreams.Subscriber]] subscribes.
     * <p>
     * Subscribing and immediately unsubscribing Subscribers also contribute the the subscription count
@@ -32,10 +32,10 @@ class ConnectableFlux[T]private (private val jConnectableFlux: JConnectableFlux[
     * @param minSubscribers the minimum number of subscribers
     * @return a [[SFlux]] that connects to the upstream source when the given amount of Subscribers subscribe
     */
-  final def autoConnect(minSubscribers: Int): SFlux[T] = SFlux.fromPublisher(jConnectableFlux.autoConnect(minSubscribers))
+  final def autoConnect(minSubscribers: Int): SFlux[T] = SFlux.fromPublisher(connectableFlux.autoConnect(minSubscribers))
 
   /**
-    * Connects this [[ConnectableFlux]] to the upstream source when the specified amount of
+    * Connects this [[ConnectableSFlux]] to the upstream source when the specified amount of
     * [[org.reactivestreams.Subscriber]] subscribes and calls the supplied consumer with a runnable that allows disconnecting.
     *
     * @param minSubscribers the minimum number of subscribers
@@ -45,18 +45,18 @@ class ConnectableFlux[T]private (private val jConnectableFlux: JConnectableFlux[
     *                                                                  <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/v3.0.6.RELEASE/src/docs/marble/autoconnect.png" alt="">
     * @return a { @link Flux} that connects to the upstream source when the given amount of subscribers subscribed
     */
-  final def autoConnect(minSubscribers: Int, cancelSupport: Disposable => Unit): SFlux[T] = SFlux.fromPublisher(jConnectableFlux.autoConnect(minSubscribers, cancelSupport))
+  final def autoConnect(minSubscribers: Int, cancelSupport: Disposable => Unit): SFlux[T] = SFlux.fromPublisher(connectableFlux.autoConnect(minSubscribers, cancelSupport))
 
   /**
-    * Connect this [[ConnectableFlux]] to its source and return a [[Runnable]] that
+    * Connect this [[ConnectableSFlux]] to its source and return a [[Runnable]] that
     * can be used for disconnecting.
     *
     * @return the [[Disposable]] that allows disconnecting the connection after.
     */
-  final def connect(): Disposable = jConnectableFlux.connect()
+  final def connect(): Disposable = connectableFlux.connect()
 
   /**
-    * Connects this [[ConnectableFlux]] to its source and sends a [[Disposable]] to a callback that
+    * Connects this [[ConnectableSFlux]] to its source and sends a [[Disposable]] to a callback that
     * can be used for disconnecting.
     *
     * <p>The call should be idempotent in respect of connecting the first
@@ -66,7 +66,7 @@ class ConnectableFlux[T]private (private val jConnectableFlux: JConnectableFlux[
     * @param cancelSupport the callback is called with a Cancellation instance that can
     *                      be called to disconnect the source, even synchronously.
     */
-  final def connect(cancelSupport: Disposable => Unit): Unit = jConnectableFlux.connect(cancelSupport)
+  final def connect(cancelSupport: Disposable => Unit): Unit = connectableFlux.connect(cancelSupport)
 
   /**
     * Connects to the upstream source when the first [[org.reactivestreams.Subscriber]] subscribes and disconnects
@@ -77,7 +77,7 @@ class ConnectableFlux[T]private (private val jConnectableFlux: JConnectableFlux[
     *
     * @return a reference counting [[SFlux]]
     */
-  final def refCount(): SFlux[T] = SFlux.fromPublisher(jConnectableFlux.refCount())
+  final def refCount(): SFlux[T] = SFlux.fromPublisher(connectableFlux.refCount())
 
   /**
     * Connects to the upstream source when the given number of [[org.reactivestreams.Subscriber]] subscribes and disconnects
@@ -89,7 +89,7 @@ class ConnectableFlux[T]private (private val jConnectableFlux: JConnectableFlux[
     * @param minSubscribers the number of subscribers expected to subscribe before connection
     * @return a reference counting [[SFlux]]
     */
-  final def refCount(minSubscribers: Int): SFlux[T] = SFlux.fromPublisher(jConnectableFlux.refCount(minSubscribers))
+  final def refCount(minSubscribers: Int): SFlux[T] = SFlux.fromPublisher(connectableFlux.refCount(minSubscribers))
 
   /**
     * Connects to the upstream source when the given number of [[org.reactivestreams.Subscriber]] subscribes.
@@ -106,7 +106,7 @@ class ConnectableFlux[T]private (private val jConnectableFlux: JConnectableFlux[
     *                                   disconnecting when all subscribers have cancelled.
     * @return a reference counting [[SFlux]] with a grace period for disconnection
     */
-  final def refCount(minSubscribers: Int, gracePeriod: Duration): SFlux[T] = SFlux.fromPublisher(jConnectableFlux.refCount(minSubscribers, gracePeriod))
+  final def refCount(minSubscribers: Int, gracePeriod: Duration): SFlux[T] = SFlux.fromPublisher(connectableFlux.refCount(minSubscribers, gracePeriod))
 
   /**
     * Connects to the upstream source when the given number of [[org.reactivestreams.Subscriber]] subscribes.
@@ -124,11 +124,11 @@ class ConnectableFlux[T]private (private val jConnectableFlux: JConnectableFlux[
     * @param scheduler the [[Scheduler]] on which to run timeouts
     * @return a reference counting [[SFlux]] with a grace period for disconnection
     */
-  final def refCount(minSubscribers: Int, gracePeriod: Duration, scheduler: Scheduler): SFlux[T] = SFlux.fromPublisher(jConnectableFlux.refCount(minSubscribers, gracePeriod, scheduler))
+  final def refCount(minSubscribers: Int, gracePeriod: Duration, scheduler: Scheduler): SFlux[T] = SFlux.fromPublisher(connectableFlux.refCount(minSubscribers, gracePeriod, scheduler))
 
-  override private[publisher] def coreFlux: JConnectableFlux[T] = jConnectableFlux
+  override private[publisher] def coreFlux: ConnectableFlux[T] = connectableFlux
 }
 
-object ConnectableFlux {
-  def apply[T](jConnectableFlux: JConnectableFlux[T]) = new ConnectableFlux[T](jConnectableFlux)
+object ConnectableSFlux {
+  def apply[T](jConnectableFlux: ConnectableFlux[T]) = new ConnectableSFlux[T](jConnectableFlux)
 }

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

@@ -600,6 +600,32 @@ trait SFlux[T] extends SFluxLike[T, SFlux] with MapablePublisher[T] with ScalaCo
     */
   final def parallel(parallelism: Int = Runtime.getRuntime.availableProcessors(), prefetch: Int = Queues.SMALL_BUFFER_SIZE) = SParallelFlux(coreFlux.parallel(parallelism, prefetch))
 
+  /**
+    * Prepare a [[ConnectableSFlux]] which shares this [[SFlux]] sequence and
+    * dispatches values to subscribers in a backpressure-aware manner. This will
+    * effectively turn any type of sequence into a hot sequence.
+    * <p>
+    * Backpressure will be coordinated on [[org.reactivestreams.Subscription.request]] and if any
+    * [[Subscriber]] is missing demand (requested = 0), multicast will pause
+    * pushing/pulling.
+    * <p>
+    * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/master/reactor-core/src/main/java/reactor/core/publisher/doc-files/marbles/publish.svg" alt="">
+    *
+    * @param prefetch bounded requested demand
+    * @return a new [[ConnectableSFlux]]
+    */
+  final def publish(prefetch: Int = Queues.SMALL_BUFFER_SIZE): ConnectableSFlux[T] = coreFlux.publish(prefetch).asScala
+
+  /**
+    * Prepare a [[SMono]] which shares this [[SFlux]] sequence and dispatches the
+    * first observed item to subscribers in a backpressure-aware manner.
+    * This will effectively turn any type of sequence into a hot sequence when the first
+    * [[Subscriber]] subscribes.
+    * <p>
+    * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/master/reactor-core/src/main/java/reactor/core/publisher/doc-files/marbles/publishNext.svg" alt="">
+    *
+    * @return a new [[SMono]]
+    */
   final def publishNext(): SMono[T] = SMono.fromPublisher(coreFlux.publishNext())
 
   final def reduce(aggregator: (T, T) => T): SMono[T] = coreFlux.reduce(aggregator).asScala

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

@@ -1,6 +1,6 @@
 package reactor.core.scala.publisher
 
-import reactor.core.publisher.{Flux => JFlux, Mono => JMono}
+import reactor.core.publisher.{ConnectableFlux, Flux => JFlux, Mono => JMono}
 
 import scala.language.implicitConversions
 
@@ -12,6 +12,10 @@ trait ScalaConverters {
   implicit class PimpJFlux[T](jFlux: JFlux[T]) {
     def asScala: SFlux[T] = new ReactiveSFlux[T](jFlux)
   }
+
+  implicit class PimpConnectableFlux[T](connectableFlux: ConnectableFlux[T]) {
+    def asScala: ConnectableSFlux[T] = ConnectableSFlux(connectableFlux)
+  }
 }
 
 object ScalaConverters extends ScalaConverters

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

@@ -10,9 +10,11 @@ import java.util.function.{Consumer, Predicate}
 
 import org.mockito.{ArgumentMatchersSugar, IdiomaticMockito}
 import org.reactivestreams.Subscription
+import org.scalatest.concurrent.{Eventually, PatienceConfiguration}
 import org.scalatest.freespec.AnyFreeSpec
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.prop.TableDrivenPropertyChecks
+import org.scalatest.time.{Seconds, Span}
 import reactor.core.publisher.BufferOverflowStrategy.DROP_LATEST
 import reactor.core.publisher.{Flux => JFlux, _}
 import reactor.core.scala.Scannable
@@ -33,7 +35,11 @@ import scala.math.Ordering.IntOrdering
 import scala.math.ScalaNumber
 import scala.util.{Failure, Try}
 
-class SFluxTest extends AnyFreeSpec with Matchers with TableDrivenPropertyChecks with TestSupport with IdiomaticMockito with ArgumentMatchersSugar {
+class SFluxTest extends AnyFreeSpec with Matchers with TableDrivenPropertyChecks with TestSupport with IdiomaticMockito with ArgumentMatchersSugar
+  with PatienceConfiguration with Eventually {
+
+  implicit override def patienceConfig: PatienceConfig = PatienceConfig(Span(2, Seconds))
+
   "SFlux" - {
     ".apply should return a proper SFlux when provided a Publisher" in {
       StepVerifier.create(SFlux(JFlux.just(1,2,3)))
@@ -1680,6 +1686,20 @@ class SFluxTest extends AnyFreeSpec with Matchers with TableDrivenPropertyChecks
         .verifyComplete()
     }
 
+    ".publish" - {
+      "without transformer should produce connectable flux" in {
+        val buffer = mutable.ListBuffer.empty[Int]
+        val base = SFlux.just(1, 2, 3, 4, 5).delayElements(1 second).publish().autoConnect()
+        base.subscribe()
+        Thread.sleep(1000)
+        base.subscribe(i => buffer += i)
+        eventually {
+          buffer.size should be >= 1
+          buffer should not contain (1)
+        }
+      }
+    }
+
     ".publishNext should make this flux a hot mono" in {
       StepVerifier.create(SFlux.just(1, 2, 3).publishNext())
         .expectNext(1)