import com.mongodb.client.gridfs.model.GridFSUploadOptions; //导入依赖的package包/类
@Override
public Uri putBlob(Uri uri, Blob blob) {
	GridFSUploadOptions options = new GridFSUploadOptions();

	Document document = JsonBsonCodec.toBson(mapper, blob.getMetadata());
	options.metadata(document);

	GridFSUploadStream file = bucket.openUploadStream(uri.toString(), options);
	try {
		IOUtils.copy(blob.getPayload().openStream(), file);
	} catch (IOException e) {
		throw Throwables.propagate(e);
	}
	file.close();

	return Uri.create("mongodb://" + databaseName + "/" + bucket.getBucketName() + "/" + file.getFileId().toString());
}
 

import com.mongodb.client.gridfs.model.GridFSUploadOptions; //导入依赖的package包/类
@Override
public InputStream store(DownloadItem downloadItem) throws IOException, JSONException {
	rwl.r.lock();
	try {
		final URI uri = downloadItem.getUri();
		if (!uri.equals(this.uri))
			throw new IOException("The URI does not match: " + uri + " / " + this.uri);

		final Document newDocument = Document.parse(downloadItem.getMetaAsJson());
		newDocument.put("uri", uriString);

		final BsonValue id = metaCollection.replaceOne(eq("uri", uriString), newDocument, UPSERT)
				.getUpsertedId();

		final GridFSUploadOptions options = new GridFSUploadOptions().metadata(new Document("_id", id));

		contentGrid.uploadFromStream(id, uriString, downloadItem.getContentInputStream(), options);

		return contentGrid.openDownloadStream(id);
	} finally {
		rwl.r.unlock();
	}
}
 

import com.mongodb.client.gridfs.model.GridFSUploadOptions; //导入依赖的package包/类
private void uploadStream(SmofGridRef ref, String name, InputStream stream) {
	final String bucketName = ref.getBucketName();
	final ObjectId id;
	final GridFSBucket bucket;
	Preconditions.checkNotNull(bucketName, "No bucket specified");
	final GridFSUploadOptions options = new GridFSUploadOptions().metadata(ref.getMetadata());
	bucket = pool.getBucket(bucketName);
	id = bucket.uploadFromStream(name, stream, options);
	ref.setId(id);
}
 

import com.mongodb.client.gridfs.model.GridFSUploadOptions; //导入依赖的package包/类
@Override
public Observable<ObjectId> uploadFromStream(final String filename, final AsyncInputStream source, final GridFSUploadOptions options) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<ObjectId>>() {
        @Override
        public void apply(final SingleResultCallback<ObjectId> callback) {
            wrapped.uploadFromStream(filename, toCallbackAsyncInputStream(source), options, callback);
        }
    }), observableAdapter);
}
 

import com.mongodb.client.gridfs.model.GridFSUploadOptions; //导入依赖的package包/类
@Override
public Publisher<ObjectId> uploadFromStream(final String filename, final AsyncInputStream source, final GridFSUploadOptions options) {
    return new ObservableToPublisher<ObjectId>(observe(new Block<SingleResultCallback<ObjectId>>() {
        @Override
        public void apply(final SingleResultCallback<ObjectId> callback) {
            wrapped.uploadFromStream(filename, toCallbackAsyncInputStream(source), options, callback);
        }
    }));
}
 

import com.mongodb.client.gridfs.model.GridFSUploadOptions; //导入依赖的package包/类
private GridFSUploadOptions getGridFSUploadOptions(String uniqueId, String fileName, boolean compress, long timestamp, Map<String, String> metadataMap) {
	Document metadata = new Document();
	if (metadataMap != null) {
		for (String key : metadataMap.keySet()) {
			metadata.put(key, metadataMap.get(key));
		}
	}
	metadata.put(TIMESTAMP, timestamp);
	metadata.put(COMPRESSED_FLAG, compress);
	metadata.put(DOCUMENT_UNIQUE_ID_KEY, uniqueId);
	metadata.put(FILE_UNIQUE_ID_KEY, getGridFsId(uniqueId, fileName));

	return new GridFSUploadOptions().chunkSizeBytes(1024).metadata(metadata);
}
 

import com.mongodb.client.gridfs.model.GridFSUploadOptions; //导入依赖的package包/类
public FileStoreBucket() {
	gridFSUploadOptions = new GridFSUploadOptions();
	gridFSUploadOptions.chunkSizeBytes(CHUNK_SIZE_BYTES);
}
 

import com.mongodb.client.gridfs.model.GridFSUploadOptions; //导入依赖的package包/类
@Override
public GridFSUploadStream openUploadStream(final BsonValue id, final String filename, final GridFSUploadOptions options) {
    return new GridFSUploadStreamImpl(wrapped.openUploadStream(id, filename, options), observableAdapter);
}
 

import com.mongodb.client.gridfs.model.GridFSUploadOptions; //导入依赖的package包/类
@Override
public GridFSUploadStream openUploadStream(final String filename) {
    return openUploadStream(filename, new GridFSUploadOptions());
}
 

import com.mongodb.client.gridfs.model.GridFSUploadOptions; //导入依赖的package包/类
/**
 * Opens a AsyncOutputStream that the application can write the contents of the file to.
 * <p>
 * As the application writes the contents to the returned Stream, the contents are uploaded as chunks in the chunks collection. When
 * the application signals it is done writing the contents of the file by calling close on the returned Stream, a files collection
 * document is created in the files collection.
 * </p>
 *
 * @param filename the filename for the stream
 * @param options  the GridFSUploadOptions
 * @return the GridFSUploadStream that provides the ObjectId for the file to be uploaded and the Stream to which the
 * application will write the contents.
 */
GridFSUploadStream openUploadStream(String filename, GridFSUploadOptions options);
 

import com.mongodb.client.gridfs.model.GridFSUploadOptions; //导入依赖的package包/类
/**
 * Uploads the contents of the given {@code AsyncInputStream} to a GridFS bucket.
 * <p>
 * Reads the contents of the user file from the {@code source} and uploads it as chunks in the chunks collection. After all the
 * chunks have been uploaded, it creates a files collection document for {@code filename} in the files collection.
 * </p>
 *
 * @param filename the filename for the stream
 * @param source   the Stream providing the file data
 * @param options  the GridFSUploadOptions
 * @return an observable with a single element, the ObjectId of the uploaded file.
 */
Observable<ObjectId> uploadFromStream(String filename, AsyncInputStream source, GridFSUploadOptions options);
 

import com.mongodb.client.gridfs.model.GridFSUploadOptions; //导入依赖的package包/类
/**
 * Uploads the contents of the given {@code AsyncInputStream} to a GridFS bucket.
 * <p>
 * Reads the contents of the user file from the {@code source} and uploads it as chunks in the chunks collection. After all the
 * chunks have been uploaded, it creates a files collection document for {@code filename} in the files collection.
 * </p>
 *
 * @param filename the filename for the stream
 * @param source   the Stream providing the file data
 * @param options  the GridFSUploadOptions
 * @return a publisher with a single element, the ObjectId of the uploaded file.
 */
Publisher<ObjectId> uploadFromStream(String filename, AsyncInputStream source, GridFSUploadOptions options);