diff --git a/spring-web/src/main/java/org/springframework/http/codec/multipart/Content.java b/spring-web/src/main/java/org/springframework/http/codec/multipart/Content.java deleted file mode 100644 index 28178c8c8e..0000000000 --- a/spring-web/src/main/java/org/springframework/http/codec/multipart/Content.java +++ /dev/null @@ -1,129 +0,0 @@ -/* - * Copyright 2002-2021 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.http.codec.multipart; - -import java.io.IOException; -import java.io.UncheckedIOException; -import java.nio.file.Files; -import java.nio.file.Path; -import java.nio.file.StandardOpenOption; - -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.core.scheduler.Scheduler; - -import org.springframework.core.io.buffer.DataBuffer; -import org.springframework.core.io.buffer.DataBufferUtils; -import org.springframework.core.io.buffer.DefaultDataBufferFactory; - -/** - * Part content abstraction used by {@link DefaultParts}. - * - * @author Arjen Poutsma - * @since 5.3.13 - */ -abstract class Content { - - - protected Content() { - } - - /** - * Return the content. - */ - public abstract Flux content(); - - /** - * Delete this content. Default implementation does nothing. - */ - public Mono delete() { - return Mono.empty(); - } - - /** - * Returns a new {@code Content} based on the given flux of data buffers. - */ - public static Content fromFlux(Flux content) { - return new FluxContent(content); - } - - /** - * Return a new {@code Content} based on the given file path. - */ - public static Content fromFile(Path file, Scheduler scheduler) { - return new FileContent(file, scheduler); - } - - - /** - * {@code Content} implementation based on a flux of data buffers. - */ - private static final class FluxContent extends Content { - - private final Flux content; - - - public FluxContent(Flux content) { - this.content = content; - } - - - @Override - public Flux content() { - return this.content; - } - } - - - /** - * {@code Content} implementation based on a file. - */ - private static final class FileContent extends Content { - - private final Path file; - - private final Scheduler scheduler; - - - public FileContent(Path file, Scheduler scheduler) { - this.file = file; - this.scheduler = scheduler; - } - - - @Override - public Flux content() { - return DataBufferUtils.readByteChannel( - () -> Files.newByteChannel(this.file, StandardOpenOption.READ), - DefaultDataBufferFactory.sharedInstance, 1024) - .subscribeOn(this.scheduler); - } - - @Override - public Mono delete() { - return Mono.fromRunnable(() -> { - try { - Files.delete(this.file); - } - catch (IOException ex) { - throw new UncheckedIOException(ex); - } - }) - .subscribeOn(this.scheduler); - } - } -} diff --git a/spring-web/src/main/java/org/springframework/http/codec/multipart/DefaultParts.java b/spring-web/src/main/java/org/springframework/http/codec/multipart/DefaultParts.java index f18ec157b9..284c82497b 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/multipart/DefaultParts.java +++ b/spring-web/src/main/java/org/springframework/http/codec/multipart/DefaultParts.java @@ -16,10 +16,15 @@ package org.springframework.http.codec.multipart; +import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.StandardCopyOption; +import java.nio.file.StandardOpenOption; +import java.util.concurrent.Callable; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.core.scheduler.Scheduler; import org.springframework.core.io.buffer.DataBuffer; import org.springframework.core.io.buffer.DataBufferUtils; @@ -50,17 +55,40 @@ abstract class DefaultParts { } /** - * Create a new {@link Part} or {@link FilePart} with the given parameters. + * Create a new {@link Part} or {@link FilePart} based on a flux of data + * buffers. Returns {@link FilePart} if the {@code Content-Disposition} of + * the given headers contains a filename, or a "normal" {@link Part} + * otherwise. + * @param headers the part headers + * @param dataBuffers the content of the part + * @return {@link Part} or {@link FilePart}, depending on {@link HttpHeaders#getContentDisposition()} + */ + public static Part part(HttpHeaders headers, Flux dataBuffers) { + Assert.notNull(headers, "Headers must not be null"); + Assert.notNull(dataBuffers, "DataBuffers must not be null"); + + return partInternal(headers, new FluxContent(dataBuffers)); + } + + /** + * Create a new {@link Part} or {@link FilePart} based on the given file. * Returns {@link FilePart} if the {@code Content-Disposition} of the given * headers contains a filename, or a "normal" {@link Part} otherwise * @param headers the part headers - * @param content the content of the part + * @param file the file + * @param scheduler the scheduler used for reading the file * @return {@link Part} or {@link FilePart}, depending on {@link HttpHeaders#getContentDisposition()} */ - public static Part part(HttpHeaders headers, Content content) { + public static Part part(HttpHeaders headers, Path file, Scheduler scheduler) { Assert.notNull(headers, "Headers must not be null"); - Assert.notNull(content, "Content must not be null"); + Assert.notNull(file, "File must not be null"); + Assert.notNull(scheduler, "Scheduler must not be null"); + return partInternal(headers, new FileContent(file, scheduler)); + } + + + private static Part partInternal(HttpHeaders headers, Content content) { String filename = headers.getContentDisposition().getFilename(); if (filename != null) { return new DefaultFilePart(headers, content); @@ -142,7 +170,8 @@ abstract class DefaultParts { */ private static class DefaultPart extends AbstractPart { - private final Content content; + protected final Content content; + public DefaultPart(HttpHeaders headers, Content content) { super(headers); @@ -191,7 +220,7 @@ abstract class DefaultParts { @Override public Mono transferTo(Path dest) { - return DataBufferUtils.write(content(), dest); + return this.content.transferTo(dest); } @Override @@ -200,7 +229,7 @@ abstract class DefaultParts { String name = contentDisposition.getName(); String filename = contentDisposition.getFilename(); if (name != null) { - return "DefaultFilePart{" + name() + " (" + filename + ")}"; + return "DefaultFilePart{" + name + " (" + filename + ")}"; } else { return "DefaultFilePart{(" + filename + ")}"; @@ -209,4 +238,100 @@ abstract class DefaultParts { } + + /** + * Part content abstraction. + */ + private interface Content { + + Flux content(); + + Mono transferTo(Path dest); + + Mono delete(); + + } + + /** + * {@code Content} implementation based on a flux of data buffers. + */ + private static final class FluxContent implements Content { + + private final Flux content; + + + public FluxContent(Flux content) { + this.content = content; + } + + + @Override + public Flux content() { + return this.content; + } + + @Override + public Mono transferTo(Path dest) { + return DataBufferUtils.write(this.content, dest); + } + + @Override + public Mono delete() { + return Mono.empty(); + } + + } + + + /** + * {@code Content} implementation based on a file. + */ + private static final class FileContent implements Content { + + private final Path file; + + private final Scheduler scheduler; + + + public FileContent(Path file, Scheduler scheduler) { + this.file = file; + this.scheduler = scheduler; + } + + + @Override + public Flux content() { + return DataBufferUtils.readByteChannel( + () -> Files.newByteChannel(this.file, StandardOpenOption.READ), + DefaultDataBufferFactory.sharedInstance, 1024) + .subscribeOn(this.scheduler); + } + + @Override + public Mono transferTo(Path dest) { + return blockingOperation(() -> Files.copy(this.file, dest, StandardCopyOption.REPLACE_EXISTING)); + } + + @Override + public Mono delete() { + return blockingOperation(() -> { + Files.delete(this.file); + return null; + }); + } + + private Mono blockingOperation(Callable callable) { + return Mono.create(sink -> { + try { + callable.call(); + sink.success(); + } + catch (Exception ex) { + sink.error(ex); + } + }) + .subscribeOn(this.scheduler); + } + } + } diff --git a/spring-web/src/main/java/org/springframework/http/codec/multipart/PartGenerator.java b/spring-web/src/main/java/org/springframework/http/codec/multipart/PartGenerator.java index 42b43e2eb5..88d689d90e 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/multipart/PartGenerator.java +++ b/spring-web/src/main/java/org/springframework/http/codec/multipart/PartGenerator.java @@ -160,7 +160,7 @@ final class PartGenerator extends BaseSubscriber { requestToken(); } }); - emitPart(DefaultParts.part(headers, Content.fromFlux(streamingContent))); + emitPart(DefaultParts.part(headers, streamingContent)); } } @@ -518,7 +518,7 @@ final class PartGenerator extends BaseSubscriber { } this.content.clear(); Flux content = Flux.just(DefaultDataBufferFactory.sharedInstance.wrap(bytes)); - emitPart(DefaultParts.part(this.headers, Content.fromFlux(content))); + emitPart(DefaultParts.part(this.headers, content)); } @Override @@ -674,8 +674,7 @@ final class PartGenerator extends BaseSubscriber { @Override public void partComplete(boolean finalPart) { MultipartUtils.closeChannel(this.channel); - emitPart(DefaultParts.part(this.headers, - Content.fromFile(this.file, PartGenerator.this.blockingOperationScheduler))); + emitPart(DefaultParts.part(this.headers, this.file, PartGenerator.this.blockingOperationScheduler)); if (finalPart) { emitComplete(); }