从 Micronaut 调用文件上传 API

Question

我需要调用 API 并使用 Micronaut 上传文件字节数据

这与我们在 OkHTTP 中使用 do 类似：

Response uploadResult = okHttpClient.newCall(new Request.Builder()
                .url(uploadUrl)
                .post(RequestBody.create(byteData))
                .build()
        ).execute();

我尝试使用 micronaut HTTP 客户端添加传递 Multipart 表单数据来调用 API

MultipartBody requestBody = MultipartBody.builder().addPart("data", uploadFile.getFilename(), fileContentBytes).build();
        httpClient = HttpClient.create(new URL(uploadUrl.getUpload_url()));
        HttpRequest<?> uploadRequest = HttpRequest.POST(uploadUrl.getUpload_url(), requestBody)
                .contentType(MediaType.MULTIPART_FORM_DATA_TYPE)
                .header(HttpHeaders.AUTHORIZATION, v2Request.getToken());
        Publisher<?> uploadResponse = httpClient.exchange(uploadRequest, HttpResponse.class);

我只是不知道如何在上传完成后找到该事件并继续！

---

我尝试使用 Flux 订阅发布者，如下

 HttpClient httpClient = HttpClient.create(new URL(url));
            HttpRequest<FbPagePostRequest> request = HttpRequest.POST(url, postRequest).contentType(MediaType.APPLICATION_JSON).header(HttpHeaders.AUTHORIZATION, token);
            Publisher<HttpResponse<FbPagePostResponse>> response = httpClient.exchange(request, FbPagePostResponse.class);
            Flux.from(response).map(data -> {
                if (data.status() != HttpStatus.OK || data.getBody().isEmpty()) {
                    log.error("Response of facebook page post " + response);
                    throw new PublicationException("Error posting on Facebook on Page : " + pageName);
                }
                return data.getBody().get();
            }).subscribe(returnObj::set);

但是当我运行这个时，我得到了错误

Response of facebook page post FluxLift

Answer 1

您需要将

subscribe

添加到返回的

Publisher

，如此处

所述

subscribe(); // (1)

subscribe(Consumer<? super T> consumer);  // (2)

subscribe(Consumer<? super T> consumer,
          Consumer<? super Throwable> errorConsumer); // (3)

subscribe(Consumer<? super T> consumer,
          Consumer<? super Throwable> errorConsumer,
          Runnable completeConsumer); // (4)

subscribe(Consumer<? super T> consumer,
          Consumer<? super Throwable> errorConsumer,
          Runnable completeConsumer,
          Consumer<? super Subscription> subscriptionConsumer); // (5)

1. 订阅并触发序列。
2. 对每个产生的价值做一些事情。
3. 处理值，但也要对错误做出反应。
4. 处理值和错误，但在序列成功完成时运行一些代码。
5. 处理值和错误并成功完成，但也对此订阅调用生成的订阅执行某些操作。