passing an Akka stream to an upstream service to populate

I need to call an upstream service (Azure Blob Service) to push data to an OutputStream, which then i need to turn around and push it back to the client, thru akka. Without akka (and just servlet code), i'd just get the ServletOutputStream and pass it to the azure service's method.

The closest i can try to stumble upon, and clearly this is wrong, is something like this

        Source<ByteString, OutputStream> source = StreamConverters.asOutputStream().mapMaterializedValue(os -> {
            blobClient.download(os);
            return os;
        });

        ResponseEntity resposeEntity = HttpEntities.create(ContentTypes.APPLICATION_OCTET_STREAM, preAuthData.getFileSize(), source);

        sender().tell(new RequestResult(resposeEntity, StatusCodes.OK), self());

The idea is i'm calling an upstream service to get an outputstream populated by calling
blobClient.download(os);

It seems like the the lambda function gets called and returns, but then afterwards it fails, because there's no data or something. As if i'm not supposed to be have that lambda function do the work, but perhaps return some object that does the work? Not sure.

How does one do this?

The real issue here is that the Azure API is not designed for back-pressuring. There is no way for the output stream to signal back to Azure that it is not ready for more data. To put it another way: if Azure pushes data faster than you are able to consume it, there will have to be some ugly buffer overflow failure somewhere.

Accepting this fact, the next best thing we can do is:

• Use Source.lazySource to only start downloading data when there is downstream demand (aka. the source is being run and data is being requested).
• Put the download call in some other thread so that it continues executing without blocking the source from being returned. Once way to do this is with a Future (I'm not sure what Java best practices are, but should work fine either way). Although it won't matter initially, you may need to choose an execution context other than system.dispatcher - it all depends on whether download is blocking or not.

I apologize in advance if this Java code is malformed - I use Akka with Scala, so this is all from looking at the Akka Java API and Java syntax reference.

ResponseEntity responseEntity = HttpEntities.create(
  ContentTypes.APPLICATION_OCTET_STREAM,
  preAuthData.getFileSize(),

  // Wait until there is downstream demand to intialize the source...
  Source.lazySource(() -> {
    // Pre-materialize the outputstream before the source starts running
    Pair<OutputStream, Source<ByteString, NotUsed>> pair =
      StreamConverters.asOutputStream().preMaterialize(system);

    // Start writing into the download stream in a separate thread
    Futures.future(() -> { blobClient.download(pair.first()); return pair.first(); }, system.getDispatcher());

    // Return the source - it should start running since `lazySource` indicated demand
    return pair.second();
  })
);

sender().tell(new RequestResult(responseEntity, StatusCodes.OK), self());

</details>



# 答案2
**得分**: 0

```java
在这种情况下，`OutputStream` 是 `Source` 的“物化值”，只有在流被运行（或“物化”为正在运行的流）时才会被创建。运行它是不在你的控制范围内的，因为你将 `Source` 交给了 Akka HTTP，而后者将在以后实际运行你的源。

`.mapMaterializedValue(matval -&gt; ...)` 通常用于转换物化值，但由于它作为物化的一部分被调用，你可以在其中执行诸如将 matval 发送到消息中的副作用，就像你已经找出的那样，即使看起来有些奇怪，这并没有必然的问题。重要的是要理解在 lambda 完成之前，流不会完成其物化并开始运行。这意味着如果 `download()` 方法是阻塞的，而不是在不同的线程上启动一些工作并立即返回，则会出现问题。

然而，还有另一种解决方案：`Source.preMaterialize()`，它将源物化并为你提供一个物化值的 `Pair`，以及一个可以用来消耗已经启动的源的新 `Source`：

```java
Pair&lt;OutputStream, Source&lt;ByteString, NotUsed&gt;&gt; pair = 
  StreamConverters.asOutputStream().preMaterialize(system);
OutputStream os = pair.first();
Source&lt;ByteString, NotUsed&gt; source = pair.second();

请注意，在你的代码中还有一些其他要考虑的事项，最重要的是，如果 blobClient.download(os) 调用会阻塞直到完成，并且你从 actor 中调用它，在这种情况下，你必须确保你的 actor 不会饿死分发器，并阻止应用程序中的其他 actor 执行（请参阅 Akka 文档：https://doc.akka.io/docs/akka/current/typed/dispatchers.html#blocking-needs-careful-management）。

<details>
<summary>英文:</summary>
The `OutputStream` in this case is the &quot;materialized value&quot; of the `Source` and it will only be created once the stream is run (or &quot;materialized&quot; into a running stream). Running it is out of your control since you hand the `Source` to Akka HTTP and that will later actually run your source.
`.mapMaterializedValue(matval -&gt; ...)` is usually used to transform the materialized value but since it is invoked as a part of materialization you can use that to do side effects such as sending the matval in a message, just like you have figured out, there isn&#39;t necessarily anything wrong with that even if it looks funky. It is important to understand that the stream will not complete its materialization and become running until that lambda completes. This means problems if `download()` is blocking rather than forking off some work on a different thread and immediately returning.
There is however another solution: `Source.preMaterialize()`, it materializes the source and gives you a `Pair` of the materialized value and a new `Source` that can be used to consume the already started source:
```java
Pair&lt;OutputStream, Source&lt;ByteString, NotUsed&gt;&gt; pair = 
StreamConverters.asOutputStream().preMaterialize(system);
OutputStream os = pair.first();
Source&lt;ByteString, NotUsed&gt; source = pair.second();

Note that there are a few additional things to think of in your code, most importantly if the blobClient.download(os) call blocks until it is done and you call that from the actor, in that case you must make sure that your actor does not starve the dispatcher and stop other actors in your application from executing (see Akka docs: https://doc.akka.io/docs/akka/current/typed/dispatchers.html#blocking-needs-careful-management ).