You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "injae-kim (via GitHub)" <gi...@apache.org> on 2024/01/27 10:29:32 UTC
[PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
injae-kim opened a new pull request, #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053
### Motivation
- Pekko has `mapwithResource` now, would be better to support `AutoCloseable` in Java/Scala dsl too.
### Modification
- Add `AutoCloseable` shortcut on `mapWithResource`
### Result
- Now `mapWithResource` will close the `AutoCloseable` so user don't need to pass close function
- Closes https://github.com/apache/incubator-pekko/issues/1044
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "injae-kim (via GitHub)" <gi...@apache.org>.
injae-kim commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1913109773
(I'm fixing format and checkStyle~)
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468516974
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala:
##########
@@ -2541,6 +2541,45 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
(resource, out) => f(resource, out),
resource => close.apply(resource).toScala))
+ /**
+ * Transform each stream element with the help of a [[AutoCloseable]] resource and close it when the stream finishes or fails.
+ *
+ * The resource creation function is invoked once when the stream is materialized and the returned resource is passed to
+ * the mapping function for mapping the first element. The mapping function returns a mapped element to emit
+ * downstream. The returned `T` MUST NOT be `null` as it is illegal as stream element - according to the Reactive Streams specification.
+ *
+ * The [[AutoCloseable]] resource is closed only once when the upstream or downstream finishes or fails.
+ *
+ * Early completion can be done with combination of the [[takeWhile]] operator.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * You can configure the default dispatcher for this Source by changing the `akka.stream.materializer.blocking-io-dispatcher` or
Review Comment:
don't fix this - this is not your mistake - I will do a PR fixing the legacy bug in the docs
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1473839887
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala:
##########
@@ -276,6 +276,45 @@ class SubSource[Out, Mat](
(resource, out) => f(resource, out),
resource => close.apply(resource).toScala))
+ /**
+ * Transform each stream element with the help of a [[AutoCloseable]] resource and close it when the stream finishes or fails.
Review Comment:
`an`
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1913289541
Good οΌI that will be a nice addition, but that will need more with stop, restart, resume strategy. @injae-kim so there will need 3 more test cases, because this operator is too essential, thanks .
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1917641076
@Roiocam ping for additional CR, as a Java CRUD guy, I love this.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "injae-kim (via GitHub)" <gi...@apache.org>.
injae-kim commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1913123914
oh~ I'm not sure should I update doc. I'll update `mapWithResource.md` doc too and update PR~! thanks!
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "injae-kim (via GitHub)" <gi...@apache.org>.
injae-kim commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1913109962
FYI) it's my first time to use `scala` so feel free to share any suggestion~! I really newbie on `scala`π
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468443645
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapWithResourceSpec.scala:
##########
@@ -410,6 +410,24 @@ class FlowMapWithResourceSpec extends StreamSpec(UnboundedMailboxConfig) {
Await.result(promise.future, 3.seconds) shouldBe Done
}
+ "will close the autocloseable resource" in {
+ val closedCounter = new AtomicInteger(0)
+ val create = () => new AutoCloseable {
+ override def close(): Unit = closedCounter.incrementAndGet()
+ }
+ val (pub, sub) = TestSource
+ .probe[Int]
+ .mapWithResource(create, (_, count) => count)
+ .toMat(TestSink.probe)(Keep.both)
+ .run()
+ sub.expectSubscription().request(2)
+ pub.sendNext(1)
+ sub.expectNext(1)
+ pub.sendComplete()
+ sub.expectComplete()
+ closedCounter.get shouldBe 1
Review Comment:
is there any chance that this test could check the function runs before the close? I would create a class that is AutoCloseable - the close counts the close calls, but the function that the class exposes is set up to check the close count and if it is non-zero, the function should fail.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "injae-kim (via GitHub)" <gi...@apache.org>.
injae-kim commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468506714
##########
stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java:
##########
@@ -237,6 +238,22 @@ public void mustBeAbleToUseMapWithResource() {
Assert.assertFalse(gate.get());
}
+ @Test
+ public void mustBeAbleToUseMapWithAutoCloseableResource() {
+ final AtomicInteger closed = new AtomicInteger();
+ Source.from(Arrays.asList("1", "2", "3"))
+ .via(
+ Flow.of(String.class)
+ .mapWithResource(
+ () -> (AutoCloseable) closed::incrementAndGet,
+ (resource, elem) -> elem))
+ .runWith(TestSink.create(system), system)
+ .request(4)
+ .expectNext("1", "2", "3")
+ .expectComplete();
+ Assert.assertEquals(closed.get(), 3);
Review Comment:
β
updated
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala:
##########
@@ -828,6 +828,44 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends Gr
(resource, out) => f(resource, out),
resource => close.apply(resource).toScala))
+ /**
+ * Transform each stream element with the help of a [[AutoCloseable]] resource and close it.
Review Comment:
β
updated!
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "injae-kim (via GitHub)" <gi...@apache.org>.
injae-kim commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468532759
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapWithResourceSpec.scala:
##########
@@ -410,6 +410,114 @@ class FlowMapWithResourceSpec extends StreamSpec(UnboundedMailboxConfig) {
Await.result(promise.future, 3.seconds) shouldBe Done
}
+ "will close the autocloseable resource when upstream complete" in {
Review Comment:
https://github.com/apache/incubator-pekko/blob/a2a42eab33bbe68e2dd89f4cef9fb960af9bdce5/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapWithResourceSpec.scala#L304
https://github.com/apache/incubator-pekko/blob/a2a42eab33bbe68e2dd89f4cef9fb960af9bdce5/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapWithResourceSpec.scala#L326
oh I just copy&paste from 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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "injae-kim (via GitHub)" <gi...@apache.org>.
injae-kim commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468516459
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala:
##########
@@ -2541,6 +2541,45 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
(resource, out) => f(resource, out),
resource => close.apply(resource).toScala))
+ /**
+ * Transform each stream element with the help of a [[AutoCloseable]] resource and close it when the stream finishes or fails.
+ *
+ * The resource creation function is invoked once when the stream is materialized and the returned resource is passed to
+ * the mapping function for mapping the first element. The mapping function returns a mapped element to emit
+ * downstream. The returned `T` MUST NOT be `null` as it is illegal as stream element - according to the Reactive Streams specification.
+ *
+ * The [[AutoCloseable]] resource is closed only once when the upstream or downstream finishes or fails.
+ *
+ * Early completion can be done with combination of the [[takeWhile]] operator.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * You can configure the default dispatcher for this Source by changing the `akka.stream.materializer.blocking-io-dispatcher` or
Review Comment:
β
updated
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "injae-kim (via GitHub)" <gi...@apache.org>.
injae-kim commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468506922
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapWithResourceSpec.scala:
##########
@@ -410,6 +410,24 @@ class FlowMapWithResourceSpec extends StreamSpec(UnboundedMailboxConfig) {
Await.result(promise.future, 3.seconds) shouldBe Done
}
+ "will close the autocloseable resource" in {
+ val closedCounter = new AtomicInteger(0)
+ val create = () => new AutoCloseable {
+ override def close(): Unit = closedCounter.incrementAndGet()
+ }
+ val (pub, sub) = TestSource
+ .probe[Int]
+ .mapWithResource(create, (_, count) => count)
+ .toMat(TestSink.probe)(Keep.both)
+ .run()
+ sub.expectSubscription().request(2)
+ pub.sendNext(1)
+ sub.expectNext(1)
+ pub.sendComplete()
+ sub.expectComplete()
+ closedCounter.get shouldBe 1
Review Comment:
β
updated, I enhance test to check AutoCloseable is closed after stream is completed~!
Also I add more test about upstream/downstream failure.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468442746
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala:
##########
@@ -828,6 +828,44 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends Gr
(resource, out) => f(resource, out),
resource => close.apply(resource).toScala))
+ /**
+ * Transform each stream element with the help of a [[AutoCloseable]] resource and close it.
Review Comment:
close it when the stream finishes or fails.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1473839731
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala:
##########
@@ -285,6 +285,45 @@ class SubFlow[In, Out, Mat](
(resource, out) => f(resource, out),
resource => close.apply(resource).toScala))
+ /**
+ * Transform each stream element with the help of a [[AutoCloseable]] resource and close it when the stream finishes or fails.
Review Comment:
`an`
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1921033094
Thank you @injae-kim
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin merged PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1913110142
Thank you.
1. We will need an apache cla before we can merge this
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1913111342
Can't wait to use this on may daily job
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1913511535
Yeah, that's true, But we are not using Scala 2.13, maybe a conditional compiling can help us add that `using` to 2.13? I don't know when you plan to drop the 2.12 support:), seems like 2 years later.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1913263904
It's received, @injae-kim ,thank you.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468530877
##########
stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SourceTest.java:
##########
@@ -815,6 +816,18 @@ public void mustBeAbleToUseMapWithResource() {
Assert.assertFalse(gate.get());
}
+ @Test
+ public void mustBeAbleToUseMapWithAutoCloseableResource() {
+ final TestKit probe = new TestKit(system);
+ final AtomicInteger closed = new AtomicInteger();
+ Source.from(Arrays.asList("1", "2", "3"))
Review Comment:
I want add something like `Flux.elements` for java dsl...
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468531103
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapWithResourceSpec.scala:
##########
@@ -410,6 +410,114 @@ class FlowMapWithResourceSpec extends StreamSpec(UnboundedMailboxConfig) {
Await.result(promise.future, 3.seconds) shouldBe Done
}
+ "will close the autocloseable resource when upstream complete" in {
Review Comment:
completes?
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1471667811
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapWithResourceSpec.scala:
##########
@@ -518,6 +518,118 @@ class FlowMapWithResourceSpec extends StreamSpec(UnboundedMailboxConfig) {
closedCounter.get shouldBe 1
}
+ "continue with autoCloseable when Strategy is Resume and exception happened" in {
+ val closedCounter = new AtomicInteger(0)
+ val create = () =>
+ new AutoCloseable {
+ override def close(): Unit = closedCounter.incrementAndGet()
+ }
+ val p = Source
+ .fromIterator(() => (0 to 50).iterator)
+ .mapWithResource(create,
+ (_: AutoCloseable, elem) => {
+ if (elem == 10) throw TE("") else elem
+ })
+ .withAttributes(supervisionStrategy(resumingDecider))
+ .runWith(Sink.asPublisher(false))
+ val c = TestSubscriber.manualProbe[Int]()
+
+ p.subscribe(c)
+ val sub = c.expectSubscription()
+
+ (0 to 48).foreach(i => {
+ sub.request(1)
+ c.expectNext() should ===(if (i < 10) i else i + 1)
+ })
+ sub.request(1)
+ c.expectNext(50)
+ c.expectComplete()
+ closedCounter.get shouldBe 1
+ }
+
+ "continue with autoCloseable when Strategy is Resume and exception happened on map" in {
+ val closedCounter = new AtomicInteger(0)
+ val create = () =>
+ new AutoCloseable {
+ override def close(): Unit = closedCounter.incrementAndGet()
+ }
+ val p = Source
+ .fromIterator(() => (0 to 50).iterator)
+ .mapWithResource(create, (_: AutoCloseable, elem) => elem)
+ .map(elem => {
+ if (elem == 10) throw TE("") else elem
+ })
+ .withAttributes(supervisionStrategy(resumingDecider))
+ .runWith(Sink.asPublisher(false))
+ val c = TestSubscriber.manualProbe[Int]()
+
+ p.subscribe(c)
+ val sub = c.expectSubscription()
+
+ (0 to 48).foreach(i => {
+ sub.request(1)
+ c.expectNext() should ===(if (i < 10) i else i + 1)
+ })
+ sub.request(1)
+ c.expectNext(50)
+ c.expectComplete()
+ closedCounter.get shouldBe 1
+ }
+
+ "close and open stream with autocloseable again when Strategy is Restart" in {
+ val closedCounter = new AtomicInteger(0)
+ val create = () =>
+ new AutoCloseable {
+ override def close(): Unit = closedCounter.incrementAndGet()
+ }
+ val p = Source
+ .fromIterator(() => (0 to 50).iterator)
+ .mapWithResource(create,
+ (_: AutoCloseable, elem) => {
+ if (elem == 10) throw TE("") else elem
+ })
+ .withAttributes(supervisionStrategy(restartingDecider))
Review Comment:
for a restart, the Resource is recreated again, so how about add a assert for the Resource is been created x times and closed x times?
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1918930300
@pjfanning Would you like give this another round review?
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "injae-kim (via GitHub)" <gi...@apache.org>.
injae-kim commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1929966485
Thank you to all reviewers! without your kind helps, I wouldn't have created this PR :)
I'll continue to contribute on pekko with another PR π
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468511471
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala:
##########
@@ -2541,6 +2541,45 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
(resource, out) => f(resource, out),
resource => close.apply(resource).toScala))
+ /**
+ * Transform each stream element with the help of a [[AutoCloseable]] resource and close it when the stream finishes or fails.
+ *
+ * The resource creation function is invoked once when the stream is materialized and the returned resource is passed to
+ * the mapping function for mapping the first element. The mapping function returns a mapped element to emit
+ * downstream. The returned `T` MUST NOT be `null` as it is illegal as stream element - according to the Reactive Streams specification.
+ *
+ * The [[AutoCloseable]] resource is closed only once when the upstream or downstream finishes or fails.
+ *
+ * Early completion can be done with combination of the [[takeWhile]] operator.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * You can configure the default dispatcher for this Source by changing the `akka.stream.materializer.blocking-io-dispatcher` or
Review Comment:
`akka.stream.materializer` should be `pekko.stream.materializer`
and other places, thanks.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468508781
##########
stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SourceTest.java:
##########
@@ -815,6 +816,19 @@ public void mustBeAbleToUseMapWithResource() {
Assert.assertFalse(gate.get());
}
+ @Test
+ public void mustBeAbleToUseMapWithAutoCloseableResource() {
+ final TestKit probe = new TestKit(system);
+ final AtomicInteger closed = new AtomicInteger();
+ Source.from(Arrays.asList("1", "2", "3"))
+ .mapWithResource(() -> (AutoCloseable) closed::incrementAndGet, (resource, elem) -> elem)
+ .runWith(Sink.foreach(elem -> probe.getRef().tell(elem, ActorRef.noSender())), system);
+
+ Assert.assertEquals(closed.get(), 0);
Review Comment:
This will be flaky, maybe we can just remove it.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "injae-kim (via GitHub)" <gi...@apache.org>.
injae-kim commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1913110566
aha~ I checked it on contributing.md and I'll submit cla~!
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468442746
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala:
##########
@@ -828,6 +828,44 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends Gr
(resource, out) => f(resource, out),
resource => close.apply(resource).toScala))
+ /**
+ * Transform each stream element with the help of a [[AutoCloseable]] resource and close it.
Review Comment:
close it when the stream completes.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "injae-kim (via GitHub)" <gi...@apache.org>.
injae-kim commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468526428
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala:
##########
@@ -2541,6 +2541,45 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
(resource, out) => f(resource, out),
resource => close.apply(resource).toScala))
+ /**
+ * Transform each stream element with the help of a [[AutoCloseable]] resource and close it when the stream finishes or fails.
+ *
+ * The resource creation function is invoked once when the stream is materialized and the returned resource is passed to
+ * the mapping function for mapping the first element. The mapping function returns a mapped element to emit
+ * downstream. The returned `T` MUST NOT be `null` as it is illegal as stream element - according to the Reactive Streams specification.
+ *
+ * The [[AutoCloseable]] resource is closed only once when the upstream or downstream finishes or fails.
+ *
+ * Early completion can be done with combination of the [[takeWhile]] operator.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * You can configure the default dispatcher for this Source by changing the `akka.stream.materializer.blocking-io-dispatcher` or
Review Comment:
oh I got it. thanks for your fix!
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "injae-kim (via GitHub)" <gi...@apache.org>.
injae-kim commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468516343
##########
stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java:
##########
@@ -237,6 +238,22 @@ public void mustBeAbleToUseMapWithResource() {
Assert.assertFalse(gate.get());
}
+ @Test
+ public void mustBeAbleToUseMapWithAutoCloseableResource() {
+ final TestKit probe = new TestKit(system);
+ final AtomicInteger closed = new AtomicInteger();
+ Source.from(Arrays.asList("1", "2", "3"))
+ .via(
+ Flow.of(String.class)
+ .mapWithResource(
+ () -> (AutoCloseable) closed::incrementAndGet, (resource, elem) -> elem))
+ .runWith(Sink.foreach(elem -> probe.getRef().tell(elem, ActorRef.noSender())), system);
Review Comment:
Hmm actually I just copy&paste this `probe.getRef` from other test π
I don't know how I can avoid this `probe.getRef` so If you suggest some example code then I'll update it~! thanks!
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "injae-kim (via GitHub)" <gi...@apache.org>.
injae-kim commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1913116339
I submitted cls agreement PDF by email~ I'll share you when I got reply.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "injae-kim (via GitHub)" <gi...@apache.org>.
injae-kim commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468516012
##########
stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java:
##########
@@ -237,6 +238,22 @@ public void mustBeAbleToUseMapWithResource() {
Assert.assertFalse(gate.get());
}
+ @Test
+ public void mustBeAbleToUseMapWithAutoCloseableResource() {
+ final TestKit probe = new TestKit(system);
+ final AtomicInteger closed = new AtomicInteger();
+ Source.from(Arrays.asList("1", "2", "3"))
+ .via(
+ Flow.of(String.class)
+ .mapWithResource(
+ () -> (AutoCloseable) closed::incrementAndGet, (resource, elem) -> elem))
+ .runWith(Sink.foreach(elem -> probe.getRef().tell(elem, ActorRef.noSender())), system);
+
+ Assert.assertEquals(closed.get(), 0);
Review Comment:
β
updated
##########
stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SourceTest.java:
##########
@@ -815,6 +816,19 @@ public void mustBeAbleToUseMapWithResource() {
Assert.assertFalse(gate.get());
}
+ @Test
+ public void mustBeAbleToUseMapWithAutoCloseableResource() {
+ final TestKit probe = new TestKit(system);
+ final AtomicInteger closed = new AtomicInteger();
+ Source.from(Arrays.asList("1", "2", "3"))
+ .mapWithResource(() -> (AutoCloseable) closed::incrementAndGet, (resource, elem) -> elem)
+ .runWith(Sink.foreach(elem -> probe.getRef().tell(elem, ActorRef.noSender())), system);
+
+ Assert.assertEquals(closed.get(), 0);
Review Comment:
β
updated
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468510931
##########
stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java:
##########
@@ -237,6 +238,22 @@ public void mustBeAbleToUseMapWithResource() {
Assert.assertFalse(gate.get());
}
+ @Test
+ public void mustBeAbleToUseMapWithAutoCloseableResource() {
+ final TestKit probe = new TestKit(system);
+ final AtomicInteger closed = new AtomicInteger();
+ Source.from(Arrays.asList("1", "2", "3"))
+ .via(
+ Flow.of(String.class)
+ .mapWithResource(
+ () -> (AutoCloseable) closed::incrementAndGet, (resource, elem) -> elem))
+ .runWith(Sink.foreach(elem -> probe.getRef().tell(elem, ActorRef.noSender())), system);
Review Comment:
The StreamTestkit can be used in Java too, so we can avoid the `probe.getRef`, it's fine too, up to you:)
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1920590866
@injae-kim Good afternoon, seems only the last typo need to change, do you have any further improvement, thanks.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "jxnu-liguobin (via GitHub)" <gi...@apache.org>.
jxnu-liguobin commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1473824002
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala:
##########
@@ -2541,6 +2541,45 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
(resource, out) => f(resource, out),
resource => close.apply(resource).toScala))
+ /**
+ * Transform each stream element with the help of a [[AutoCloseable]] resource and close it when the stream finishes or fails.
Review Comment:
an
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "injae-kim (via GitHub)" <gi...@apache.org>.
injae-kim commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1913227732
Now I can build&run test on my local π thanks a lot for your help~!
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "injae-kim (via GitHub)" <gi...@apache.org>.
injae-kim commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468532511
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala:
##########
@@ -2541,6 +2541,45 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
(resource, out) => f(resource, out),
resource => close.apply(resource).toScala))
+ /**
+ * Transform each stream element with the help of a [[AutoCloseable]] resource and close it when the stream finishes or fails.
+ *
+ * The resource creation function is invoked once when the stream is materialized and the returned resource is passed to
+ * the mapping function for mapping the first element. The mapping function returns a mapped element to emit
+ * downstream. The returned `T` MUST NOT be `null` as it is illegal as stream element - according to the Reactive Streams specification.
+ *
+ * The [[AutoCloseable]] resource is closed only once when the upstream or downstream finishes or fails.
+ *
+ * Early completion can be done with combination of the [[takeWhile]] operator.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * You can configure the default dispatcher for this Source by changing the `akka.stream.materializer.blocking-io-dispatcher` or
Review Comment:
β
rebased to latest main branch.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468525630
##########
stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java:
##########
@@ -237,6 +238,22 @@ public void mustBeAbleToUseMapWithResource() {
Assert.assertFalse(gate.get());
}
+ @Test
+ public void mustBeAbleToUseMapWithAutoCloseableResource() {
+ final TestKit probe = new TestKit(system);
+ final AtomicInteger closed = new AtomicInteger();
+ Source.from(Arrays.asList("1", "2", "3"))
+ .via(
+ Flow.of(String.class)
+ .mapWithResource(
+ () -> (AutoCloseable) closed::incrementAndGet, (resource, elem) -> elem))
+ .runWith(Sink.foreach(elem -> probe.getRef().tell(elem, ActorRef.noSender())), system);
Review Comment:
I knew it, because that me.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1913279918
> don't make any changes about this (yet) - but what about [Closeable](https://docs.oracle.com/javase/8/docs/api/java/io/Closeable.html)? Most resource management frameworks work with both.
Iirc, closable extends autocloseableοΌ
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1913199139
@injae-kim In very good shape, just need another update, thanks.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "injae-kim (via GitHub)" <gi...@apache.org>.
injae-kim commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1913478168
> interface `Closeable` extends `AutoCloseable` {
Hmm `Closeable` is sub type of `AutoCloseable`, so isn't it better to support super type `AutoCloseable`?
Maybe there's some class that implements only `AutoCloseable` π€
> at least 1 test when the map function fails with an exception.
> need more ,with stop, restart, resume strategy test
Nice suggestion! I'll add test cases and update PR soon~!
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1917551780
@injae-kim Thank you ,seems you will just need a final update and then looks good to me. as @mdedetrich , I think that's will not be a problem, because Even `FS2` supports `AutoCloseable`.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468444576
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapWithResourceSpec.scala:
##########
@@ -410,6 +410,24 @@ class FlowMapWithResourceSpec extends StreamSpec(UnboundedMailboxConfig) {
Await.result(promise.future, 3.seconds) shouldBe Done
}
+ "will close the autocloseable resource" in {
+ val closedCounter = new AtomicInteger(0)
+ val create = () => new AutoCloseable {
+ override def close(): Unit = closedCounter.incrementAndGet()
+ }
+ val (pub, sub) = TestSource
+ .probe[Int]
+ .mapWithResource(create, (_, count) => count)
+ .toMat(TestSink.probe)(Keep.both)
+ .run()
+ sub.expectSubscription().request(2)
+ pub.sendNext(1)
+ sub.expectNext(1)
+ pub.sendComplete()
+ sub.expectComplete()
+ closedCounter.get shouldBe 1
Review Comment:
and we need tests where the function fails so that we can see that the close still happens (but only after the function fails)
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1913123543
@injae-kim I think you can update the `mapWithResource.md` too, you can do a local preview with `sbt paradoxBrowse` ,have a nice weekend.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468517708
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala:
##########
@@ -2541,6 +2541,45 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
(resource, out) => f(resource, out),
resource => close.apply(resource).toScala))
+ /**
+ * Transform each stream element with the help of a [[AutoCloseable]] resource and close it when the stream finishes or fails.
+ *
+ * The resource creation function is invoked once when the stream is materialized and the returned resource is passed to
+ * the mapping function for mapping the first element. The mapping function returns a mapped element to emit
+ * downstream. The returned `T` MUST NOT be `null` as it is illegal as stream element - according to the Reactive Streams specification.
+ *
+ * The [[AutoCloseable]] resource is closed only once when the upstream or downstream finishes or fails.
+ *
+ * Early completion can be done with combination of the [[takeWhile]] operator.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * You can configure the default dispatcher for this Source by changing the `akka.stream.materializer.blocking-io-dispatcher` or
Review Comment:
I created #1057
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468527434
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala:
##########
@@ -2541,6 +2541,45 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
(resource, out) => f(resource, out),
resource => close.apply(resource).toScala))
+ /**
+ * Transform each stream element with the help of a [[AutoCloseable]] resource and close it when the stream finishes or fails.
+ *
+ * The resource creation function is invoked once when the stream is materialized and the returned resource is passed to
+ * the mapping function for mapping the first element. The mapping function returns a mapped element to emit
+ * downstream. The returned `T` MUST NOT be `null` as it is illegal as stream element - according to the Reactive Streams specification.
+ *
+ * The [[AutoCloseable]] resource is closed only once when the upstream or downstream finishes or fails.
+ *
+ * Early completion can be done with combination of the [[takeWhile]] operator.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * You can configure the default dispatcher for this Source by changing the `akka.stream.materializer.blocking-io-dispatcher` or
Review Comment:
the fix is now Merged, please do a rebase, sorry for the bother.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468511471
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala:
##########
@@ -2541,6 +2541,45 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
(resource, out) => f(resource, out),
resource => close.apply(resource).toScala))
+ /**
+ * Transform each stream element with the help of a [[AutoCloseable]] resource and close it when the stream finishes or fails.
+ *
+ * The resource creation function is invoked once when the stream is materialized and the returned resource is passed to
+ * the mapping function for mapping the first element. The mapping function returns a mapped element to emit
+ * downstream. The returned `T` MUST NOT be `null` as it is illegal as stream element - according to the Reactive Streams specification.
+ *
+ * The [[AutoCloseable]] resource is closed only once when the upstream or downstream finishes or fails.
+ *
+ * Early completion can be done with combination of the [[takeWhile]] operator.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * You can configure the default dispatcher for this Source by changing the `akka.stream.materializer.blocking-io-dispatcher` or
Review Comment:
`akka.stream.materializer` should be pekko.stream.materializer`
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala:
##########
@@ -2541,6 +2541,45 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
(resource, out) => f(resource, out),
resource => close.apply(resource).toScala))
+ /**
+ * Transform each stream element with the help of a [[AutoCloseable]] resource and close it when the stream finishes or fails.
+ *
+ * The resource creation function is invoked once when the stream is materialized and the returned resource is passed to
+ * the mapping function for mapping the first element. The mapping function returns a mapped element to emit
+ * downstream. The returned `T` MUST NOT be `null` as it is illegal as stream element - according to the Reactive Streams specification.
+ *
+ * The [[AutoCloseable]] resource is closed only once when the upstream or downstream finishes or fails.
+ *
+ * Early completion can be done with combination of the [[takeWhile]] operator.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * You can configure the default dispatcher for this Source by changing the `akka.stream.materializer.blocking-io-dispatcher` or
Review Comment:
`akka.stream.materializer` should be `pekko.stream.materializer`
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1471665460
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapWithResourceSpec.scala:
##########
@@ -518,6 +518,118 @@ class FlowMapWithResourceSpec extends StreamSpec(UnboundedMailboxConfig) {
closedCounter.get shouldBe 1
}
+ "continue with autoCloseable when Strategy is Resume and exception happened" in {
+ val closedCounter = new AtomicInteger(0)
+ val create = () =>
+ new AutoCloseable {
+ override def close(): Unit = closedCounter.incrementAndGet()
+ }
+ val p = Source
+ .fromIterator(() => (0 to 50).iterator)
+ .mapWithResource(create,
+ (_: AutoCloseable, elem) => {
+ if (elem == 10) throw TE("") else elem
+ })
+ .withAttributes(supervisionStrategy(resumingDecider))
+ .runWith(Sink.asPublisher(false))
+ val c = TestSubscriber.manualProbe[Int]()
+
+ p.subscribe(c)
+ val sub = c.expectSubscription()
+
+ (0 to 48).foreach(i => {
+ sub.request(1)
+ c.expectNext() should ===(if (i < 10) i else i + 1)
+ })
+ sub.request(1)
+ c.expectNext(50)
+ c.expectComplete()
+ closedCounter.get shouldBe 1
+ }
+
+ "continue with autoCloseable when Strategy is Resume and exception happened on map" in {
+ val closedCounter = new AtomicInteger(0)
+ val create = () =>
+ new AutoCloseable {
+ override def close(): Unit = closedCounter.incrementAndGet()
+ }
+ val p = Source
+ .fromIterator(() => (0 to 50).iterator)
+ .mapWithResource(create, (_: AutoCloseable, elem) => elem)
+ .map(elem => {
+ if (elem == 10) throw TE("") else elem
Review Comment:
This is not right, @pjfanning 's map means the block inside `mapWithResource`:)
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "injae-kim (via GitHub)" <gi...@apache.org>.
injae-kim commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1472688447
##########
docs/src/main/paradox/stream/operators/Source-or-Flow/mapWithResource.md:
##########
@@ -11,6 +11,10 @@ Map elements with the help of a resource that can be opened, transform each elem
2. `f`: Transform each element inputs with the help of resource.
Review Comment:
β
updated!
##########
docs/src/main/paradox/stream/operators/Source-or-Flow/mapWithResource.md:
##########
@@ -11,6 +11,10 @@ Map elements with the help of a resource that can be opened, transform each elem
2. `f`: Transform each element inputs with the help of resource.
3. `close`: Close the resource, invoked on end of stream or if the stream fails, optionally outputting a last element.
+@apidoc[Flow.mapWithResource](Flow) { scala="#mapWithResource%5BS%20%3C%3A%20AutoCloseable%2C%20T%5D%28create%3A%20%28%29%20%3D%3E%20S%2C%20f%3A%20%28S%2C%20Out%29%20%3D%3E%20T%29%3A%20Repr%5BT%5D" java="#mapWithResource(org.apache.pekko.japi.function.Creator,org.apache.pekko.japi.function.Function2)" }
Review Comment:
β
updated!
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "injae-kim (via GitHub)" <gi...@apache.org>.
injae-kim commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1471718132
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapWithResourceSpec.scala:
##########
@@ -518,6 +518,118 @@ class FlowMapWithResourceSpec extends StreamSpec(UnboundedMailboxConfig) {
closedCounter.get shouldBe 1
}
+ "continue with autoCloseable when Strategy is Resume and exception happened" in {
+ val closedCounter = new AtomicInteger(0)
+ val create = () =>
+ new AutoCloseable {
+ override def close(): Unit = closedCounter.incrementAndGet()
+ }
+ val p = Source
+ .fromIterator(() => (0 to 50).iterator)
+ .mapWithResource(create,
+ (_: AutoCloseable, elem) => {
+ if (elem == 10) throw TE("") else elem
+ })
+ .withAttributes(supervisionStrategy(resumingDecider))
+ .runWith(Sink.asPublisher(false))
+ val c = TestSubscriber.manualProbe[Int]()
+
+ p.subscribe(c)
+ val sub = c.expectSubscription()
+
+ (0 to 48).foreach(i => {
+ sub.request(1)
+ c.expectNext() should ===(if (i < 10) i else i + 1)
+ })
+ sub.request(1)
+ c.expectNext(50)
+ c.expectComplete()
+ closedCounter.get shouldBe 1
+ }
+
+ "continue with autoCloseable when Strategy is Resume and exception happened on map" in {
+ val closedCounter = new AtomicInteger(0)
+ val create = () =>
+ new AutoCloseable {
+ override def close(): Unit = closedCounter.incrementAndGet()
+ }
+ val p = Source
+ .fromIterator(() => (0 to 50).iterator)
+ .mapWithResource(create, (_: AutoCloseable, elem) => elem)
+ .map(elem => {
+ if (elem == 10) throw TE("") else elem
Review Comment:
oh~ thanks! β
updated.
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapWithResourceSpec.scala:
##########
@@ -518,6 +518,118 @@ class FlowMapWithResourceSpec extends StreamSpec(UnboundedMailboxConfig) {
closedCounter.get shouldBe 1
}
+ "continue with autoCloseable when Strategy is Resume and exception happened" in {
+ val closedCounter = new AtomicInteger(0)
+ val create = () =>
+ new AutoCloseable {
+ override def close(): Unit = closedCounter.incrementAndGet()
+ }
+ val p = Source
+ .fromIterator(() => (0 to 50).iterator)
+ .mapWithResource(create,
+ (_: AutoCloseable, elem) => {
+ if (elem == 10) throw TE("") else elem
+ })
+ .withAttributes(supervisionStrategy(resumingDecider))
+ .runWith(Sink.asPublisher(false))
+ val c = TestSubscriber.manualProbe[Int]()
+
+ p.subscribe(c)
+ val sub = c.expectSubscription()
+
+ (0 to 48).foreach(i => {
+ sub.request(1)
+ c.expectNext() should ===(if (i < 10) i else i + 1)
+ })
+ sub.request(1)
+ c.expectNext(50)
+ c.expectComplete()
+ closedCounter.get shouldBe 1
+ }
+
+ "continue with autoCloseable when Strategy is Resume and exception happened on map" in {
+ val closedCounter = new AtomicInteger(0)
+ val create = () =>
+ new AutoCloseable {
+ override def close(): Unit = closedCounter.incrementAndGet()
+ }
+ val p = Source
+ .fromIterator(() => (0 to 50).iterator)
+ .mapWithResource(create, (_: AutoCloseable, elem) => elem)
+ .map(elem => {
+ if (elem == 10) throw TE("") else elem
+ })
+ .withAttributes(supervisionStrategy(resumingDecider))
+ .runWith(Sink.asPublisher(false))
+ val c = TestSubscriber.manualProbe[Int]()
+
+ p.subscribe(c)
+ val sub = c.expectSubscription()
+
+ (0 to 48).foreach(i => {
+ sub.request(1)
+ c.expectNext() should ===(if (i < 10) i else i + 1)
+ })
+ sub.request(1)
+ c.expectNext(50)
+ c.expectComplete()
+ closedCounter.get shouldBe 1
+ }
+
+ "close and open stream with autocloseable again when Strategy is Restart" in {
+ val closedCounter = new AtomicInteger(0)
+ val create = () =>
+ new AutoCloseable {
+ override def close(): Unit = closedCounter.incrementAndGet()
+ }
+ val p = Source
+ .fromIterator(() => (0 to 50).iterator)
+ .mapWithResource(create,
+ (_: AutoCloseable, elem) => {
+ if (elem == 10) throw TE("") else elem
+ })
+ .withAttributes(supervisionStrategy(restartingDecider))
Review Comment:
β
updated, nice suggestion!
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "injae-kim (via GitHub)" <gi...@apache.org>.
injae-kim commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1474158630
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala:
##########
@@ -2541,6 +2541,45 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
(resource, out) => f(resource, out),
resource => close.apply(resource).toScala))
+ /**
+ * Transform each stream element with the help of a [[AutoCloseable]] resource and close it when the stream finishes or fails.
Review Comment:
β
updated. oh thanks for catching!
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1473839166
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala:
##########
@@ -828,6 +828,45 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends Gr
(resource, out) => f(resource, out),
resource => close.apply(resource).toScala))
+ /**
+ * Transform each stream element with the help of a [[AutoCloseable]] resource and close it when the stream finishes or fails.
Review Comment:
`an`
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1913488517
Hate to ruin the party, but I would hold our horses on rushing this through as resource management is a very difficult thing to get right on most platforms. While `AuthoClosable` may be fine on Java, Scala users tend to have their own solutions which involve everything from `scala.io.Using` to the solutions we see in Cats/FS2.
I don't want to block progress/experimentation on this but I would be more comfortable looking at this with a fine tooth comb and hence it may be more appropriate for Pekko 1.2.x. We can even set up different branches to help experiment with different ideas as well, I know that @alexandru also had some thoughts on this when the fork was occuring.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1473840214
##########
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala:
##########
@@ -1142,6 +1142,43 @@ trait FlowOps[+Out, +Mat] {
resource => close(resource))
.withAttributes(DefaultAttributes.mapWithResource))
+ /**
+ * Transform each stream element with the help of a [[AutoCloseable]] resource and close it when the stream finishes or fails.
Review Comment:
`an [[AutoCloseable]] `
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468442521
##########
stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java:
##########
@@ -237,6 +238,22 @@ public void mustBeAbleToUseMapWithResource() {
Assert.assertFalse(gate.get());
}
+ @Test
+ public void mustBeAbleToUseMapWithAutoCloseableResource() {
+ final AtomicInteger closed = new AtomicInteger();
+ Source.from(Arrays.asList("1", "2", "3"))
+ .via(
+ Flow.of(String.class)
+ .mapWithResource(
+ () -> (AutoCloseable) closed::incrementAndGet,
+ (resource, elem) -> elem))
+ .runWith(TestSink.create(system), system)
+ .request(4)
+ .expectNext("1", "2", "3")
+ .expectComplete();
+ Assert.assertEquals(closed.get(), 3);
Review Comment:
Should assert the Autoclosable has been closed after the stream is completed.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "injae-kim (via GitHub)" <gi...@apache.org>.
injae-kim commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1913121640
Thanks for quick review! I'm struggling with build failure on my local env,
so I'll update this PR based on your review after build&run test on my local env~! thanks! π
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468442521
##########
stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java:
##########
@@ -237,6 +238,22 @@ public void mustBeAbleToUseMapWithResource() {
Assert.assertFalse(gate.get());
}
+ @Test
+ public void mustBeAbleToUseMapWithAutoCloseableResource() {
+ final AtomicInteger closed = new AtomicInteger();
+ Source.from(Arrays.asList("1", "2", "3"))
+ .via(
+ Flow.of(String.class)
+ .mapWithResource(
+ () -> (AutoCloseable) closed::incrementAndGet,
+ (resource, elem) -> elem))
+ .runWith(TestSink.create(system), system)
+ .request(4)
+ .expectNext("1", "2", "3")
+ .expectComplete();
+ Assert.assertEquals(closed.get(), 3);
Review Comment:
Should assert the Autoclosable is been closed after the stream is completed.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#discussion_r1468508664
##########
stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java:
##########
@@ -237,6 +238,22 @@ public void mustBeAbleToUseMapWithResource() {
Assert.assertFalse(gate.get());
}
+ @Test
+ public void mustBeAbleToUseMapWithAutoCloseableResource() {
+ final TestKit probe = new TestKit(system);
+ final AtomicInteger closed = new AtomicInteger();
+ Source.from(Arrays.asList("1", "2", "3"))
+ .via(
+ Flow.of(String.class)
+ .mapWithResource(
+ () -> (AutoCloseable) closed::incrementAndGet, (resource, elem) -> elem))
+ .runWith(Sink.foreach(elem -> probe.getRef().tell(elem, ActorRef.noSender())), system);
+
+ Assert.assertEquals(closed.get(), 0);
Review Comment:
This will be flaky, maybe we can just remove it.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1913277825
don't make any changes about this (yet) - but what about Closeable? Most resource management frameworks work with both.
--
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
Re: [PR] feat: Add AutoCloseable shortcut on mapWithResource [incubator-pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on PR #1053:
URL: https://github.com/apache/incubator-pekko/pull/1053#issuecomment-1913283588
@injae-kim I would still like at least 1 test when the map function fails with an exception.
--
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