scala - 如何流式传输调度程序的响应？

Question

我必须导出 CSV 数据。数据量非常大。所以我正在流式传输来自微服务的响应。我们使用调度程序访问我们的微服务。

def stream(method: String, urlString: String): Future[Source[ByteString, NotUsed]] =
    method match {
      case GET    =>
        val request = Http(url(urlString))
        request.map { response =>
          response.getStatusCode match {
            case StatusOk => Source.single(ByteString(response.getResponseBody))
          }
        }
    }

它将带来所有数据。所以为了解决这个问题，我喜欢修改它并从这里传输数据。

我搜索了很多，发现这个问题Scala dispatch stream response line by line

但它没有答案。

谢谢，任何帮助将不胜感激。

Answer 1

经过大量搜索，我将其读取为输入流并转换为 Akka 流。它对我有用。

def stream(method: String, urlString: String): Future[Source[ByteString, Future[IOResult]]] =
    method match {
      case GET    =>
        val futureStream = Http(url(urlString) > as.Response(_.getResponseBodyAsStream))
        futureStream.map { inputStream =>
          val source = () => inputStream
          StreamConverters.fromInputStream(source)
        }
    }