Minor improvements

Author: BalmungSan

core/shared/src/main/scala/fs2/interop/flow/StreamSubscriber.scala

@@ -50,49 +50,49 @@ private[flow] final class StreamSubscriber[F[_], A] private (
   // Subscriber API.
 
   /** Receives a subscription from the upstream reactive-streams system. */
-  override def onSubscribe(subscription: Subscription): Unit = {
+  override final def onSubscribe(subscription: Subscription): Unit = {
     requireNonNull(
       subscription,
       "The subscription provided to onSubscribe must not be null"
     )
-    nextState(Subscribe(subscription))
+    nextState(input = Subscribe(subscription))
   }
 
   /** Receives the next record from the upstream reactive-streams system. */
-  override def onNext(a: A): Unit = {
+  override final def onNext(a: A): Unit = {
     requireNonNull(
       a,
       "The element provided to onNext must not be null"
     )
-    nextState(Next(a))
+    nextState(input = Next(a))
   }
 
   /** Called by the upstream reactive-streams system when it fails. */
-  override def onError(ex: Throwable): Unit = {
+  override final def onError(ex: Throwable): Unit = {
     requireNonNull(
       ex,
       "The throwable provided to onError must not be null"
     )
-    nextState(Error(ex))
+    nextState(input = Error(ex))
   }
 
   /** Called by the upstream reactive-streams system when it has finished sending records. */
-  override def onComplete(): Unit =
-    nextState(Complete(canceled = false))
+  override final def onComplete(): Unit =
+    nextState(input = Complete(canceled = false))
 
   // Interop API.
 
   /** Creates a downstream [[Stream]] from this [[Subscriber]]. */
   private[flow] def stream(subscribe: F[Unit]): Stream[F, A] = {
     // Called when downstream has finished consuming records.
     val finalize =
-      F.delay(nextState(Complete(canceled = true)))
+      F.delay(nextState(input = Complete(canceled = true)))
 
     // Producer for downstream.
     val dequeue1 =
       F.async[Option[Chunk[Any]]] { cb =>
         F.delay {
-          nextState(Dequeue(cb))
+          nextState(input = Dequeue(cb))
 
           Some(finalize)
         }
@@ -112,8 +112,8 @@ private[flow] final class StreamSubscriber[F[_], A] private (
   private def run(block: => Unit): () => Unit = () => block
 
   /** Runs a single step of the state machine. */
-  private def step(in: Input): State => (State, () => Unit) =
-    in match {
+  private def step(input: Input): State => (State, () => Unit) =
+    input match {
       case Subscribe(s) => {
         case Uninitialized(None) =>
           Idle(s) -> noop
@@ -263,9 +263,9 @@ private[flow] final class StreamSubscriber[F[_], A] private (
     * + `Error` & `Dequeue`: No matter the order in which they are processed, we will complete the callback with the error.
     * + cancellation & any other thing: Worst case, we will lose some data that we not longer care about; and eventually reach `Terminal`.
     */
-  private def nextState(in: Input): Unit = {
+  private def nextState(input: Input): Unit = {
     val (_, effect) = currentState.updateAndGet { case (state, _) =>
-      step(in)(state)
+      step(input)(state)
     }
     // Only run the effect after the state update took place.
     effect()

core/shared/src/main/scala/fs2/interop/flow/StreamSubscription.scala

@@ -33,7 +33,7 @@ import java.util.concurrent.atomic.{AtomicLong, AtomicReference}
 
 /** Implementation of a [[Subscription]].
   *
-  * This is used by the [[StreamUnicastPublisher]] to send elements from a [[Stream]] to a downstream reactive-streams system.
+  * This is used by the [[StreamPublisher]] to send elements from a [[Stream]] to a downstream reactive-streams system.
   *
   * @see [[https://github.com/reactive-streams/reactive-streams-jvm#3-subscription-code]]
   */
@@ -58,7 +58,7 @@ private[flow] final class StreamSubscription[F[_], A] private (
     sub.onComplete()
   }
 
-  private[flow] def run: F[Unit] = {
+  val run: F[Unit] = {
     val subscriptionPipe: Pipe[F, A, A] = in => {
       def go(s: Stream[F, A]): Pull[F, A, Unit] =
         Pull.eval(F.delay(requests.get())).flatMap { n =>
@@ -133,14 +133,14 @@ private[flow] final class StreamSubscription[F[_], A] private (
   // then the request must be a NOOP.
   // See https://github.com/zainab-ali/fs2-reactive-streams/issues/29
   // and https://github.com/zainab-ali/fs2-reactive-streams/issues/46
-  override def cancel(): Unit = {
+  override final def cancel(): Unit = {
     val cancelCB = canceled.getAndSet(null)
     if (cancelCB ne null) {
       cancelCB.apply()
     }
   }
 
-  override def request(n: Long): Unit =
+  override final def request(n: Long): Unit =
     // First, confirm we are not yet cancelled.
     if (canceled.get() ne null) {
       // Second, ensure we were requested a positive number of elements.

core/shared/src/main/scala/fs2/interop/flow/package.scala

@@ -102,7 +102,7 @@ package object flow {
         subscriber.stream(subscribe(subscriber))
       }
 
-  /** Creates a [[Stream]] from an [[Publisher]].
+  /** Creates a [[Stream]] from a [[Publisher]].
     *
     * @example {{{
     * scala> import cats.effect.IO
@@ -118,7 +118,7 @@ package object flow {
     * res0: Stream[IO, Int] = Stream(..)
     * }}}
     *
-    * @note The publisher will not receive a subscriber until the stream is run.
+    * @note The [[Publisher]] will not receive a [[Subscriber]] until the stream is run.
     *
     * @see the `toStream` extension method added to `Publisher`
     *

core/shared/src/main/scala/fs2/interop/flow/syntax.scala

@@ -32,7 +32,7 @@ import java.util.concurrent.Flow.{Publisher, Subscriber}
 object syntax {
   implicit final class PublisherOps[A](private val publisher: Publisher[A]) extends AnyVal {
 
-    /** Creates a [[Stream]] from an [[Publisher]].
+    /** Creates a [[Stream]] from a [[Publisher]].
       *
       * @example {{{
       * scala> import cats.effect.IO
@@ -49,6 +49,8 @@ object syntax {
       * res0: Stream[IO, Int] = Stream(..)
       * }}}
       *
+      * @note The [[Publisher]] will not receive a [[Subscriber]] until the stream is run.
+      *
       * @param chunkSize setup the number of elements asked each time from the [[Publisher]].
       *                  A high number may be useful if the publisher is triggering from IO,
       *                  like requesting elements from a database.