Added size() to BoundedSourceQueue #30124 (#30125)

akka-stream-tests/src/test/scala/akka/stream/scaladsl/BoundedSourceQueueSpec.scala

@@ -193,5 +193,18 @@ class BoundedSourceQueueSpec extends StreamSpec("""akka.loglevel = debug
 
       downstream.cancel()
     }
+
+    "provide info about number of messages" in {
+      val sub = TestSubscriber.probe[Int]()
+      val queue = Source.queue[Int](100).toMat(Sink.fromSubscriber(sub))(Keep.left).run()
+
+      queue.offer(1)
+      queue.size() shouldBe 1
+
+      (2 to 100).map { i =>
+        queue.offer(i)
+      }
+      queue.size() shouldBe 100
+    }
   }
 }

akka-stream/src/main/mima-filters/2.6.13.backwards.excludes/pr-30125-added-size-to-bounded-source-queue.excludes

@@ -0,0 +1,2 @@
+# disable compatibility check for a new method in BoundedSourceQueue
+ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.stream.BoundedSourceQueue.size")

akka-stream/src/main/scala/akka/stream/BoundedSourceQueue.scala

@@ -32,4 +32,9 @@ trait BoundedSourceQueue[T] {
    * Completes the stream with a failure.
    */
   def fail(ex: Throwable): Unit
+
+  /**
+   * Returns the approximate number of elements in this queue.
+   */
+  def size(): Int
 }

akka-stream/src/main/scala/akka/stream/impl/BoundedSourceQueue.scala

@@ -133,6 +133,8 @@ import akka.stream.stage.{ GraphStageLogic, GraphStageWithMaterializedValue, Out
         if (setDone(Done(QueueOfferResult.Failure(ex))))
           Logic.callback.invoke(()) // if this thread won the completion race also schedule an async callback
       }
+
+      override def size(): Int = queue.size()
     }
 
     // some state transition helpers