Webflux 文件上传 Minio

Question

我正在尝试弄清楚为什么无法通过 WebFlux 端点将文件上传到 minio（S3 兼容文档存储）；我总是只将 4kb 的文件放入 Minio。

我的终点：

    public Mono<ServerResponse> uploadFile(ServerRequest request) {
        log.info("Uploading file...");
        log.info("Content Type: {}", request.headers().contentType().orElse(MediaType.TEXT_PLAIN));

        return request.body(BodyExtractors.toMultipartData())
            .flatMap(map -> {
                Map<String, Part> parts = map.toSingleValueMap();
                return Mono.just((FilePart) parts.get("files"));
            })
            .flatMap(this::saveFile)
            .flatMap(part -> ServerResponse.ok().body(BodyInserters.fromObject(part)));
    }

    private Mono<String> saveFile(FilePart part) {
        return part.content().map(dataBuffer -> {
            try {
                log.info("Putting file {} to minio...", part.filename());
                client.putObject("files", part.filename(), dataBuffer.asInputStream(), part.headers().getContentType().getType());
            } catch(Exception e) {
                log.error("Error storing file to minio", e);
                return part.filename();
            }
        return part.filename();
        }).next();
    }

我很确定这是一个阻塞与非阻塞问题，但如果我尝试添加一个 blockFirst() 调用，我会收到一个异常，说它在运行时不允许。

有没有办法有效地流式传输数据，或者这是 Minio 客户端与 WebFlux 不兼容的情况？

我正在尝试从 React 组件 post 像这样：

class DataUpload extends React.Component {
    constructor(props) {
        super(props);

        this.state = {
            fileURL: '',
        };

        this.handleUploadData = this.handleUploadData.bind(this);
    }

    handleUploadData = ev => {
        ev.preventDefault()

        const data = new FormData();
        data.append('files', this.uploadInput.files[0]);
        data.append('filename', this.fileName.value);
        fetch('http://localhost:8080/requestor/upload', {
            method: 'POST',
            body: data,
        }).then((response) => {
            response.json().then((body) => {
                this.setState({ fileURL: `http://localhost:8080/${body.file}`});
            });
        });
    }

    render() {
        return (
            <form onSubmit={this.handleUploadData}>
                <div>
                    <input ref={(ref) => { this.uploadInput = ref; }} type="file" />
                </div>
                <div>
                    <input ref={(ref) => { this.fileName = ref; }} type="text" placeholder="Enter the desired name of the file" />
                </div>
                <br/>
                <div><button>Upload</button></div>
            </form>
        );
    }
}

export default DataUpload;

Answer 1

I always only get 4kb of the file into Minio

那是因为spring将文件分成4kb的部分，你必须自己收集它们。你可以这样做：

request.body(BodyExtractors.toMultipartData())
    .map(dataBuffers -> dataBuffers.get("files"))
    .filter(Objects::nonNull)
    //get the file name and pair it with it's "Part"
    .map(partsList -> {
        List<Pair<String, Part>> pairedList = new ArrayList<>();

        for (Part part : partsList) {
            String fileName = ((FilePart) part).filename();
            pairedList.add(new Pair<>(fileName, part));
        }

        return pairedList;
    })
    .flux()
    .flatMap(Flux::fromIterable)
    //here we collect all of the file parts with the buffer operator and zip them with filename
    .flatMap(partWithName -> Mono.zip(Mono.just(partWithName.getFirst()), partWithName.getSecond().content().buffer().single()))
    .buffer()
    .single()
    .doOnNext(filePartsWithNames -> {
        //here we have a list of all uploading file parts and their names
        for (Tuple2<String, List<DataBuffer>> filePartsWithName : filePartsWithNames) {
            String fileName = filePartsWithName.getT1();
            List<DataBuffer> buffers = filePartsWithName.getT2();

            System.out.println("Filename = " + fileName);

            //"buffers" is a list of 4kb chunks of the files
            for (DataBuffer buffer : buffers) {
                System.out.println("Buffer size = " + buffer.readableByteCount());

                //here you can use buffer.asInputStream() to read the file part and
                //then save it on disk or do something else with it
            }
        }
    })

Answer 2

对于未来的读者，

我通过在作为 inputStream 公开之前加入 dataBuffers 来解决这个问题，使用包 org.springframework.core.io.buffer 中的 DataBufferUtils.join()。

   Mono<ServerResponse> uploadSingleImageToS3(ServerRequest request) {
    return request.body(toMultipartData())
            .flatMap(parts -> {
                Map<String, Part> part = parts.toSingleValueMap();
                return Mono.just((FilePart) part.get("file"));
            })
            .flatMap(this::uploadToS3Bucket)
            .flatMap(filename -> ServerResponse.ok().body(fromObject(filename)));
}

private Mono<String> uploadToS3Bucket(FilePart part) {
    return DataBufferUtils.join(part.content())
            .map(dataBuffer -> {
                String filename = UUID.randomUUID().toString();
                log.info("filename : {}", filename);

                ObjectMetadata metadata = new ObjectMetadata();
                metadata.setContentLength(dataBuffer.capacity());
                PutObjectRequest putObjectRequest = new PutObjectRequest(answerImagesBucket, filename, dataBuffer.asInputStream(), metadata);
                transferManager.upload(putObjectRequest);

                return filename;
            });
}

Webflux 文件上传 Minio

Webflux FileUpload Minio

asyncfileupload

reactjs

minio

spring-webflux