Quarkus通过反应性REST API发送/接收“巨大”溪流

Question

我为自己的概念证明而苦苦挣扎。 这是主要想法：

• 在一侧，REST API服务器应允许
• 在另一侧获取并发送“流”（作为InputStream，而不是文件），客户端应像InPustream一样进行相同的操作（可能是/从文件中，但是拥有一个输入流将回答所有情况）

首先，我知道Quarkus反应式框架允许发送异步，但这不是一个选项：

• 然后将内容（请求/响应）转发到/从另一个服务中获取（对象存储）
• 不再提交文件，因为它是一种微服务，即使在此服务中暂时存储大量文件是反对体系结构逻辑的，
• 因此当然不是出于明显的原因（不允许Microservice拥有1 tb的RAM），

我尝试为了找出如何做到这一点，当然要考虑到后压力，以避免任何方式（或几乎）在内存中（无论是client＆lt; - ＆gt; server）。 我的主要问题是目前在将InputStream发送回客户端以进行响应时与服务器端。

因此，我的主要问题是：如何通过REST HTTP Service Server Side发送一个巨大的流，而无需将所有内容放入内存或使用Quarkus的文件中？？ 然后，下一个问题将是：

• 相同但在客户端（可能是通过网络定位和管道，但是很难看如何做，因为我有一个输入式的响应，并且想消耗一个输入流，而不是writestream） ？
• 最后在Re​​vese Way上（接收服务器端的输入流，由客户端发送）？

到目前为止，我已经尝试了几种方法：

• 使用 inputStream 的直接发送/接收，但是
• 使用 inputStream （使用 uni ）但是，
• 使用反应性发送/接收 Multi＆lt; buff＆gt; 或 multi＆lt; byte []＆gt; ，但
• 试图通过请求实现后压力的 相同问题但是没有解决（我相信HTTP不是HTTP/2）
• 来弄清楚如何使用Mutiny使用 Overflow ，但没有找到该如何做（当然不要掉落缓冲区
• ）要创建一个asyncinputStream，请从vert.x和mutiny创建 asyncfile ，但无法使其
• 在服务器和客户端定义上具有类似的效果（因此使用Resteasy Mutiny Way，用于服务器和客户端）（请注意，请注意客户端，我也尝试了Vert.x Web客户端，但主要问题在服务器端）。

@Path("/objects/{name}"}
@POST
@Consumes(MediaType.APPLICATION_OCTET_STREAM)
public Uni<MyResponse> createObject(@RestPath String name, InputStream content) {
   // For the moment not the main issue, but I believed the InpuStream rely in memory, which is very bad
}

@Path("/objects/{name}"}
@GET
@Produces(MediaType.APPLICATION_OCTET_STREAM)
public Uni<InputStream> getObject1(@RestPath String name) {
  // OOME
}

@Path("/objects2/{name}"}
@GET
@Produces(MediaType.APPLICATION_OCTET_STREAM)
public Multi<Buffer> getObject2(@RestPath String name) {
  // OOME or Blocked
}

@Path("/objects2/{name}"}
@GET
@Produces(MediaType.APPLICATION_OCTET_STREAM)
public Multi<byte[]> getObject3(@RestPath String name) {
  // OOME or Blocked
}

因此，重点关注服务器端，我尝试过：（伪代码）

• 使用直接输入流：限制为10 MB（quarkus默认值），并且在内存中，上面是错误的（InputStream在客户端上是无效的，并且如果服务器发送了，则没有线索一切都不是在响应时间）。
• 没有反向压力：带有

Multi.createFrom().emitter(em -> {
  // Functional interface as "sendAsRequested"
  while (InputStream as something) {
     byte[] bufferBytes = read from InpuStream
     if (read ended) {
        em.complete();
        break;
     }
     Buffer buffer = Buffer.buffer().appendBytes(bufferBytes);
     em.emit(buffer);    
  }
}).onCompletion().invoke(() -> {
    Close the InputStream
});

• 后压的内存：但是，即使在客户端上，我都不会回到服务器，我发送了我发送 request（1）

Multi.createFrom().emitter(em -> {
  // Functional interface as "sendAsRequested"
  long nb = em.requested();
  for (loop on nb) {
     byte[] bufferBytes = read from InpuStream
     if (read ended) {
        em.complete();
        break;
     }
     Buffer buffer = Buffer.buffer().appendBytes(bufferBytes);
     em.emit(buffer);    
  }
}).onRequest().invoke(() -> {
    // will relaunch the functional interface "sendAsRequested"
}).onCompletion().invoke(() -> {
    Close the InputStream
});

和客户端的每个项目：（似乎请求从未到达服务器）

Multi<Buffer> received;
received.subscribe().withSubcriber(new MultiSubscriber<Buffer>() {
  Subscription s;
  @Override
  public void onSubscribe(Subscription s) {
    this.s = s;
    s.request(1);
  }
  @Override
  public void onItem(Buffer buffer) {
    s.request(1);
    consume buffer
  }
});

• 查看 asyncfile 实现以编写一些“ asyncinputstream”，但是
  • 似乎当前的兵变实现不允许添加新的“类型”（各种类型的硬编码）
  • 试图将异步范围扩展到“模仿”其行为的行为，但不确定我是否会朝着正确的方向。

我有点好奇默认的Quarkus将10 MB表示为身体的限制（身体块不受限制）。似乎所有努力都将用于小型或小流（如JSON方面），但没有inputStream之类的流（在文件除外，但在传统的容器微服务方式中不可能，而内存和磁盘太小了对于这样的文件方式）。我了解原因（占需求的80％），但似乎Inputstream目前在Quarkus中缺乏支持（除非少于10 MB）。

有人有任何想法或建议吗？