Change overflow-strategy onOverflow handling

build.sbt

@@ -106,6 +106,9 @@ lazy val sharedSettings = warnUnusedImport ++ Seq(
     Resolver.sonatypeRepo("releases")
   ),
 
+  // https://github.com/sbt/sbt/issues/2654
+  incOptions := incOptions.value.withLogRecompileOnMacro(false),
+
   // -- Settings meant for deployment on oss.sonatype.org
 
   useGpg := true,
@@ -260,9 +263,7 @@ lazy val typesJS = project.in(file("monix-types/js"))
   .settings(scalaJSSettings)
 
 lazy val executionCommon = crossVersionSharedSources ++ Seq(
-  name := "monix-execution",
-  // https://github.com/sbt/sbt/issues/2654
-  incOptions := incOptions.value.withLogRecompileOnMacro(false)
+  name := "monix-execution"
 )
 
 lazy val executionJVM = project.in(file("monix-execution/jvm"))

monix-reactive/js/src/main/scala/monix/reactive/observers/buffers/SyncBufferedSubscriber.scala

@@ -30,7 +30,7 @@ import scala.util.control.NonFatal
   * [[monix.reactive.OverflowStrategy.DropNew DropNew]] overflow strategy.
   */
 private[buffers] final class SyncBufferedSubscriber[-T] private
