He-Pin commented on code in PR #931:
URL: https://github.com/apache/incubator-pekko/pull/931#discussion_r1452150340
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala:
##########
@@ -770,6 +770,47 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In,
Out, Mat]) extends Gr
(s: S, out: Out) => f.apply(s, out).toScala,
(s: S) => onComplete.apply(s).toScala))
+ /**
+ * Transform each stream element with the help of a resource.
+ *
+ * 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 `close` function is called only once when the upstream or downstream
finishes or fails. You can do some clean-up here,
+ * and if the returned value is not empty, it will be emitted to the
downstream if available, otherwise the value will be dropped.
+ *
+ * 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
+ * set it for a given Source by using [[ActorAttributes]].
+ *
+ * '''Emits when''' the mapping function returns an element and downstream
is ready to consume it
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * @tparam R the type of the resource
+ * @tparam T the type of the output elements
+ * @param create function that creates the resource
+ * @param f function that transforms the upstream element and the resource
to output element
+ * @param close function that closes the resource, optionally outputting a
last element
+ * @since 1.1.0
+ */
+ def mapWithResource[R, T](
+ create: java.util.function.Supplier[R],
Review Comment:
eg: fs2's `Stream#bracket` where `def bracket[F[_], R](acquire:
F[R])(release: R => F[Unit]): Stream[F, R]` the `F` is not limited to `Try`
too. so does the `Resource.make`.
And if a use change the checked exception to a runtime exception, the flow
will fails too, eg: using `@SneakyThrows` .
--
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-unsubscr...@pekko.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscr...@pekko.apache.org
For additional commands, e-mail: notifications-h...@pekko.apache.org
