DATAMONGO-1855 - Initial reactive GridFS support.

We now support reactive GridFS using MongoDB's reactive GridFS API. Files can be consumed and provided as binary stream.

ReactiveGridFsOperations operations = …;

Publisher<DataBuffer> buffers = …
Mono<ObjectId> id = operations.store(buffers, "foo.xml");

Flux<DataBuffer> download = operations.getResource("foo.xml").flatMap(ReactiveGridFsResource::getDownloadStream);

Author: mp911de

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

@@ -0,0 +1,204 @@
+/*
+ * Copyright 2019 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
+ *
+ *      http://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 reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.bson.Document;
+import org.bson.types.ObjectId;
+import org.reactivestreams.Publisher;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.support.ResourcePatternResolver;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.lang.Nullable;
+
+import com.mongodb.client.gridfs.GridFSFindIterable;
+import com.mongodb.client.gridfs.model.GridFSFile;
+import com.mongodb.reactivestreams.client.gridfs.AsyncInputStream;
+
+/**
+ * Collection of operations to store and read files from MongoDB GridFS using reactive infrastructure.
+ *
+ * @author Mark Paluch
+ * @since 2.2
+ */
+public interface ReactiveGridFsOperations {
+
+	/**
+	 * Stores the given content into a file with the given name.
+	 *
+	 * @param content must not be {@literal null}.
+	 * @param filename must not be {@literal null} or empty.
+	 * @return the {@link ObjectId} of the {@link com.mongodb.client.gridfs.model.GridFSFile} just created.
+	 */
+	default Mono<ObjectId> store(Publisher<DataBuffer> content, String filename) {
+		return store(content, filename, (Object) null);
+	}
+
+	/**
+	 * Stores the given content into a file with the given name.
+	 *
+	 * @param content must not be {@literal null}.
+	 * @param metadata can be {@literal null}.
+	 * @return the {@link ObjectId} of the {@link com.mongodb.client.gridfs.model.GridFSFile} just created.
+	 */
+	default Mono<ObjectId> store(Publisher<DataBuffer> content, @Nullable Object metadata) {
+		return store(content, null, metadata);
+	}
+
+	/**
+	 * Stores the given content into a file with the given name.
+	 *
+	 * @param content must not be {@literal null}.
+	 * @param metadata can be {@literal null}.
+	 * @return the {@link ObjectId} of the {@link com.mongodb.client.gridfs.model.GridFSFile} just created.
+	 */
+	default Mono<ObjectId> store(Publisher<DataBuffer> content, @Nullable Document metadata) {
+		return store(content, null, metadata);
+	}
+
+	/**
+	 * Stores the given content into a file with the given name and content type.
+	 *
+	 * @param content must not be {@literal null}.
+	 * @param filename must not be {@literal null} or empty.
+	 * @param contentType can be {@literal null}.
+	 * @return the {@link ObjectId} of the {@link com.mongodb.client.gridfs.model.GridFSFile} just created.
+	 */
+	default Mono<ObjectId> store(Publisher<DataBuffer> content, @Nullable String filename, @Nullable String contentType) {
+		return store(content, filename, contentType, (Object) null);
+	}
+
+	/**
+	 * Stores the given content into a file with the given name using the given metadata. The metadata object will be
+	 * marshalled before writing.
+	 *
+	 * @param content must not be {@literal null}.
+	 * @param filename can be {@literal null} or empty.
+	 * @param metadata can be {@literal null}.
+	 * @return the {@link ObjectId} of the {@link com.mongodb.client.gridfs.model.GridFSFile} just created.
+	 */
+	default Mono<ObjectId> store(Publisher<DataBuffer> content, @Nullable String filename, @Nullable Object 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. The metadata
+	 * object will be marshalled before writing.
+	 *
+	 * @param content must not be {@literal null}.
+	 * @param filename must not be {@literal null} or empty.
+	 * @param contentType can be {@literal null}.
+	 * @param metadata can be {@literal null}
+	 * @return the {@link ObjectId} of the {@link com.mongodb.client.gridfs.model.GridFSFile} just created.
+	 */
+	Mono<ObjectId> store(AsyncInputStream content, @Nullable String filename, @Nullable String contentType,
+			@Nullable Object metadata);
+
+	/**
+	 * Stores the given content into a file with the given name and content type using the given metadata. The metadata
+	 * object will be marshalled before writing.
+	 *
+	 * @param content must not be {@literal null}.
+	 * @param filename must not be {@literal null} or empty.
+	 * @param contentType can be {@literal null}.
+	 * @param metadata can be {@literal null}
+	 * @return 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,
+			@Nullable Object metadata);
+
+	/**
+	 * Stores the given content into a file with the given name using the given metadata.
+	 *
+	 * @param content must not be {@literal null}.
+	 * @param filename must not be {@literal null} or empty.
+	 * @param metadata can be {@literal null}.
+	 * @return the {@link ObjectId} of the {@link com.mongodb.client.gridfs.model.GridFSFile} just created.
+	 */
+	default Mono<ObjectId> store(Publisher<DataBuffer> 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.
+	 *
+	 * @param content must not be {@literal null}.
+	 * @param filename must not be {@literal null} or empty.
+	 * @param contentType can be {@literal null}.
+	 * @param metadata can be {@literal null}.
+	 * @return the {@link ObjectId} of the {@link com.mongodb.client.gridfs.model.GridFSFile} just created.
+	 */
+	Mono<ObjectId> store(AsyncInputStream content, @Nullable String filename, @Nullable String contentType,
+			@Nullable Document metadata);
+
+	Mono<ObjectId> store(Publisher<DataBuffer> content, String filename, String contentType, Document metadata);
+
+	/**
+	 * Returns all files matching the given query. Note, that currently {@link Sort} criterias defined at the
+	 * {@link Query} will not be regarded as MongoDB does not support ordering for GridFS file access.
+	 *
+	 * @see <a href="https://jira.mongodb.org/browse/JAVA-431">MongoDB Jira: JAVA-431</a>
+	 * @param query must not be {@literal null}.
+	 * @return {@link GridFSFindIterable} to obtain results from. Eg. by calling
+	 *         {@link GridFSFindIterable#into(java.util.Collection)}.
+	 */
+	Flux<GridFSFile> find(Query query);
+
+	/**
+	 * Returns a single {@link com.mongodb.client.gridfs.model.GridFSFile} matching the given query or {@literal null} in
+	 * case no file matches.
+	 *
+	 * @param query must not be {@literal null}.
+	 * @return
+	 */
+	Mono<GridFSFile> findOne(Query query);
+
+	/**
+	 * Deletes all files matching the given {@link Query}.
+	 *
+	 * @param query must not be {@literal null}.
+	 */
+	Mono<Void> delete(Query query);
+
+	/**
+	 * Returns the {@link GridFsResource} with the given file name.
+	 *
+	 * @param filename must not be {@literal null}.
+	 * @return the resource. Use {@link org.springframework.core.io.Resource#exists()} to check if the returned
+	 *         {@link GridFsResource} is actually present.
+	 * @see ResourcePatternResolver#getResource(String)
+	 */
+	Mono<ReactiveGridFsResource> getResource(String filename);
+
+	/**
+	 * Returns the {@link GridFsResource} for a {@link GridFSFile}.
+	 *
+	 * @param file must not be {@literal null}.
+	 * @return the resource for the file.
+	 */
+	Mono<ReactiveGridFsResource> getResource(GridFSFile file);
+
+	/**
+	 * Returns all {@link GridFsResource}s matching the given file name pattern.
+	 *
+	 * @param filenamePattern must not be {@literal null}.
+	 * @return
+	 */
+	Flux<ReactiveGridFsResource> getResources(String filenamePattern);
+}

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

@@ -0,0 +1,180 @@
+/*
+ * Copyright 2019 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
+ *
+ *      http://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 reactor.core.publisher.Flux;
+
+import java.io.ByteArrayInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.reactivestreams.Publisher;
+import org.springframework.core.io.AbstractResource;
+import org.springframework.core.io.Resource;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+import com.mongodb.client.gridfs.model.GridFSFile;
+
+/**
+ * Reactive {@link GridFSFile} based {@link Resource} implementation.
+ *
+ * @author Mark Paluch
+ * @since 2.2
+ */
+public class ReactiveGridFsResource extends AbstractResource {
+
+	static final String CONTENT_TYPE_FIELD = "_contentType";
+	private static final ByteArrayInputStream EMPTY_INPUT_STREAM = new ByteArrayInputStream(new byte[0]);
+
+	private final @Nullable GridFSFile file;
+	private final String filename;
+	private final Flux<DataBuffer> content;
+
+	/**
+	 * Creates a new, absent {@link ReactiveGridFsResource}.
+	 *
+	 * @param filename filename of the absent resource.
+	 * @param content
+	 * @since 2.1
+	 */
+	private ReactiveGridFsResource(String filename, Publisher<DataBuffer> content) {
+
+		this.file = null;
+		this.filename = filename;
+		this.content = Flux.from(content);
+	}
+
+	/**
+	 * Creates a new {@link ReactiveGridFsResource} from the given {@link GridFSFile}.
+	 *
+	 * @param file must not be {@literal null}.
+	 * @param content
+	 */
+	public ReactiveGridFsResource(GridFSFile file, Publisher<DataBuffer> content) {
+
+		this.file = file;
+		this.filename = file.getFilename();
+		this.content = Flux.from(content);
+	}
+
+	/**
+	 * Obtain an absent {@link ReactiveGridFsResource}.
+	 *
+	 * @param filename filename of the absent resource, must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @since 2.1
+	 */
+	public static ReactiveGridFsResource absent(String filename) {
+
+		Assert.notNull(filename, "Filename must not be null");
+
+		return new ReactiveGridFsResource(filename, Flux.empty());
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.core.io.InputStreamResource#getInputStream()
+	 */
+	@Override
+	public InputStream getInputStream() throws IllegalStateException {
+		throw new UnsupportedOperationException();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.core.io.AbstractResource#contentLength()
+	 */
+	@Override
+	public long contentLength() throws IOException {
+
+		verifyExists();
+		return file.getLength();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.core.io.AbstractResource#getFilename()
+	 */
+	@Override
+	public String getFilename() throws IllegalStateException {
+		return filename;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.core.io.AbstractResource#exists()
+	 */
+	@Override
+	public boolean exists() {
+		return file != null;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.core.io.AbstractResource#lastModified()
+	 */
+	@Override
+	public long lastModified() throws IOException {
+
+		verifyExists();
+		return file.getUploadDate().getTime();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.core.io.AbstractResource#getDescription()
+	 */
+	@Override
+	public String getDescription() {
+		return String.format("GridFs resource [%s]", this.getFilename());
+	}
+
+	/**
+	 * Returns the {@link Resource}'s id.
+	 *
+	 * @return never {@literal null}.
+	 * @throws IllegalStateException if the file does not {@link #exists()}.
+	 */
+	public Object getId() {
+
+		Assert.state(exists(), () -> String.format("%s does not exist.", getDescription()));
+
+		return file.getId();
+	}
+
+	/**
+	 * Retrieve the download stream.
+	 *
+	 * @return
+	 */
+	public Flux<DataBuffer> getDownloadStream() {
+
+		if (!exists()) {
+			return Flux.error(new FileNotFoundException(String.format("%s does not exist.", getDescription())));
+		}
+		return content;
+	}
+
+	private void verifyExists() throws FileNotFoundException {
+
+		if (!exists()) {
+			throw new FileNotFoundException(String.format("%s does not exist.", getDescription()));
+		}
+	}
+}