pekko/akka-docs/src/main/paradox/stream/operators/Source/unfoldResourceAsync.md

Source.unfoldResourceAsync

Wrap any resource that can be opened, queried for next element and closed in an asynchronous way.

@refSource operators

Signature

@apidocSource.unfoldResourceAsync { scala="#unfoldResourceAsyncT,S:akka.stream.scaladsl.Source[T,akka.NotUsed]" java="#unfoldResourceAsync(akka.japi.function.Creator,akka.japi.function.Function,akka.japi.function.Function)" }

Description

Wrap any resource that can be opened, queried for next element and closed in an asynchronous way with three distinct functions into a source. This operator is the equivalent of @refunfoldResource but for resources with asynchronous APIs.

Source.unfoldResourceAsync allows us to safely extract stream elements from a resource with an async API by providing it with three functions that all return a @scala[Future]@java[CompletionStage]:

1. create: Open or create the resource
2. read: Fetch the next element or signal that we reached the end of the stream by completing the @scala[Future]@java[CompletionStage] with a @java[Optional.empty]@scala[None]
3. close: Close the resource, invoked on end of stream or if the stream fails

All exceptions thrown by create and close as well as the @scala[Future]@java[CompletionStage]s completing with failure will fail the stream. The supervision strategy is used to handle exceptions from read, create and from the @scala[Future]@java[CompletionStage]s.

Note that there are pre-built unfoldResourceAsync-like operators to wrap java.io.InputStreams in @ref:Additional Sink and Source converters, Iterator in @ref:fromIterator and File IO in @ref:File IO Sinks and Sources. Additional prebuilt technology specific connectors can also be found in the Alpakka project.

Examples

Imagine we have an async database API which we initially perform an async query and then can check if there are more results in an asynchronous way.

Scala

@@snip UnfoldResourceAsync.scala { #unfoldResource-async-api }

Java

@@snip UnfoldResourceAsync.java { #unfoldResource-async-api }

Let's see how we use the API above safely through unfoldResourceAsync:

Scala

@@snip UnfoldResourceAsync.scala { #unfoldResourceAsync }

Java

@@snip UnfoldResource.java { #unfoldResourceAsync }

Reactive Streams semantics

@@@div { .callout }

emits when there is demand and @scala[Future] @java[CompletionStage] from read function returns value

completes when @scala[Future] @java[CompletionStage] from read function returns None

@@@