You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "He-Pin (via GitHub)" <gi...@apache.org> on 2023/08/22 04:59:56 UTC
[GitHub] [incubator-pekko] He-Pin opened a new pull request, #582: =str Add IterableSource.
He-Pin opened a new pull request, #582:
URL: https://github.com/apache/incubator-pekko/pull/582
was https://github.com/akka/akka/pull/31372
This introduce a dedicated IterableSource, which can be later used for performance optimization. And also obey the supervisionStrategy now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] nvollmar commented on a diff in pull request #582: =str Add IterableSource.
Posted by "nvollmar (via GitHub)" <gi...@apache.org>.
nvollmar commented on code in PR #582:
URL: https://github.com/apache/incubator-pekko/pull/582#discussion_r1301340999
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/SourceSpec.scala:
##########
@@ -311,19 +311,60 @@ class SourceSpec extends StreamSpec with DefaultTimeout {
}
"use decider when iterator throws" in {
+
Review Comment:
I'd prefer to have separate tests with descriptive titles if they test different edge cases, or at least a comment per run hinting what each is testing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin merged pull request #582: =str Add IterableSource.
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin merged PR #582:
URL: https://github.com/apache/incubator-pekko/pull/582
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on a diff in pull request #582: =str Add IterableSource.
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #582:
URL: https://github.com/apache/incubator-pekko/pull/582#discussion_r1301426233
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/IterableSource.scala:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.impl.fusing
+
+import scala.collection.immutable
+import scala.util.control.NonFatal
+
+import org.apache.pekko
+import pekko.stream.{ Attributes, Outlet, SourceShape, Supervision }
+import pekko.stream.ActorAttributes.SupervisionStrategy
+import pekko.stream.impl.ReactiveStreamsCompliance
+import pekko.stream.impl.Stages.DefaultAttributes
+import pekko.stream.stage.{ GraphStage, GraphStageLogic, OutHandler }
+
+private[pekko] final class IterableSource[T](val elements: immutable.Iterable[T]) extends GraphStage[SourceShape[T]] {
+ ReactiveStreamsCompliance.requireNonNullElement(elements)
+
+ override protected def initialAttributes: Attributes = DefaultAttributes.iterableSource
+
+ private val out = Outlet[T]("IterableSource.out")
+ override val shape: SourceShape[T] = SourceShape(out)
+
+ override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
+ new GraphStageLogic(shape) with OutHandler {
+ private lazy val decider = inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider
+ private var currentIterator: Iterator[T] = _
+
+ override def onPull(): Unit =
+ try {
+ if (currentIterator eq null) {
Review Comment:
IIRC the JOOQ does this, and our internal Lindorm Nosql db driver does the same too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] nvollmar commented on a diff in pull request #582: =str Add IterableSource.
Posted by "nvollmar (via GitHub)" <gi...@apache.org>.
nvollmar commented on code in PR #582:
URL: https://github.com/apache/incubator-pekko/pull/582#discussion_r1301419805
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/IterableSource.scala:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.impl.fusing
+
+import scala.collection.immutable
+import scala.util.control.NonFatal
+
+import org.apache.pekko
+import pekko.stream.{ Attributes, Outlet, SourceShape, Supervision }
+import pekko.stream.ActorAttributes.SupervisionStrategy
+import pekko.stream.impl.ReactiveStreamsCompliance
+import pekko.stream.impl.Stages.DefaultAttributes
+import pekko.stream.stage.{ GraphStage, GraphStageLogic, OutHandler }
+
+private[pekko] final class IterableSource[T](val elements: immutable.Iterable[T]) extends GraphStage[SourceShape[T]] {
+ ReactiveStreamsCompliance.requireNonNullElement(elements)
+
+ override protected def initialAttributes: Attributes = DefaultAttributes.iterableSource
+
+ private val out = Outlet[T]("IterableSource.out")
+ override val shape: SourceShape[T] = SourceShape(out)
+
+ override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
+ new GraphStageLogic(shape) with OutHandler {
+ private lazy val decider = inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider
+ private var currentIterator: Iterator[T] = _
+
+ override def onPull(): Unit =
+ try {
+ if (currentIterator eq null) {
+ currentIterator = elements.iterator
+ }
+ tryPushNextOrComplete()
+ } catch {
+ case NonFatal(ex) =>
+ decider(ex) match {
+ case Supervision.Stop => failStage(ex)
+ case Supervision.Resume => tryPushNextOrComplete()
+ case Supervision.Restart =>
+ currentIterator = elements.iterator
+ tryPushNextOrComplete()
+ }
+ }
+
+ private def tryPushNextOrComplete(): Unit =
Review Comment:
That would even be simpler:
```
private def tryPushNextOrComplete(): Unit = {
if (currentIterator.hasNext && isAvailable(out)) {
push(out, currentIterator.next())
}
if (!currentIterator.hasNext) {
completeStage()
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] pjfanning commented on a diff in pull request #582: =str Add IterableSource.
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #582:
URL: https://github.com/apache/incubator-pekko/pull/582#discussion_r1301073530
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/SourceSpec.scala:
##########
@@ -311,19 +311,60 @@ class SourceSpec extends StreamSpec with DefaultTimeout {
}
"use decider when iterator throws" in {
+
+ Source
+ .fromIterator(() => (1 to 5).toIterator.map(k => if (k != 3) k else throw TE("a")))
+ .withAttributes(ActorAttributes.supervisionStrategy(Supervision.stoppingDecider))
+ .grouped(10)
+ .runWith(Sink.head)
+ .failed
+ .futureValue shouldBe an[TE]
+
+ Source
+ .fromIterator(() => (1 to 5).toIterator.map(k => if (k != 3) k else throw TE("a")))
+ .withAttributes(ActorAttributes.supervisionStrategy(Supervision.stoppingDecider))
+ .recoverWithRetries(1, { case _ => Source.empty })
+ .grouped(10)
+ .runWith(Sink.head)
+ .futureValue shouldBe List(1, 2)
+
+ Source
+ .fromIterator(() => (1 to 5).toIterator.map(k => if (k != 3) k else throw TE("a")))
+ .withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
+ .grouped(10)
+ .runWith(Sink.head)
+ .futureValue should ===(List(1, 2, 4, 5))
+
Source
.fromIterator(() => (1 to 5).toIterator.map(k => if (k != 3) k else throw TE("a")))
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.restartingDecider))
.grouped(10)
.runWith(Sink.head)
- .futureValue should ===(List(1, 2))
+ .futureValue should ===(List(1, 2, 1, 2, 1, 2, 1, 2, 1, 2))
Source
.fromIterator(() => (1 to 5).toIterator.map(_ => throw TE("b")))
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.restartingDecider))
.grouped(10)
.runWith(Sink.headOption)
- .futureValue should ===(None)
+ .failed
+ .futureValue shouldBe an[TE]
Review Comment:
`shouldBe a[TE]` would be better
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on a diff in pull request #582: =str Add IterableSource.
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #582:
URL: https://github.com/apache/incubator-pekko/pull/582#discussion_r1301424968
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/IterableSource.scala:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.impl.fusing
+
+import scala.collection.immutable
+import scala.util.control.NonFatal
+
+import org.apache.pekko
+import pekko.stream.{ Attributes, Outlet, SourceShape, Supervision }
+import pekko.stream.ActorAttributes.SupervisionStrategy
+import pekko.stream.impl.ReactiveStreamsCompliance
+import pekko.stream.impl.Stages.DefaultAttributes
+import pekko.stream.stage.{ GraphStage, GraphStageLogic, OutHandler }
+
+private[pekko] final class IterableSource[T](val elements: immutable.Iterable[T]) extends GraphStage[SourceShape[T]] {
+ ReactiveStreamsCompliance.requireNonNullElement(elements)
+
+ override protected def initialAttributes: Attributes = DefaultAttributes.iterableSource
+
+ private val out = Outlet[T]("IterableSource.out")
+ override val shape: SourceShape[T] = SourceShape(out)
+
+ override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
+ new GraphStageLogic(shape) with OutHandler {
+ private lazy val decider = inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider
+ private var currentIterator: Iterator[T] = _
+
+ override def onPull(): Unit =
+ try {
+ if (currentIterator eq null) {
+ currentIterator = elements.iterator
+ }
+ tryPushNextOrComplete()
+ } catch {
+ case NonFatal(ex) =>
+ decider(ex) match {
+ case Supervision.Stop => failStage(ex)
+ case Supervision.Resume => tryPushNextOrComplete()
+ case Supervision.Restart =>
+ currentIterator = elements.iterator
+ tryPushNextOrComplete()
+ }
+ }
+
+ private def tryPushNextOrComplete(): Unit =
Review Comment:
But this will do another `currentIterator.hasNext` checking.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] nvollmar commented on a diff in pull request #582: =str Add IterableSource.
Posted by "nvollmar (via GitHub)" <gi...@apache.org>.
nvollmar commented on code in PR #582:
URL: https://github.com/apache/incubator-pekko/pull/582#discussion_r1301351056
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/IterableSource.scala:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.impl.fusing
+
+import scala.collection.immutable
+import scala.util.control.NonFatal
+
+import org.apache.pekko
+import pekko.stream.{ Attributes, Outlet, SourceShape, Supervision }
+import pekko.stream.ActorAttributes.SupervisionStrategy
+import pekko.stream.impl.ReactiveStreamsCompliance
+import pekko.stream.impl.Stages.DefaultAttributes
+import pekko.stream.stage.{ GraphStage, GraphStageLogic, OutHandler }
+
+private[pekko] final class IterableSource[T](val elements: immutable.Iterable[T]) extends GraphStage[SourceShape[T]] {
+ ReactiveStreamsCompliance.requireNonNullElement(elements)
+
+ override protected def initialAttributes: Attributes = DefaultAttributes.iterableSource
+
+ private val out = Outlet[T]("IterableSource.out")
+ override val shape: SourceShape[T] = SourceShape(out)
+
+ override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
+ new GraphStageLogic(shape) with OutHandler {
+ private lazy val decider = inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider
+ private var currentIterator: Iterator[T] = _
+
+ override def onPull(): Unit =
+ try {
+ if (currentIterator eq null) {
Review Comment:
couldn't we just initialize the variable with `elements.iterator` and avoid the if check/assignment here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] nvollmar commented on a diff in pull request #582: =str Add IterableSource.
Posted by "nvollmar (via GitHub)" <gi...@apache.org>.
nvollmar commented on code in PR #582:
URL: https://github.com/apache/incubator-pekko/pull/582#discussion_r1301347238
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/IterableSource.scala:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.impl.fusing
+
+import scala.collection.immutable
+import scala.util.control.NonFatal
+
+import org.apache.pekko
+import pekko.stream.{ Attributes, Outlet, SourceShape, Supervision }
+import pekko.stream.ActorAttributes.SupervisionStrategy
+import pekko.stream.impl.ReactiveStreamsCompliance
+import pekko.stream.impl.Stages.DefaultAttributes
+import pekko.stream.stage.{ GraphStage, GraphStageLogic, OutHandler }
+
+private[pekko] final class IterableSource[T](val elements: immutable.Iterable[T]) extends GraphStage[SourceShape[T]] {
+ ReactiveStreamsCompliance.requireNonNullElement(elements)
+
+ override protected def initialAttributes: Attributes = DefaultAttributes.iterableSource
+
+ private val out = Outlet[T]("IterableSource.out")
+ override val shape: SourceShape[T] = SourceShape(out)
+
+ override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
+ new GraphStageLogic(shape) with OutHandler {
+ private lazy val decider = inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider
+ private var currentIterator: Iterator[T] = _
+
+ override def onPull(): Unit =
+ try {
+ if (currentIterator eq null) {
+ currentIterator = elements.iterator
+ }
+ tryPushNextOrComplete()
+ } catch {
+ case NonFatal(ex) =>
+ decider(ex) match {
+ case Supervision.Stop => failStage(ex)
+ case Supervision.Resume => tryPushNextOrComplete()
+ case Supervision.Restart =>
+ currentIterator = elements.iterator
+ tryPushNextOrComplete()
+ }
+ }
+
+ private def tryPushNextOrComplete(): Unit =
+ if (currentIterator.hasNext) {
+ if (isAvailable(out)) {
+ push(out, currentIterator.next())
+ }
+ if (!currentIterator.hasNext) {
Review Comment:
This check only makes sense if we pushed an element -> move it inside the `if (isAvailable(out)) {` block
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on a diff in pull request #582: =str Add IterableSource.
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #582:
URL: https://github.com/apache/incubator-pekko/pull/582#discussion_r1301374770
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/IterableSource.scala:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.impl.fusing
+
+import scala.collection.immutable
+import scala.util.control.NonFatal
+
+import org.apache.pekko
+import pekko.stream.{ Attributes, Outlet, SourceShape, Supervision }
+import pekko.stream.ActorAttributes.SupervisionStrategy
+import pekko.stream.impl.ReactiveStreamsCompliance
+import pekko.stream.impl.Stages.DefaultAttributes
+import pekko.stream.stage.{ GraphStage, GraphStageLogic, OutHandler }
+
+private[pekko] final class IterableSource[T](val elements: immutable.Iterable[T]) extends GraphStage[SourceShape[T]] {
+ ReactiveStreamsCompliance.requireNonNullElement(elements)
+
+ override protected def initialAttributes: Attributes = DefaultAttributes.iterableSource
+
+ private val out = Outlet[T]("IterableSource.out")
+ override val shape: SourceShape[T] = SourceShape(out)
+
+ override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
+ new GraphStageLogic(shape) with OutHandler {
+ private lazy val decider = inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider
+ private var currentIterator: Iterator[T] = _
+
+ override def onPull(): Unit =
+ try {
+ if (currentIterator eq null) {
Review Comment:
There are some cases where the `iterator` method call invoving some underling resources creation, that's why I defer the creation in the first place.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org