Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refactor Contents to DefaultParts

Browse Source 
This commit moves the Contents abstraction into DefaultParts

See gh-27613
This commit is contained in:
Arjen Poutsma
2021-11-03 15:02:12 +01:00
parent 694db2273f
commit 0c7e000250
3 changed files with 135 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

129
spring-web/src/main/java/org/springframework/http/codec/multipart/Content.java
View File

@@ -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<DataBuffer> content();
/**
* Delete this content. Default implementation does nothing.
*/
public Mono<Void> delete() {
return Mono.empty();
}
/**
* Returns a new {@code Content} based on the given flux of data buffers.
*/
public static Content fromFlux(Flux<DataBuffer> 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<DataBuffer> content;
public FluxContent(Flux<DataBuffer> content) {
this.content = content;
}
@Override
public Flux<DataBuffer> 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<DataBuffer> content() {
return DataBufferUtils.readByteChannel(
() -> Files.newByteChannel(this.file, StandardOpenOption.READ),
DefaultDataBufferFactory.sharedInstance, 1024)
.subscribeOn(this.scheduler);
}
@Override
public Mono<Void> delete() {
return Mono.<Void>fromRunnable(() -> {
try {
Files.delete(this.file);
}
catch (IOException ex) {
throw new UncheckedIOException(ex);
}
})
.subscribeOn(this.scheduler);
}
}
}

139
spring-web/src/main/java/org/springframework/http/codec/multipart/DefaultParts.java
View File

@@ -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<DataBuffer> 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<Void> 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<DataBuffer> content();
Mono<Void> transferTo(Path dest);
Mono<Void> delete();
}
/**
* {@code Content} implementation based on a flux of data buffers.
*/
private static final class FluxContent implements Content {
private final Flux<DataBuffer> content;
public FluxContent(Flux<DataBuffer> content) {
this.content = content;
}
@Override
public Flux<DataBuffer> content() {
return this.content;
}
@Override
public Mono<Void> transferTo(Path dest) {
return DataBufferUtils.write(this.content, dest);
}
@Override
public Mono<Void> 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<DataBuffer> content() {
return DataBufferUtils.readByteChannel(
() -> Files.newByteChannel(this.file, StandardOpenOption.READ),
DefaultDataBufferFactory.sharedInstance, 1024)
.subscribeOn(this.scheduler);
}
@Override
public Mono<Void> transferTo(Path dest) {
return blockingOperation(() -> Files.copy(this.file, dest, StandardCopyOption.REPLACE_EXISTING));
}
@Override
public Mono<Void> delete() {
return blockingOperation(() -> {
Files.delete(this.file);
return null;
});
}
private Mono<Void> blockingOperation(Callable<?> callable) {
return Mono.<Void>create(sink -> {
try {
callable.call();
sink.success();
}
catch (Exception ex) {
sink.error(ex);
}
})
.subscribeOn(this.scheduler);
}
}
}

7
spring-web/src/main/java/org/springframework/http/codec/multipart/PartGenerator.java
View File

@@ -160,7 +160,7 @@ final class PartGenerator extends BaseSubscriber<MultipartParser.Token> {
requestToken();
}
});
emitPart(DefaultParts.part(headers, Content.fromFlux(streamingContent)));
emitPart(DefaultParts.part(headers, streamingContent));
}
}
@@ -518,7 +518,7 @@ final class PartGenerator extends BaseSubscriber<MultipartParser.Token> {
}
this.content.clear();
Flux<DataBuffer> 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<MultipartParser.Token> {
@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();
}