DATAMONGO-625 - Add support for GridFS with predefined id.

We now support storing GridFS content with predefined id, which allows to replace an existing resource instead of having to delete and reupload it.

Original pull request: #842.

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/GridFsObject.java

@@ -0,0 +1,167 @@
+/*
+ * Copyright 2020 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.data.mongodb.gridfs;
+
+import org.bson.Document;
+import org.springframework.lang.Nullable;
+
+import com.mongodb.client.gridfs.model.GridFSFile;
+
+/**
+ * A common interface when dealing with GridFs items using Spring Data.o
+ *
+ * @author Christoph Strobl
+ * @since 3.0
+ */
+public interface GridFsObject<ID, CONTENT> {
+
+	/**
+	 * The {@link GridFSFile#getId()} value converted into its simple java type. <br />
+	 * A {@link org.bson.BsonString} will be converted to plain {@link String}.
+	 * 
+	 * @return can be {@literal null} depending on the implementation.
+	 */
+	@Nullable
+	ID getFileId();
+
+	/**
+	 * The filename.
+	 * 
+	 * @return
+	 */
+	String getFilename();
+
+	/**
+	 * The actual file content.
+	 * 
+	 * @return
+	 */
+	CONTENT getContent();
+
+	/**
+	 * Additional information like file metadata (eg. contentType).
+	 * 
+	 * @return never {@literal null}.
+	 */
+	Options getOptions();
+
+	/**
+	 * Additional, context relevant information.
+	 * 
+	 * @author Christoph Strobl
+	 */
+	class Options {
+
+		private Document metadata = new Document();
+		private int chunkSize = -1;
+
+		private Options(Document metadata, int chunkSize) {
+
+			this.metadata = metadata;
+			this.chunkSize = chunkSize;
+		}
+
+		/**
+		 * Static factory to create empty options.
+		 *
+		 * @return new instance of {@link Options}.
+		 */
+		public static Options none() {
+			return new Options(new Document(), -1);
+		}
+
+		/**
+		 * Static factory method to create {@link Options} with given chunk size.
+		 *
+		 * @param chunkSize
+		 * @return new instance of {@link Options}.
+		 */
+		public static Options chunked(int chunkSize) {
+			return new Options(new Document(), chunkSize);
+		}
+
+		/**
+		 * Static factory method to create {@link Options} with given content type.
+		 *
+		 * @param contentType
+		 * @return new instance of {@link Options}.
+		 */
+		public static Options typed(String contentType) {
+			return new Options(new Document("_contentType", contentType), -1);
+		}
+
+		/**
+		 * Static factory method to create {@link Options} by extracting information from the given {@link GridFSFile}.
+		 *
+		 * @param gridFSFile can be {@literal null}, returns {@link #none()} in that case.
+		 * @return new instance of {@link Options}.
+		 */
+		public static Options from(@Nullable GridFSFile gridFSFile) {
+			return gridFSFile != null ? new Options(gridFSFile.getMetadata(), gridFSFile.getChunkSize()) : none();
+		}
+
+		/**
+		 * Set the associated content type.
+		 * 
+		 * @param contentType must not be {@literal null}.
+		 * @return new instance of {@link Options}.
+		 */
+		public Options contentType(String contentType) {
+
+			Options target = new Options(new Document(metadata), chunkSize);
+			target.metadata.put("_contentType", contentType);
+			return target;
+		}
+
+		/**
+		 * @param metadata
+		 * @return new instance of {@link Options}.
+		 */
+		public Options metadata(Document metadata) {
+			return new Options(metadata, chunkSize);
+		}
+
+		/**
+		 * @param chunkSize the file chunk size to use.
+		 * @return new instance of {@link Options}.
+		 */
+		public Options chunkSize(int chunkSize) {
+			return new Options(metadata, chunkSize);
+		}
+
+		/**
+		 * @return never {@literal null}.
+		 */
+		public Document getMetadata() {
+			return metadata;
+		}
+
+		/**
+		 * @return the chunk size to use.
+		 */
+		public int getChunkSize() {
+			return chunkSize;
+		}
+
+		/**
+		 * @return {@literal null} if not set.
+		 */
+		@Nullable
+		String getContentType() {
+			return (String) metadata.get("_contentType");
+		}
+	}
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/GridFsOperations.java

@@ -22,7 +22,10 @@ import org.bson.types.ObjectId;
 import org.springframework.core.io.support.ResourcePatternResolver;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.data.mongodb.gridfs.GridFsUpload.GridFsUploadBuilder;
 import org.springframework.lang.Nullable;
+import org.springframework.util.ObjectUtils;
+import org.springframework.util.StringUtils;
 
 import com.mongodb.client.gridfs.GridFSFindIterable;
 
@@ -45,7 +48,9 @@ public interface GridFsOperations extends ResourcePatternResolver {
 	 * @param filename must not be {@literal null} or empty.
 	 * @return the {@link ObjectId} of the {@link com.mongodb.client.gridfs.model.GridFSFile} just created.
 	 */
-	ObjectId store(InputStream content, String filename);
+	default ObjectId store(InputStream content, String filename) {
+		return store(content, filename, null, null);
+	}
 
 	/**
 	 * Stores the given content into a file with the given name.
@@ -63,7 +68,9 @@ public interface GridFsOperations extends ResourcePatternResolver {
 	 * @param metadata can be {@literal null}.
 	 * @return the {@link ObjectId} of the {@link com.mongodb.client.gridfs.model.GridFSFile} just created.
 	 */
-	ObjectId store(InputStream content, @Nullable Document metadata);
+	default ObjectId store(InputStream content, @Nullable Document metadata) {
+		return store(content, null, metadata);
+	}
 
 	/**
 	 * Stores the given content into a file with the given name and content type.
@@ -107,7 +114,9 @@ public interface GridFsOperations extends ResourcePatternResolver {
 	 * @param metadata can be {@literal null}.
 	 * @return the {@link ObjectId} of the {@link com.mongodb.client.gridfs.model.GridFSFile} just created.
 	 */
-	ObjectId store(InputStream content, @Nullable String filename, @Nullable Document metadata);
+	default ObjectId store(InputStream content, @Nullable String filename, @Nullable Document metadata) {
+		return store(content, filename, null, metadata);
+	}
 
 	/**
 	 * Stores the given content into a file with the given name and content type using the given metadata.
@@ -118,8 +127,35 @@ public interface GridFsOperations extends ResourcePatternResolver {
 	 * @param metadata can be {@literal null}.
 	 * @return the {@link ObjectId} of the {@link com.mongodb.client.gridfs.model.GridFSFile} just created.
 	 */
-	ObjectId store(InputStream content, @Nullable String filename, @Nullable String contentType,
+	default ObjectId store(InputStream content, @Nullable String filename, @Nullable String contentType,
-			@Nullable Document metadata);
+			@Nullable Document metadata) {
+
+		GridFsUploadBuilder<ObjectId> uploadBuilder = GridFsUpload.fromStream(content);
+		if (StringUtils.hasText(filename)) {
+			uploadBuilder.filename(filename);
+		}
+		if (StringUtils.hasText(contentType)) {
+			uploadBuilder.contentType(contentType);
+		}
+		if (!ObjectUtils.isEmpty(metadata)) {
+			uploadBuilder.metadata(metadata);
+		}
+
+		return save(uploadBuilder.build());
+	}
+
+	/**
+	 * Stores the given {@link GridFsObject}, likely a {@link GridFsUpload}, into into a file with given
+	 * {@link GridFsObject#getFilename() name}. If the {@link GridFsObject#getFileId()} is set, the file will be stored
+	 * with that id, otherwise the server auto creates a new id. <br />
+	 * 
+	 * @param upload the {@link GridFsObject} (most likely a {@link GridFsUpload}) to be stored.
+	 * @param <T> id type of the underlying {@link com.mongodb.client.gridfs.model.GridFSFile}
+	 * @return the id of the stored file. Either an auto created value or {@link GridFsObject#getFileId()}, but never
+	 *         {@literal null}.
+	 * @since 3.0
+	 */
+	<T> T save(GridFsObject<T, InputStream> upload);
 
 	/**
 	 * Returns all files matching the given query. Note, that currently {@link Sort} criterias defined at the

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/GridFsResource.java

@@ -23,6 +23,7 @@ import java.util.Optional;
 
 import org.springframework.core.io.InputStreamResource;
 import org.springframework.core.io.Resource;
+import org.springframework.data.mongodb.util.BsonUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -37,7 +38,7 @@ import com.mongodb.client.gridfs.model.GridFSFile;
  * @author Hartmut Lang
  * @author Mark Paluch
  */
-public class GridFsResource extends InputStreamResource {
+public class GridFsResource extends InputStreamResource implements GridFsObject<Object, InputStream> {
 
 	static final String CONTENT_TYPE_FIELD = "_contentType";
 	private static final ByteArrayInputStream EMPTY_INPUT_STREAM = new ByteArrayInputStream(new byte[0]);
@@ -169,6 +170,17 @@ public class GridFsResource extends InputStreamResource {
 		return getGridFSFile().getId();
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.gridfs.GridFsObject#getFileId()
+	 */
+	@Override
+	public Object getFileId() {
+
+		Assert.state(exists(), () -> String.format("%s does not exist.", getDescription()));
+		return BsonUtils.toJavaType(getGridFSFile().getId());
+	}
+
 	/**
 	 * @return the underlying {@link GridFSFile}. Can be {@literal null} if absent.
 	 * @since 2.2
@@ -195,6 +207,29 @@ public class GridFsResource extends InputStreamResource {
 				.orElseThrow(() -> new MongoGridFSException("No contentType data for this GridFS file"));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.gridfs.GridFsObject#getContent()
+	 */
+	@Override
+	public InputStream getContent() {
+
+		try {
+			return getInputStream();
+		} catch (IOException e) {
+			throw new IllegalStateException("Failed to obtain input stream for " + filename, e);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.gridfs.GridFsObject#getOptions()
+	 */
+	@Override
+	public Options getOptions() {
+		return Options.from(getGridFSFile());
+	}
+
 	private void verifyExists() throws FileNotFoundException {
 
 		if (!exists()) {

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/GridFsTemplate.java

@@ -29,6 +29,7 @@ import org.springframework.core.io.support.ResourcePatternResolver;
 import org.springframework.data.mongodb.MongoDatabaseFactory;
 import org.springframework.data.mongodb.core.convert.MongoConverter;
 import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.data.mongodb.util.BsonUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
@@ -38,6 +39,7 @@ import com.mongodb.client.gridfs.GridFSBucket;
 import com.mongodb.client.gridfs.GridFSBuckets;
 import com.mongodb.client.gridfs.GridFSFindIterable;
 import com.mongodb.client.gridfs.model.GridFSFile;
+import com.mongodb.client.gridfs.model.GridFSUploadOptions;
 
 /**
  * {@link GridFsOperations} implementation to store content into MongoDB GridFS.
@@ -85,14 +87,6 @@ public class GridFsTemplate extends GridFsOperationsSupport implements GridFsOpe
 		this.bucket = bucket;
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.mongodb.gridfs.GridFsOperations#store(java.io.InputStream, java.lang.String)
-	 */
-	public ObjectId store(InputStream content, String filename) {
-		return store(content, filename, (Object) null);
-	}
-
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.mongodb.gridfs.GridFsOperations#store(java.io.InputStream, java.lang.Object)
@@ -102,15 +96,6 @@ public class GridFsTemplate extends GridFsOperationsSupport implements GridFsOpe
 		return store(content, null, metadata);
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.mongodb.gridfs.GridFsOperations#store(java.io.InputStream, com.mongodb.Document)
-	 */
-	@Override
-	public ObjectId store(InputStream content, @Nullable Document metadata) {
-		return store(content, null, metadata);
-	}
-
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.mongodb.gridfs.GridFsOperations#store(java.io.InputStream, java.lang.String, java.lang.String)
@@ -138,21 +123,24 @@ public class GridFsTemplate extends GridFsOperationsSupport implements GridFsOpe
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.mongodb.gridfs.GridFsOperations#store(java.io.InputStream, java.lang.String, com.mongodb.Document)
+	 * @see org.springframework.data.mongodb.gridfs.GridFsOperations#save(org.springframework.data.mongodb.gridfs.GridFsObject)
 	 */
-	public ObjectId store(InputStream content, @Nullable String filename, @Nullable Document metadata) {
+	public <T> T save(GridFsObject<T, InputStream> upload) {
-		return this.store(content, filename, null, metadata);
-	}
 
-	/*
+		GridFSUploadOptions uploadOptions = computeUploadOptionsFor(upload.getOptions().getContentType(),
-	 * (non-Javadoc)
+				upload.getOptions().getMetadata());
-	 * @see org.springframework.data.mongodb.gridfs.GridFsOperations#store(java.io.InputStream, java.lang.String, com.mongodb.Document)
+
-	 */
+		if (upload.getOptions().getChunkSize() > 0) {
-	public ObjectId store(InputStream content, @Nullable String filename, @Nullable String contentType,
+			uploadOptions.chunkSizeBytes(upload.getOptions().getChunkSize());
-			@Nullable Document metadata) {
+		}
+
+		if (upload.getFileId() == null) {
+			return (T) getGridFs().uploadFromStream(upload.getFilename(), upload.getContent(), uploadOptions);
+		}
 
-		Assert.notNull(content, "InputStream must not be null!");
+		getGridFs().uploadFromStream(BsonUtils.simpleToBsonValue(upload.getFileId()), upload.getFilename(),
-		return getGridFs().uploadFromStream(filename, content, computeUploadOptionsFor(contentType, metadata));
+				upload.getContent(), uploadOptions);
+		return upload.getFileId();
 	}
 
 	/*

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/GridFsUpload.java

@@ -0,0 +1,214 @@
+/*
+ * Copyright 2020 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.data.mongodb.gridfs;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.bson.Document;
+import org.bson.types.ObjectId;
+import org.springframework.data.util.Lazy;
+import org.springframework.lang.Nullable;
+
+import com.mongodb.client.gridfs.model.GridFSFile;
+
+/**
+ * @author Christoph Strobl
+ * @since 3.0
+ */
+public class GridFsUpload<ID> implements GridFsObject<ID, InputStream> {
+
+	private static final InputStream EMPTY_STREAM = new InputStream() {
+		@Override
+		public int read() throws IOException {
+			return -1;
+		}
+	};
+
+	private ID id;
+	private Lazy<InputStream> dataStream;
+	private String filename;
+	private Options options;
+
+	/**
+	 * The {@link GridFSFile#getId()} value converted into its simple java type. <br />
+	 * A {@link org.bson.BsonString} will be converted to plain {@link String}.
+	 * 
+	 * @return can be {@literal null}.
+	 * @see org.springframework.data.mongodb.gridfs.GridFsObject#getFileId()
+	 */
+	@Override
+	public ID getFileId() {
+		return id;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.gridfs.GridFsObject#getFielname()
+	 */
+	@Override
+	public String getFilename() {
+		return filename;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.gridfs.GridFsObject#getContent()
+	 */
+	@Override
+	public InputStream getContent() {
+		return dataStream.orElse(EMPTY_STREAM);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.gridfs.GridFsObject#getOptions()
+	 */
+	@Override
+	public Options getOptions() {
+		return options;
+	}
+
+	/**
+	 * Create a new instance of {@link GridFsUpload} for the given {@link InputStream}.
+	 *
+	 * @param stream must not be {@literal null}.
+	 * @return new instance of {@link GridFsUpload}.
+	 */
+	public static GridFsUploadBuilder<ObjectId> fromStream(InputStream stream) {
+		return new GridFsUploadBuilder().content(stream);
+	}
+
+	/**
+	 * Builder to create {@link GridFsUpload} in a fluent way.
+	 * 
+	 * @param <T> the target id type.
+	 */
+	public static class GridFsUploadBuilder<T> {
+
+		private GridFsUpload upload;
+
+		public GridFsUploadBuilder() {
+			this.upload = new GridFsUpload();
+			this.upload.options = Options.none();
+		}
+
+		/**
+		 * Define the content of the file to upload.
+		 *
+		 * @param stream the upload content.
+		 * @return this.
+		 */
+		public GridFsUploadBuilder<T> content(InputStream stream) {
+
+			upload.dataStream = Lazy.of(() -> stream);
+			return this;
+		}
+
+		/**
+		 * Set the id to use.
+		 *
+		 * @param id the id to save the content to.
+		 * @param <T1>
+		 * @return this.
+		 */
+		public <T1> GridFsUploadBuilder<T1> id(T1 id) {
+
+			upload.id = id;
+			return (GridFsUploadBuilder<T1>) this;
+		}
+
+		/**
+		 * Set the filename.
+		 *
+		 * @param filename the filename to use.
+		 * @return this.
+		 */
+		public GridFsUploadBuilder<T> filename(String filename) {
+
+			upload.filename = filename;
+			return this;
+		}
+
+		/**
+		 * Set additional file information.
+		 *
+		 * @param options must not be {@literal null}.
+		 * @return this.
+		 */
+		public GridFsUploadBuilder<T> options(Options options) {
+
+			upload.options = options;
+			return this;
+		}
+
+		/**
+		 * Set the file metadata.
+		 *
+		 * @param metadata must not be {@literal null}.
+		 * @return
+		 */
+		public GridFsUploadBuilder<T> metadata(Document metadata) {
+
+			upload.options = upload.options.metadata(metadata);
+			return this;
+		}
+
+		/**
+		 * Set the upload chunk size in bytes.
+		 *
+		 * @param chunkSize use negative number for default.
+		 * @return this.
+		 */
+		public GridFsUploadBuilder<T> chunkSize(int chunkSize) {
+
+			upload.options = upload.options.chunkSize(chunkSize);
+			return this;
+		}
+
+		/**
+		 * Set id, filename, metadata and chunk size from given file.
+		 * 
+		 * @param gridFSFile must not be {@literal null}.
+		 * @return this.
+		 */
+		public GridFsUploadBuilder<T> gridFsFile(GridFSFile gridFSFile) {
+
+			upload.id = gridFSFile.getId();
+			upload.filename = gridFSFile.getFilename();
+			upload.options = upload.options.metadata(gridFSFile.getMetadata());
+			upload.options = upload.options.chunkSize(gridFSFile.getChunkSize());
+
+			return this;
+		}
+
+		/**
+		 * Set the content type.
+		 * 
+		 * @param contentType must not be {@literal null}.
+		 * @return this.
+		 */
+		public GridFsUploadBuilder<T> contentType(String contentType) {
+
+			upload.options = upload.options.contentType(contentType);
+			return this;
+		}
+
+		public GridFsUpload<T> build() {
+			return (GridFsUpload<T>) upload;
+		}
+	}
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsOperations.java

@@ -24,7 +24,10 @@ import org.reactivestreams.Publisher;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.data.mongodb.gridfs.ReactiveGridFsUpload.ReactiveGridFsUploadBuilder;
 import org.springframework.lang.Nullable;
+import org.springframework.util.ObjectUtils;
+import org.springframework.util.StringUtils;
 
 import com.mongodb.client.gridfs.model.GridFSFile;
 
@@ -137,8 +140,36 @@ public interface ReactiveGridFsOperations {
 	 * @return a {@link Mono} emitting the {@link ObjectId} of the {@link com.mongodb.client.gridfs.model.GridFSFile} just
 	 *         created.
 	 */
-	Mono<ObjectId> store(Publisher<DataBuffer> content, @Nullable String filename, @Nullable String contentType,
+	default Mono<ObjectId> store(Publisher<DataBuffer> content, @Nullable String filename, @Nullable String contentType,
-			@Nullable Document metadata);
+			@Nullable Document metadata) {
+
+		ReactiveGridFsUploadBuilder<ObjectId> uploadBuilder = ReactiveGridFsUpload.fromPublisher(content);
+
+		if (StringUtils.hasText(filename)) {
+			uploadBuilder.filename(filename);
+		}
+		if (StringUtils.hasText(contentType)) {
+			uploadBuilder.contentType(contentType);
+		}
+		if (!ObjectUtils.isEmpty(metadata)) {
+			uploadBuilder.metadata(metadata);
+		}
+
+		return save(uploadBuilder.build());
+	}
+
+	/**
+	 * Stores the given {@link GridFsObject}, likely a {@link GridFsUpload}, into into a file with given
+	 * {@link GridFsObject#getFilename() name}. If the {@link GridFsObject#getFileId()} is set, the file will be stored
+	 * with that id, otherwise the server auto creates a new id. <br />
+	 *
+	 * @param upload the {@link GridFsObject} (most likely a {@link GridFsUpload}) to be stored.
+	 * @param <T> id type of the underlying {@link com.mongodb.client.gridfs.model.GridFSFile}
+	 * @return {@link Mono} emitting the id of the stored file which is either an auto created value or
+	 *         {@link GridFsObject#getFileId()}.
+	 * @since 3.0
+	 */
+	<T> Mono<T> save(GridFsObject<T, Publisher<DataBuffer>> upload);
 
 	/**
 	 * Returns a {@link Flux} emitting all files matching the given query. <br />

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsResource.java

@@ -21,12 +21,14 @@ import reactor.core.publisher.Mono;
 import java.io.InputStream;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.bson.BsonValue;
 import org.reactivestreams.Publisher;
 import org.springframework.core.io.Resource;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferFactory;
 import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.core.io.buffer.DefaultDataBufferFactory;
+import org.springframework.data.mongodb.util.BsonUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -41,10 +43,12 @@ import com.mongodb.reactivestreams.client.gridfs.GridFSDownloadPublisher;
  * @author Christoph Strobl
  * @since 2.2
  */
-public class ReactiveGridFsResource {
+public class ReactiveGridFsResource implements GridFsObject<Object, Publisher<DataBuffer>> {
 
 	private final AtomicBoolean consumed = new AtomicBoolean(false);
 
+	private final @Nullable Object id;
+	private final Options options;
 	private final String filename;
 	private final @Nullable GridFSDownloadPublisher downloadPublisher;
 	private final DataBufferFactory dataBufferFactory;
@@ -56,21 +60,43 @@ public class ReactiveGridFsResource {
 	 * @param downloadPublisher
 	 */
 	public ReactiveGridFsResource(String filename, @Nullable GridFSDownloadPublisher downloadPublisher) {
-		this(filename, downloadPublisher, new DefaultDataBufferFactory());
+		this(null, filename, Options.none(), downloadPublisher);
 	}
 
 	/**
 	 * Creates a new, absent {@link ReactiveGridFsResource}.
 	 *
+	 * @param id
 	 * @param filename filename of the absent resource.
+	 * @param options
+	 * @param downloadPublisher
+	 * @since 3.0
+	 */
+	public ReactiveGridFsResource(@Nullable Object id, String filename, Options options,
+			@Nullable GridFSDownloadPublisher downloadPublisher) {
+		this(id, filename, options, downloadPublisher, new DefaultDataBufferFactory());
+	}
+
+	ReactiveGridFsResource(GridFSFile file, @Nullable GridFSDownloadPublisher downloadPublisher, DataBufferFactory dataBufferFactory) {
+		this(file.getId(), file.getFilename(), Options.from(file), downloadPublisher, dataBufferFactory);
+	}
+
+	/**
+	 * Creates a new, absent {@link ReactiveGridFsResource}.
+	 *
+	 * @param id
+	 * @param filename filename of the absent resource.
+	 * @param options
 	 * @param downloadPublisher
 	 * @param dataBufferFactory
 	 * @since 3.0
 	 */
-	ReactiveGridFsResource(String filename, @Nullable GridFSDownloadPublisher downloadPublisher,
+	ReactiveGridFsResource(@Nullable Object id, String filename, Options options,
-			DataBufferFactory dataBufferFactory) {
+			@Nullable GridFSDownloadPublisher downloadPublisher, DataBufferFactory dataBufferFactory) {
 
+		this.id = id;
 		this.filename = filename;
+		this.options = options;
 		this.downloadPublisher = downloadPublisher;
 		this.dataBufferFactory = dataBufferFactory;
 	}
@@ -88,6 +114,15 @@ public class ReactiveGridFsResource {
 		return new ReactiveGridFsResource(filename, null);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.gridfs.GridFsObject#getFileId()
+	 */
+	@Override
+	public Object getFileId() {
+		return id instanceof BsonValue ? BsonUtils.toJavaType((BsonValue) id) : id;
+	}
+
 	/**
 	 * @see org.springframework.core.io.AbstractResource#getFilename()
 	 */
@@ -140,6 +175,24 @@ public class ReactiveGridFsResource {
 		return createDownloadStream(downloadPublisher);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.gridfs.GridFsObject#getContent()
+	 */
+	@Override
+	public Flux<DataBuffer> getContent() {
+		return getDownloadStream();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.gridfs.GridFsObject#getOptions()
+	 */
+	@Override
+	public Options getOptions() {
+		return options;
+	}
+
 	/**
 	 * Obtain the download stream emitting chunks of data with given {@code chunkSize} as they come in.
 	 *

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsTemplate.java

@@ -21,8 +21,6 @@ import static org.springframework.data.mongodb.gridfs.GridFsCriteria.*;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-import java.nio.ByteBuffer;
-
 import org.bson.Document;
 import org.bson.types.ObjectId;
 import org.reactivestreams.Publisher;
@@ -34,6 +32,7 @@ import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory;
 import org.springframework.data.mongodb.core.convert.MongoConverter;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.SerializationUtils;
+import org.springframework.data.mongodb.util.BsonUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
@@ -120,20 +119,29 @@ public class ReactiveGridFsTemplate extends GridFsOperationsSupport implements R
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.mongodb.gridfs.ReactiveGridFsOperations#store(org.reactivestreams.Publisher, java.lang.String, java.lang.String, org.bson.Document)
+	 * @see org.springframework.data.mongodb.gridfs.ReactiveGridFsOperations#save(org.springframework.data.mongodb.gridfs.GridFsObject)
 	 */
-	@Override
+	public <T> Mono<T> save(GridFsObject<T, Publisher<DataBuffer>> upload) {
-	public Mono<ObjectId> store(Publisher<DataBuffer> content, @Nullable String filename, @Nullable String contentType,
+
-			@Nullable Document metadata) {
+		GridFSUploadOptions uploadOptions = computeUploadOptionsFor(upload.getOptions().getContentType(),
+				upload.getOptions().getMetadata());
 
-		Assert.notNull(content, "Content must not be null!");
+		if (upload.getOptions().getChunkSize() > 0) {
+			uploadOptions.chunkSizeBytes(upload.getOptions().getChunkSize());
+		}
+
+		if (upload.getFileId() == null) {
+			GridFSUploadPublisher<ObjectId> publisher = getGridFs().uploadFromPublisher(upload.getFilename(),
+					Flux.from(upload.getContent()).map(DataBuffer::asByteBuffer), uploadOptions);
+
+			return (Mono<T>) Mono.from(publisher);
+		}
 
-		GridFSUploadOptions uploadOptions = new GridFSUploadOptions();
+		GridFSUploadPublisher<Void> publisher = getGridFs().uploadFromPublisher(
-		uploadOptions.metadata(metadata);
+				BsonUtils.simpleToBsonValue(upload.getFileId()), upload.getFilename(),
+				Flux.from(upload.getContent()).map(DataBuffer::asByteBuffer), uploadOptions);
 
-		GridFSUploadPublisher<ObjectId> publisher = getGridFs().uploadFromPublisher(filename,
+		return Mono.from(publisher).then(Mono.just(upload.getFileId()));
-				Flux.from(content).map(DataBuffer::asByteBuffer), uploadOptions);
-		return Mono.from(publisher);
 	}
 
 	/*
@@ -209,7 +217,7 @@ public class ReactiveGridFsTemplate extends GridFsOperationsSupport implements R
 		Assert.notNull(file, "GridFSFile must not be null!");
 
 		return Mono.fromSupplier(() -> {
-			return new ReactiveGridFsResource(file.getFilename(), getGridFs().downloadToPublisher(file.getId()), dataBufferFactory);
+			return new ReactiveGridFsResource(file, getGridFs().downloadToPublisher(file.getId()), dataBufferFactory);
 		});
 	}
 

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsUpload.java

@@ -0,0 +1,205 @@
+/*
+ * Copyright 2020 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.data.mongodb.gridfs;
+
+import org.bson.Document;
+import org.bson.types.ObjectId;
+import org.reactivestreams.Publisher;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.lang.Nullable;
+
+import com.mongodb.client.gridfs.model.GridFSFile;
+
+/**
+ * @author Christoph Strobl
+ * @since 3.0
+ */
+public class ReactiveGridFsUpload<ID> implements GridFsObject<ID, Publisher<DataBuffer>> {
+
+	private ID id;
+	private Publisher<DataBuffer> dataStream;
+	private String filename;
+	private Options options;
+
+	/**
+	 * The {@link GridFSFile#getId()} value converted into its simple java type. <br />
+	 * A {@link org.bson.BsonString} will be converted to plain {@link String}.
+	 *
+	 * @return can be {@literal null}.
+	 * @see org.springframework.data.mongodb.gridfs.GridFsObject#getFileId()
+	 */
+	@Override
+	public ID getFileId() {
+		return id;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.gridfs.GridFsObject#getFielname()
+	 */
+	@Override
+	public String getFilename() {
+		return filename;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.gridfs.GridFsObject#getContent()
+	 */
+	@Override
+	public Publisher<DataBuffer> getContent() {
+		return dataStream;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.gridfs.GridFsObject#getOptions()
+	 */
+	@Override
+	public Options getOptions() {
+		return options;
+	}
+
+	/**
+	 * Create a new instance of {@link ReactiveGridFsUpload} for the given {@link Publisher}.
+	 *
+	 * @param source must not be {@literal null}.
+	 * @return new instance of {@link GridFsUpload}.
+	 */
+	public static ReactiveGridFsUploadBuilder<ObjectId> fromPublisher(Publisher<DataBuffer> source) {
+		return new ReactiveGridFsUploadBuilder().content(source);
+	}
+
+	/**
+	 * Builder to create {@link ReactiveGridFsUpload} in a fluent way.
+	 *
+	 * @param <T> the target id type.
+	 */
+	public static class ReactiveGridFsUploadBuilder<T> {
+
+		ReactiveGridFsUpload upload;
+
+		public ReactiveGridFsUploadBuilder() {
+
+			this.upload = new ReactiveGridFsUpload();
+			this.upload.options = Options.none();
+		}
+
+		/**
+		 * Define the content of the file to upload.
+		 *
+		 * @param source the upload content.
+		 * @return this.
+		 */
+		public ReactiveGridFsUploadBuilder<T> content(Publisher<DataBuffer> source) {
+			upload.dataStream = source;
+			return this;
+		}
+
+		/**
+		 * Set the id to use.
+		 *
+		 * @param id the id to save the content to.
+		 * @param <T1>
+		 * @return this.
+		 */
+		public <T1> ReactiveGridFsUploadBuilder<T1> id(T1 id) {
+
+			upload.id = id;
+			return (ReactiveGridFsUploadBuilder<T1>) this;
+		}
+
+		/**
+		 * Set the filename.
+		 *
+		 * @param filename the filename to use.
+		 * @return this.
+		 */
+		public ReactiveGridFsUploadBuilder<T> filename(String filename) {
+
+			upload.filename = filename;
+			return this;
+		}
+
+		/**
+		 * Set additional file information.
+		 *
+		 * @param options must not be {@literal null}.
+		 * @return this.
+		 */
+		public ReactiveGridFsUploadBuilder<T> options(Options options) {
+
+			upload.options = options;
+			return this;
+		}
+
+		/**
+		 * Set the file metadata.
+		 *
+		 * @param metadata must not be {@literal null}.
+		 * @return
+		 */
+		public ReactiveGridFsUploadBuilder<T> metadata(Document metadata) {
+
+			upload.options = upload.options.metadata(metadata);
+			return this;
+		}
+
+		/**
+		 * Set the upload chunk size in bytes.
+		 *
+		 * @param chunkSize use negative number for default.
+		 * @return
+		 */
+		public ReactiveGridFsUploadBuilder<T> chunkSize(int chunkSize) {
+
+			upload.options = upload.options.chunkSize(chunkSize);
+			return this;
+		}
+
+		/**
+		 * Set id, filename, metadata and chunk size from given file.
+		 *
+		 * @param gridFSFile must not be {@literal null}.
+		 * @return this.
+		 */
+		public ReactiveGridFsUploadBuilder<T> gridFsFile(GridFSFile gridFSFile) {
+
+			upload.id = gridFSFile.getId();
+			upload.filename = gridFSFile.getFilename();
+			upload.options = upload.options.metadata(gridFSFile.getMetadata());
+			upload.options = upload.options.chunkSize(gridFSFile.getChunkSize());
+
+			return this;
+		}
+
+		/**
+		 * Set the content type.
+		 *
+		 * @param contentType must not be {@literal null}.
+		 * @return this.
+		 */
+		public ReactiveGridFsUploadBuilder<T> contentType(String contentType) {
+
+			upload.options = upload.options.contentType(contentType);
+			return this;
+		}
+
+		public ReactiveGridFsUpload<T> build() {
+			return (ReactiveGridFsUpload<T>) upload;
+		}
+	}
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/util/BsonUtils.java

@@ -23,15 +23,24 @@ import java.util.StringJoiner;
 import java.util.function.Function;
 import java.util.stream.StreamSupport;
 
+import org.bson.BsonBinary;
+import org.bson.BsonBoolean;
+import org.bson.BsonDouble;
+import org.bson.BsonInt32;
+import org.bson.BsonInt64;
+import org.bson.BsonObjectId;
+import org.bson.BsonString;
 import org.bson.BsonValue;
 import org.bson.Document;
 import org.bson.codecs.DocumentCodec;
 import org.bson.conversions.Bson;
 import org.bson.json.JsonParseException;
+import org.bson.types.ObjectId;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.mongodb.CodecRegistryProvider;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
+import org.springframework.util.NumberUtils;
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.StringUtils;
 
@@ -120,6 +129,60 @@ public class BsonUtils {
 		}
 	}
 
+	/**
+	 * Convert a given simple value (eg. {@link String}, {@link Long}) to its corresponding {@link BsonValue}.
+	 *
+	 * @param source must not be {@literal null}.
+	 * @return the corresponding {@link BsonValue} representation.
+	 * @throws IllegalArgumentException if {@literal source} does not correspond to a {@link BsonValue} type.
+	 * @since 3.0
+	 */
+	public static BsonValue simpleToBsonValue(Object source) {
+
+		if (source instanceof BsonValue) {
+			return (BsonValue) source;
+		}
+
+		if (source instanceof ObjectId) {
+			return new BsonObjectId((ObjectId) source);
+		}
+
+		if (source instanceof String) {
+			return new BsonString((String) source);
+		}
+
+		if (source instanceof Double) {
+			return new BsonDouble((Double) source);
+		}
+
+		if (source instanceof Integer) {
+			return new BsonInt32((Integer) source);
+		}
+
+		if (source instanceof Long) {
+			return new BsonInt64((Long) source);
+		}
+
+		if (source instanceof byte[]) {
+			return new BsonBinary((byte[]) source);
+		}
+
+		if (source instanceof Boolean) {
+			return new BsonBoolean((Boolean) source);
+		}
+
+		if(source instanceof Float) {
+			return new BsonDouble((Float) source);
+		}
+
+		if (source instanceof Double) {
+			return new BsonDouble((Double) source);
+		}
+
+		throw new IllegalArgumentException(
+				String.format("Unable to convert % (%s) to BsonValue.", source, source != null ? source.getClass() : "null"));
+	}
+
 	/**
 	 * Merge the given {@link Document documents} into on in the given order. Keys contained within multiple documents are
 	 * overwritten by their follow ups.

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/gridfs/GridFsTemplateIntegrationTests.java

@@ -30,6 +30,7 @@ import java.util.Map.Entry;
 import java.util.stream.Stream;
 
 import org.bson.BsonObjectId;
+import org.bson.BsonString;
 import org.bson.Document;
 import org.bson.types.ObjectId;
 import org.junit.Before;
@@ -289,6 +290,29 @@ public class GridFsTemplateIntegrationTests {
 		}
 	}
 
+	@Test // DATAMONGO-625
+	public void storeSavesGridFsUploadWithGivenIdCorrectly() throws IOException {
+
+		String id = "id-1";
+
+		GridFsUpload<String> upload = GridFsUpload.fromStream(resource.getInputStream()) //
+				.id(id) //
+				.filename("gridFsUpload.xml") //
+				.contentType("xml") //
+				.build();
+
+		assertThat(operations.save(upload)).isEqualTo(id);
+
+		GridFsResource fsFile = operations.getResource(operations.findOne(query(where("_id").is(id))));
+		byte[] content = StreamUtils.copyToByteArray(fsFile.getInputStream());
+
+		assertThat(content).isEqualTo(StreamUtils.copyToByteArray(resource.getInputStream()));
+		assertThat(fsFile.getFilename()).isEqualTo("gridFsUpload.xml");
+		assertThat(fsFile.getId()).isEqualTo(new BsonString(id));
+		assertThat(fsFile.getFileId()).isEqualTo(id);
+		assertThat(fsFile.getContentType()).isEqualTo("xml");
+	}
+
 	@Test // DATAMONGO-765
 	public void considersSkipLimitWhenQueryingFiles() {
 

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsTemplateTests.java

@@ -29,12 +29,12 @@ import java.io.InputStreamReader;
 import java.nio.ByteBuffer;
 
 import org.bson.BsonObjectId;
+import org.bson.BsonString;
 import org.bson.Document;
 import org.bson.types.ObjectId;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.core.io.ClassPathResource;
 import org.springframework.core.io.Resource;
@@ -265,6 +265,41 @@ public class ReactiveGridFsTemplateTests {
 				.verifyComplete();
 	}
 
+	@Test // DATAMONGO-625
+	public void storeSavesGridFsUploadWithGivenIdCorrectly() throws IOException {
+
+		String id = "id-1";
+		byte[] content = StreamUtils.copyToByteArray(resource.getInputStream());
+		Flux<DataBuffer> data = DataBufferUtils.read(resource, new DefaultDataBufferFactory(), 256);
+
+		ReactiveGridFsUpload<String> upload = ReactiveGridFsUpload.fromPublisher(data) //
+				.id(id) //
+				.filename("gridFsUpload.xml") //
+				.contentType("xml") //
+				.build();
+
+		operations.save(upload).as(StepVerifier::create).expectNext(id).verifyComplete();
+
+		operations.findOne(query(where("_id").is(id))).flatMap(operations::getResource)
+				.flatMapMany(ReactiveGridFsResource::getDownloadStream) //
+				.transform(DataBufferUtils::join) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(dataBuffer -> {
+
+					byte[] actual = new byte[dataBuffer.readableByteCount()];
+					dataBuffer.read(actual);
+
+					assertThat(actual).isEqualTo(content);
+				}) //
+				.verifyComplete();
+
+		operations.findOne(query(where("_id").is(id))).as(StepVerifier::create).consumeNextWith(it -> {
+			assertThat(it.getFilename()).isEqualTo("gridFsUpload.xml");
+			assertThat(it.getId()).isEqualTo(new BsonString(id));
+			assertThat(it.getMetadata()).containsValue("xml");
+		}).verifyComplete();
+	}
+
 	@Test // DATAMONGO-765
 	public void considersSkipLimitWhenQueryingFiles() {
 

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/util/json/BsonUtilsTest.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2020 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.data.mongodb.util.json;
+
+import static org.assertj.core.api.Assertions.*;
+
+import org.bson.BsonDouble;
+import org.bson.BsonInt32;
+import org.bson.BsonInt64;
+import org.bson.BsonObjectId;
+import org.bson.BsonString;
+import org.bson.types.ObjectId;
+import org.junit.jupiter.api.Test;
+import org.springframework.data.mongodb.util.BsonUtils;
+
+/**
+ * @author Christoph Strobl
+ */
+class BsonUtilsTest {
+
+	@Test // DATAMONGO-625
+	void simpleToBsonValue() {
+
+		assertThat(BsonUtils.simpleToBsonValue(Long.valueOf(10))).isEqualTo(new BsonInt64(10));
+		assertThat(BsonUtils.simpleToBsonValue(new Integer(10))).isEqualTo(new BsonInt32(10));
+		assertThat(BsonUtils.simpleToBsonValue(Double.valueOf(0.1D))).isEqualTo(new BsonDouble(0.1D));
+		assertThat(BsonUtils.simpleToBsonValue("value")).isEqualTo(new BsonString("value"));
+	}
+
+	@Test // DATAMONGO-625
+	void primitiveToBsonValue() {
+		assertThat(BsonUtils.simpleToBsonValue(10L)).isEqualTo(new BsonInt64(10));
+	}
+
+	@Test // DATAMONGO-625
+	void objectIdToBsonValue() {
+
+		ObjectId source = new ObjectId();
+		assertThat(BsonUtils.simpleToBsonValue(source)).isEqualTo(new BsonObjectId(source));
+	}
+
+	@Test // DATAMONGO-625
+	void bsonValueToBsonValue() {
+
+		BsonObjectId source = new BsonObjectId(new ObjectId());
+		assertThat(BsonUtils.simpleToBsonValue(source)).isSameAs(source);
+	}
+
+	@Test // DATAMONGO-625
+	void unsupportedToBsonValue() {
+		assertThatExceptionOfType(IllegalArgumentException.class)
+				.isThrownBy(() -> BsonUtils.simpleToBsonValue(new Object()));
+	}
+}