You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by he...@apache.org on 2024/04/05 14:44:49 UTC
(pekko) branch main updated: chore: Make use of splitIterator directly (#1249)
This is an automated email from the ASF dual-hosted git repository.
hepin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 24e47b8cac chore: Make use of splitIterator directly (#1249)
24e47b8cac is described below
commit 24e47b8cac2da9ae6024bb0730848b435fa5fb6b
Author: He-Pin(kerr) <he...@gmail.com>
AuthorDate: Fri Apr 5 22:44:44 2024 +0800
chore: Make use of splitIterator directly (#1249)
---
.../pekko/stream/impl/JavaStreamSource.scala | 38 +++++++++++++---------
1 file changed, 22 insertions(+), 16 deletions(-)
diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/JavaStreamSource.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/JavaStreamSource.scala
index 3172359a15..74bba55d0a 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/impl/JavaStreamSource.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/impl/JavaStreamSource.scala
@@ -18,6 +18,9 @@ import pekko.annotation.InternalApi
import pekko.stream._
import pekko.stream.stage.{ GraphStage, GraphStageLogic, OutHandler }
+import java.util.Spliterator
+import java.util.function.Consumer
+
/** INTERNAL API */
@InternalApi private[stream] final class JavaStreamSource[T, S <: java.util.stream.BaseStream[T, S]](
open: () => java.util.stream.BaseStream[T, S])
@@ -27,28 +30,31 @@ import pekko.stream.stage.{ GraphStage, GraphStageLogic, OutHandler }
override val shape: SourceShape[T] = SourceShape(out)
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
- new GraphStageLogic(shape) with OutHandler {
- private[this] var stream: java.util.stream.BaseStream[T, S] = _
- private[this] var iter: java.util.Iterator[T] = _
-
- setHandler(out, this)
+ new GraphStageLogic(shape) with OutHandler with Consumer[T] {
+ private var stream: java.util.stream.BaseStream[T, S] = _
+ private var splitIterator: Spliterator[T] = _
- override def preStart(): Unit = {
+ final override def preStart(): Unit = {
stream = open()
- iter = stream.iterator()
+ splitIterator = stream.spliterator()
+ if (splitIterator.hasCharacteristics(Spliterator.SIZED) && splitIterator.estimateSize() == 0) {
+ complete(out)
+ }
}
- override def postStop(): Unit = {
- if (stream ne null)
- stream.close()
- }
+ final override def accept(elem: T): Unit = push(out, elem)
- override def onPull(): Unit = {
- if (iter.hasNext) {
- push(out, iter.next())
- } else {
- complete(out)
+ final override def postStop(): Unit =
+ if (stream ne null) {
+ stream.close()
}
+
+ final override def onPull(): Unit = if (!splitIterator.tryAdvance(this)) {
+ complete(out)
}
+
+ setHandler(out, this)
}
+
+ override def toString: String = "JavaStreamSource"
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org