-  (underlying: Subscriber[T], buffer: EvictingQueue[T], onOverflow: Long => T = null)
+  (underlying: Subscriber[T], buffer: EvictingQueue[T], onOverflow: Long => Option[T] = null)
   extends BufferedSubscriber[T] with Subscriber.Sync[T] {
 
   implicit val scheduler = underlying.scheduler
@@ -107,11 +107,14 @@ private[buffers] final class SyncBufferedSubscriber[-T] private
       val nextEvent =
         if (eventsDropped > 0 && onOverflow != null) {
           try {
-            val message = onOverflow(eventsDropped).asInstanceOf[AnyRef]
-            eventsDropped = 0
-            message
-          }
-          catch {
+            onOverflow(eventsDropped) match {
+              case Some(message) =>
+                eventsDropped = 0
+                message.asInstanceOf[AnyRef]
+              case None =>
+                buffer.poll()
+            }
+          } catch {
             case NonFatal(ex) =>
               errorThrown = ex
               upstreamIsComplete = true
@@ -212,7 +215,7 @@ private[monix] object SyncBufferedSubscriber {
     * for the [[monix.reactive.OverflowStrategy.DropNew DropNew]]
     * overflow strategy.
     */
-  def dropNewAndSignal[T](underlying: Subscriber[T], bufferSize: Int, onOverflow: Long => T): Subscriber.Sync[T] = {
+  def dropNewAndSignal[T](underlying: Subscriber[T], bufferSize: Int, onOverflow: Long => Option[T]): Subscriber.Sync[T] = {
     require(bufferSize > 1,
       "bufferSize must be a strictly positive number, bigger than 1")
 
@@ -239,7 +242,7 @@ private[monix] object SyncBufferedSubscriber {
     * overflow strategy, with signaling of the number of events that
     * were dropped.
     */
-  def dropOldAndSignal[T](underlying: Subscriber[T], bufferSize: Int, onOverflow: Long => T): Subscriber.Sync[T] = {
+  def dropOldAndSignal[T](underlying: Subscriber[T], bufferSize: Int, onOverflow: Long => Option[T]): Subscriber.Sync[T] = {
     require(bufferSize > 1,
       "bufferSize must be a strictly positive number, bigger than 1")
 
@@ -266,7 +269,7 @@ private[monix] object SyncBufferedSubscriber {
     * overflow strategy, with signaling of the number of events that
     * were dropped.
     */
-  def clearBufferAndSignal[T](underlying: Subscriber[T], bufferSize: Int, onOverflow: Long => T): Subscriber.Sync[T] = {
+  def clearBufferAndSignal[T](underlying: Subscriber[T], bufferSize: Int, onOverflow: Long => Option[T]): Subscriber.Sync[T] = {
     require(bufferSize > 1,
       "bufferSize must be a strictly positive number, bigger than 1")
 

monix-reactive/jvm/src/main/scala/monix/reactive/observers/buffers/DropNewBufferedSubscriber.scala

@@ -27,13 +27,12 @@ import scala.annotation.tailrec
 import scala.concurrent.Future
 import scala.util.control.NonFatal
 
-/**
- * A high-performance and non-blocking [[BufferedSubscriber]] implementation
- * for the [[monix.reactive.OverflowStrategy.DropNew DropNew]]
- * overflow strategy.
- */
+/** A high-performance and non-blocking [[BufferedSubscriber]] implementation
+  * for the [[monix.reactive.OverflowStrategy.DropNew DropNew]]
+  * overflow strategy.
+  */
 private[buffers] final class DropNewBufferedSubscriber[-T] private
-  (underlying: Subscriber[T], bufferSize: Int, onOverflow: Long => T = null)
+  (underlying: Subscriber[T], bufferSize: Int, onOverflow: Long => Option[T] = null)
   extends BufferedSubscriber[T] with Subscriber.Sync[T] { self =>
 
   require(bufferSize > 0, "bufferSize must be a strictly positive number")
@@ -75,12 +74,14 @@ private[buffers] final class DropNewBufferedSubscriber[-T] private
           // composing the overflow message; we've got to do error handling
           // because this is a user supplied function
           val shouldContinue = try {
-            val message = onOverflow(state.eventsDropped)
-            queue.offer(message)
-            pushToConsumer(update)
+            onOverflow(state.eventsDropped) match {
+              case None => ()
+              case Some(message) =>
+                queue.offer(message)
+                pushToConsumer(update)
+            }
             true
-          }
-          catch {
+          } catch {
             case NonFatal(ex) =>
               onError(ex)
               false
@@ -147,12 +148,12 @@ private[buffers] final class DropNewBufferedSubscriber[-T] private
           // composing the overflow message; we've got to do error handling
           // because this is a user supplied function
           val ex = try {
-            val message = onOverflow(state.eventsDropped)
-            queue.offer(message)
-            pushToConsumer(state)
+            for (message <- onOverflow(state.eventsDropped)) {
+              queue.offer(message)
+              pushToConsumer(state)
+            }
             null
-          }
-          catch {
+          } catch {
             case NonFatal(ref) => ref
           }
 
@@ -291,21 +292,19 @@ private[buffers] final class DropNewBufferedSubscriber[-T] private
 }
 
 private[monix] object DropNewBufferedSubscriber {
-  /**
-   * Returns an instance of a [[DropNewBufferedSubscriber]]
-   * for the [[monix.reactive.OverflowStrategy.DropNew DropNew]]
-   * overflowStrategy.
-   */
+  /** Returns an instance of a [[DropNewBufferedSubscriber]]
+    * for the [[monix.reactive.OverflowStrategy.DropNew DropNew]]
+    * overflowStrategy.
+    */
   def simple[T](underlying: Subscriber[T], bufferSize: Int): Subscriber.Sync[T] = {
     new DropNewBufferedSubscriber[T](underlying, bufferSize, null)
   }
 
-  /**
-   * Returns an instance of a [[DropNewBufferedSubscriber]]
-   * for the [[monix.reactive.OverflowStrategy.DropNew DropNew]]
-   * overflowStrategy.
-   */
-  def withSignal[T](underlying: Subscriber[T], bufferSize: Int, onOverflow: Long => T): Subscriber.Sync[T] = {
+  /** Returns an instance of a [[DropNewBufferedSubscriber]]
+    * for the [[monix.reactive.OverflowStrategy.DropNew DropNew]]
+    * overflowStrategy.
+    */
+  def withSignal[T](underlying: Subscriber[T], bufferSize: Int, onOverflow: Long => Option[T]): Subscriber.Sync[T] = {
     new DropNewBufferedSubscriber[T](underlying, bufferSize, onOverflow)
   }
 
@@ -318,24 +317,19 @@ private[monix] object DropNewBufferedSubscriber {
     isDoneInProgress: Boolean = false,
     errorThrown: Throwable = null) {
 
-    def upstreamShouldStop: Boolean = {
+    def upstreamShouldStop: Boolean =
       upstreamIsComplete || downstreamIsDone || isDoneInProgress
-    }
 
-    def downstreamComplete: State = {
+    def downstreamComplete: State =
       copy(itemsToPush = 0, downstreamIsDone = true)
-    }
 
-    def declareProcessed(processed: Int): State = {
+    def declareProcessed(processed: Int): State =
       copy(itemsToPush = itemsToPush - processed)
-    }
 
-    def incrementDropped: State = {
+    def incrementDropped: State =
       copy(eventsDropped = eventsDropped + 1)
-    }
 
-    def incrementItemsToPush: State = {
+    def incrementItemsToPush: State =
       copy(itemsToPush = itemsToPush + 1)
-    }
   }
 }

monix-reactive/jvm/src/main/scala/monix/reactive/observers/buffers/EvictingBufferedSubscriber.scala

@@ -26,12 +26,11 @@ import scala.annotation.tailrec
 import scala.util.Failure
 import scala.util.control.NonFatal
 
-/**
- * A [[BufferedSubscriber]] implementation for the
- * [[monix.reactive.OverflowStrategy.DropNew DropNew]] overflow strategy.
- */
+/** A [[BufferedSubscriber]] implementation for the
+  * [[monix.reactive.OverflowStrategy.DropNew DropNew]] overflow strategy.
+  */
 private[buffers] final class EvictingBufferedSubscriber[-T] private
-  (underlying: Subscriber[T], buffer: EvictingQueue[AnyRef], onOverflow: Long => T = null)
+  (underlying: Subscriber[T], buffer: EvictingQueue[AnyRef], onOverflow: Long => Option[T] = null)
   extends BufferedSubscriber[T] with Subscriber.Sync[T] { self =>
 
   implicit val scheduler = underlying.scheduler
@@ -95,12 +94,15 @@ private[buffers] final class EvictingBufferedSubscriber[-T] private
       val count =
         if (eventsDropped > 0 && onOverflow != null) {
           try {
-            val message = onOverflow(eventsDropped).asInstanceOf[AnyRef]
-            eventsDropped = 0
-            consumerBuffer(0) = message
-            1 + buffer.pollMany(consumerBuffer, 1)
-          }
-          catch {
+            onOverflow(eventsDropped) match {
+              case Some(message) =>
+                eventsDropped = 0
+                consumerBuffer(0) = message.asInstanceOf[AnyRef]
+                1 + buffer.pollMany(consumerBuffer, 1)
+              case None =>
+                buffer.pollMany(consumerBuffer)
+            }
+          } catch {
             case NonFatal(ex) =>
               errorThrown = ex
               upstreamIsComplete = true
@@ -221,7 +223,7 @@ private[monix] object EvictingBufferedSubscriber {
    * were dropped.
    */
   def dropOldAndSignal[A](underlying: Subscriber[A],
-    bufferSize: Int, onOverflow: Long => A): Subscriber.Sync[A] = {
+    bufferSize: Int, onOverflow: Long => Option[A]): Subscriber.Sync[A] = {
 
     require(bufferSize > 1,
       "bufferSize must be a strictly positive number, bigger than 1")
@@ -250,7 +252,7 @@ private[monix] object EvictingBufferedSubscriber {
    * were dropped.
    */
   def clearBufferAndSignal[A](underlying: Subscriber[A],
-    bufferSize: Int, onOverflow: Long => A): Subscriber.Sync[A] = {
+    bufferSize: Int, onOverflow: Long => Option[A]): Subscriber.Sync[A] = {
 
     require(bufferSize > 1,
       "bufferSize must be a strictly positive number, bigger than 1")

monix-reactive/jvm/src/test/scala/monix/reactive/observers/BufferDropNewAndSignalConcurrencySuite.scala

@@ -37,18 +37,18 @@ object BufferDropNewAndSignalConcurrencySuite
   }
 
   def buildNewForInt(bufferSize: Int, underlying: Observer[Int])(implicit s: Scheduler) = {
-    BufferedSubscriber(Subscriber(underlying, s), DropNewAndSignal(bufferSize, nr => nr.toInt))
+    BufferedSubscriber(Subscriber(underlying, s), DropNewAndSignal(bufferSize, nr => Some(nr.toInt)))
   }
 
   def buildNewForLong(bufferSize: Int, underlying: Observer[Long])(implicit s: Scheduler) = {
-    BufferedSubscriber(Subscriber(underlying, s), DropNewAndSignal(bufferSize, nr => nr))
+    BufferedSubscriber(Subscriber(underlying, s), DropNewAndSignal(bufferSize, nr => Some(nr)))
   }
 
   test("merge test should work") { implicit s =>
     val num = 100000
     val source = Observable.repeat(1L).take(num)
     val f = Observable.fromIterable(Seq(source, source, source))
-      .mergeMap(x => x)(DropNewAndSignal(1000, dropped => dropped))
+      .mergeMap(x => x)(DropNewAndSignal(1000, dropped => Some(dropped)))
       .sumF
       .runAsyncGetFirst
 

monix-reactive/shared/src/main/scala/monix/reactive/OverflowStrategy.scala

@@ -90,13 +90,20 @@ object OverflowStrategy {
     * the pipeline should begin dropping incoming events until the buffer
     * has room in it again and is free to process more elements.
     *
+    * The given `onOverflow` function get be used for logging the event
+    * and for sending a message to the downstream consumers to inform
+    * them of dropped messages. The function can return `None` in which
+    * case no message is sent and thus you can use it just to log a warning.
+    *
     * @param bufferSize specifies how many events our buffer can hold
     *        before overflowing.
+    *
     * @param onOverflow is a function that can get called on overflow with
     *        a number of messages that were dropped, a function that builds
-    *        a new message that will be sent to downstream.
+    *        a new message that will be sent to downstream. If it returns
+    *        `None`, then no message gets sent to downstream.
     */
-  final case class DropNewAndSignal[A](bufferSize: Int, onOverflow: Long => A)
+  final case class DropNewAndSignal[A](bufferSize: Int, onOverflow: Long => Option[A])
     extends Evicted[A] {
 
     require(bufferSize > 1, "bufferSize should be strictly greater than 1")
@@ -119,13 +126,20 @@ object OverflowStrategy {
     * the currently buffered events should start being dropped in a FIFO order,
     * so the oldest events from the buffer will be dropped first.
     *
+    * The given `onOverflow` function get be used for logging the event
+    * and for sending a message to the downstream consumers to inform
+    * them of dropped messages. The function can return `None` in which
+    * case no message is sent and thus you can use it just to log a warning.
+    *
     * @param bufferSize specifies how many events our buffer can hold
     *        before overflowing
+    *
     * @param onOverflow is a function that can get called on overflow with
     *        a number of messages that were dropped, a function that builds
-    *        a new message that will be sent to downstream.
+    *        a new message that will be sent to downstream. If it returns
+    *        `None`, then no message gets sent to downstream.
     */
-  final case class DropOldAndSignal[A](bufferSize: Int, onOverflow: Long => A)
+  final case class DropOldAndSignal[A](bufferSize: Int, onOverflow: Long => Option[A])
     extends Evicted[A] {
 
     require(bufferSize > 1, "bufferSize should be strictly greater than 1")
@@ -148,19 +162,24 @@ object OverflowStrategy {
     * the current buffer should be dropped completely to make room for
     * new events.
     *
+    * The given `onOverflow` function get be used for logging the event
+    * and for sending a message to the downstream consumers to inform
+    * them of dropped messages. The function can return `None` in which
+    * case no message is sent and thus you can use it just to log a warning.
+    *
     * @param bufferSize specifies how many events our buffer can hold
     *        before overflowing
+    *
     * @param onOverflow is a function that can get called on overflow with
     *        a number of messages that were dropped, a function that builds
     *        a new message that will be sent to downstream.
     */
-  final case class ClearBufferAndSignal[A](bufferSize: Int, onOverflow: Long => A)
+  final case class ClearBufferAndSignal[A](bufferSize: Int, onOverflow: Long => Option[A])
     extends Evicted[A] {
 
     require(bufferSize > 1, "bufferSize should be strictly greater than 1")
   }
 
-
   /** A category of [[OverflowStrategy]] for buffers that can be used
     * synchronously, without worrying about back-pressure concerns.
     */