java使用声明性客户端Micronaut将多个文件上载到MultipartBody

Question

正在尝试使用Micronaut声明性客户端将多个文件上载到

Http客户端

@Client("http://localhost:8080/product")
public interface IHttpClient {
    @Post(consumes = MediaType.TEXT_PLAIN, produces = MediaType.MULTIPART_FORM_DATA)
    public String post(@Body MultipartBody file);
}

向多部分添加多个文件

@Controller("/productManager")
public class ProductManagerController implements IProductOperation{
    private final IHttpClient iProduct;

    public ProductManagerController(IHttpClient iProduct) {
        this.iProduct = iProduct;
    }

    @Override
    public String post(CompletedFileUpload file) throws IOException {
        MultipartBody requestBody = MultipartBody.builder()
                .addPart("file", file.getFilename(), MediaType.MULTIPART_FORM_DATA_TYPE, file.getBytes())
                .addPart("file", file.getFilename(), MediaType.MULTIPART_FORM_DATA_TYPE, file.getBytes())
                .addPart("id", "asdasdsds")
                .build();
        return this.iProduct.post(requestBody);
    }
}

产品控制器

@Controller("/product")
public class ProductController implements IProductOperation {
    @Post(consumes = MediaType.MULTIPART_FORM_DATA, produces = MediaType.TEXT_PLAIN)
    public String post(Publisher<CompletedFileUpload> file, String id)  {
        return null;
    }
}

例外情况

io.netty.util.IllegalReferenceCountException: refCnt: 0, decrement: 1
    at io.netty.util.internal.ReferenceCountUpdater.toLiveRealRefCnt(ReferenceCountUpdater.java:74)
    at io.netty.util.internal.ReferenceCountUpdater.release(ReferenceCountUpdater.java:138)
    at io.netty.buffer.AbstractReferenceCountedByteBuf.release(AbstractReferenceCountedByteBuf.java:100)
    at io.micronaut.http.server.netty.multipart.NettyCompletedFileUpload.getBytes(NettyCompletedFileUpload.java:108)
    at com.example.ProductManagerController.post(ProductManagerController.java:26)
    at com.example.$ProductManagerControllerDefinition$$exec1.invokeInternal(Unknown Source)
    at io.micronaut.context.AbstractExecutableMethod.invoke(AbstractExecutableMethod.java:146)
    at io.micronaut.context.DefaultBeanContext$4.invoke(DefaultBeanContext.java:474)
    at io.micronaut.web.router.AbstractRouteMatch.execute(AbstractRouteMatch.java:312)
    at io.micronaut.web.router.RouteMatch.execute(RouteMatch.java:118)
    at io.micronaut.http.server.netty.RoutingInBoundHandler.lambda$buildResultEmitter$12(RoutingInBoundHandler.java:1394)
    at io.reactivex.internal.operators.flowable.FlowableDefer.subscribeActual(FlowableDefer.java:35)
    at io.reactivex.Flowable.subscribe(Flowable.java:14918)
    at io.reactivex.Flowable.subscribe(Flowable.java:14865)
    at io.micronaut.reactive.rxjava2.RxInstrumentedFlowable.subscribeActual(RxInstrumentedFlowable.java:57)
    at io.reactivex.Flowable.subscribe(Flowable.java:14918)
    at io.reactivex.Flowable.subscribe(Flowable.java:14865)
    at io.reactivex.internal.operators.flowable.FlowableDefer.subscribeActual(FlowableDefer.java:42)
    at io.reactivex.Flowable.subscribe(Flowable.java:14918)
    at io.reactivex.Flowable.subscribe(Flowable.java:14865)
    at io.micronaut.reactive.rxjava2.RxInstrumentedFlowable.subscribeActual(RxInstrumentedFlowable.java:57)
    at io.reactivex.Flowable.subscribe(Flowable.java:14918)
    at io.reactivex.Flowable.subscribe(Flowable.java:14865)
    at io.micronaut.http.server.context.ServerRequestContextFilter.lambda$doFilter$0(ServerRequestContextFilter.java:62)
    at io.reactivex.internal.operators.flowable.FlowableFromPublisher.subscribeActual(FlowableFromPublisher.java:29)
    at io.reactivex.Flowable.subscribe(Flowable.java:14918)
    at io.reactivex.Flowable.subscribe(Flowable.java:14868)
    at io.micronaut.http.server.netty.RoutingInBoundHandler.lambda$buildExecutableRoute$8(RoutingInBoundHandler.java:1099)
    at io.micronaut.web.router.DefaultUriRouteMatch$1.execute(DefaultUriRouteMatch.java:80)
    at io.micronaut.web.router.RouteMatch.execute(RouteMatch.java:118)
    at io.micronaut.http.server.netty.RoutingInBoundHandler$1.executeRoute(RoutingInBoundHandler.java:979)
    at io.micronaut.http.server.netty.RoutingInBoundHandler$1.doOnNext(RoutingInBoundHandler.java:927)
    at io.micronaut.core.async.subscriber.CompletionAwareSubscriber.onNext(CompletionAwareSubscriber.java:52)
    at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
    at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
    at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1625)
    at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484)
    at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474)
    at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
    at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
    at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
    at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:497)
    at io.micronaut.http.server.netty.FormDataHttpContentProcessor.onData(FormDataHttpContentProcessor.java:161)
    at io.micronaut.http.server.netty.AbstractHttpContentProcessor.doOnNext(AbstractHttpContentProcessor.java:78)
    at io.micronaut.http.server.netty.AbstractHttpContentProcessor.doOnNext(AbstractHttpContentProcessor.java:36)
    at io.micronaut.core.async.subscriber.CompletionAwareSubscriber.onNext(CompletionAwareSubscriber.java:52)
    at io.micronaut.http.netty.reactive.HandlerPublisher.publishMessage(HandlerPublisher.java:378)
    at io.micronaut.http.netty.reactive.HandlerPublisher.channelRead(HandlerPublisher.java:334)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    at io.micronaut.http.netty.stream.HttpStreamsHandler.handleReadHttpContent(HttpStreamsHandler.java:292)
    at io.micronaut.http.netty.stream.HttpStreamsHandler.channelRead(HttpStreamsHandler.java:257)
    at io.micronaut.http.netty.stream.HttpStreamsServerHandler.channelRead(HttpStreamsServerHandler.java:121)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
    at io.netty.handler.codec.MessageToMessageCodec.channelRead(MessageToMessageCodec.java:111)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:93)
    at io.netty.handler.codec.http.HttpServerKeepAliveHandler.channelRead(HttpServerKeepAliveHandler.java:64)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
    at io.netty.handler.flow.FlowControlHandler.dequeue(FlowControlHandler.java:200)
    at io.netty.handler.flow.FlowControlHandler.read(FlowControlHandler.java:139)
    at io.netty.channel.AbstractChannelHandlerContext.invokeRead(AbstractChannelHandlerContext.java:686)
    at io.netty.channel.AbstractChannelHandlerContext.read(AbstractChannelHandlerContext.java:671)
    at io.micronaut.http.netty.reactive.HandlerPublisher.requestDemand(HandlerPublisher.java:163)
    at io.micronaut.http.netty.stream.HttpStreamsHandler$2.requestDemand(HttpStreamsHandler.java:248)
    at io.micronaut.http.netty.reactive.HandlerPublisher$ChannelSubscription.receivedDemand(HandlerPublisher.java:547)
    at io.micronaut.http.netty.reactive.HandlerPublisher$ChannelSubscription.lambda$request$0(HandlerPublisher.java:474)
    at io.netty.util.concurrent.AbstractEventExecutor.safeExecute$$$capture(AbstractEventExecutor.java:164)
    at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java)
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:497)
    at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
    at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)

Answer 1

在做了一些研究之后，我们可以使用Micronaut声明性客户端实现上传多个文件，如下所示

客户

确保客户机中MultipartBody的导入来自io.micronaut.http.client.multipart.MultipartBody;

    @Client(id="feteBirdProduct", path = "/product")
    public interface IProductClient extends IProductOperation {
        @Post(consumes = MediaType.APPLICATION_JSON, produces = MediaType.MULTIPART_FORM_DATA)
        Maybe<HttpResponse<?>> post(@Body MultipartBody file);
    }

API网关控制器

public Maybe<HttpResponse<?>> post(Publisher<CompletedFileUpload> files) {
        return Flowable.fromPublisher(files)
                .collect(MultipartBody::builder, (requestBody, file) -> {
                    requestBody
                            .addPart("file", file.getFilename(), MediaType.MULTIPART_FORM_DATA_TYPE, file.getBytes());
                })
                .flatMapMaybe(requestBody -> {
                    requestBody.addPart("id", "asdasdsds");
                    return iProductClient.post(requestBody.build());
                });
    }

消费者控制器

    @Post(consumes = MediaType.MULTIPART_FORM_DATA, produces = MediaType.APPLICATION_JSON)
    public Maybe<HttpResponse<?>> post(Publisher<CompletedFileUpload> file,String id) {
        Flowable.fromPublisher(file).subscribe(item -> {
            System.out.println("success");
        });
        return Maybe.just(HttpResponse.accepted());
    }