Java 类com.mongodb.client.gridfs.model.GridFSFile 实例源码

@Override
protected void validateAfterStoreAdd(TestContext context, Vertx vertx,
    String path, Handler<AsyncResult<Void>> handler) {
  vertx.executeBlocking(f -> {
    try (MongoClient client = new MongoClient(mongoConnector.serverAddress)) {
      MongoDatabase db = client.getDatabase(MongoDBTestConnector.MONGODB_DBNAME);
      GridFSBucket gridFS = GridFSBuckets.create(db);

      GridFSFindIterable files = gridFS.find();

      GridFSFile file = files.first();
      ByteArrayOutputStream baos = new ByteArrayOutputStream();
      gridFS.downloadToStream(file.getFilename(), baos);
      String contents = new String(baos.toByteArray(), StandardCharsets.UTF_8);
      context.assertEquals(CHUNK_CONTENT, contents);
    }
    f.complete();
  }, handler);
}

public boolean checkAndCleanup(String userId, String fileName) {
    List<Document> l = query(new Query().equals(USERID, userId).addSortCriteria(CREATION_TIME, false));
    if (l.size() >= maxPersistedPagesPerUser) {
        Document oldest = l.iterator().next();
        if ((System.currentTimeMillis() - oldest.get(CREATION_TIME).getTime()) < minimumDelay) {
            //there have been to many page persistences for this user in a short time, so don't persist
            return false;
        } else {
            //clean up oldest to free space for new persisted page
            gridFS.find(Filters.eq("filename", oldest.get(FILENAME))).forEach(new Block<GridFSFile>() {
                @Override
                public void apply(GridFSFile file) {
                    gridFS.delete(file.getObjectId());
                }
            });
            oldest.delete();
        }
    }
    //create new entry
    Document newOne = createNew();
    newOne.set(USERID, userId);
    newOne.set(FILENAME, fileName);
    newOne.set(CREATION_TIME, new Date());
    newOne.writeToDatabase(false);
    return true;
}

@Override
public void byPages(int pageSize, Callback callback) {
    MongoCursor<GridFSFile> cursor = bucket.find().iterator();
    boolean loop = true;
    try {
        while (loop) {
            List<GridFSFile> files = new ArrayList<>(batchSize);
            int i = 0;
            while (cursor.hasNext() && i < batchSize) {
                files.add(cursor.next());
                i++;
            }
            loop = callback.on(files.stream().map(file -> bucket.openDownloadStream(file.getObjectId())).map(fromFile).collect(Collectors.toList()));
        }
    } finally {
        cursor.close();
    }
}

@Override
public InputStream getAssociatedDocumentStream(String uniqueId, String fileName) {
    GridFSBucket gridFS = createGridFSConnection();
    GridFSFile file = gridFS.find(new Document(ASSOCIATED_METADATA + "." + FILE_UNIQUE_ID_KEY, getGridFsId(uniqueId, fileName))).first();

    if (file == null) {
        return null;
    }

    InputStream is = gridFS.openDownloadStream(file.getObjectId());
    ;

    Document metadata = file.getMetadata();
    if (metadata.containsKey(COMPRESSED_FLAG)) {
        boolean compressed = (boolean) metadata.remove(COMPRESSED_FLAG);
        if (compressed) {
            is = new InflaterInputStream(is);
        }
    }

    return is;
}

public Document findMetadata(String oid) {
    GridFSFile first = fileStore.find(eq("_id", new ObjectId(oid))).first();
    if(first == null) {
        return null;
    } else {
        return first.getMetadata();
    }
}

private SmofGridRef toSmofGridRef(BsonDocument refBson) {
    final String bucketName = refBson.getString("bucket").getValue();
    final ObjectId id = refBson.getObjectId("id").getValue();
    final SmofGridRef ref = SmofGridRefFactory.newFromDB(id, bucketName);
    final GridFSFile file = dispatcher.loadMetadata(ref);
    ref.putMetadata(file.getMetadata());
    return ref;
}

@Test
public final void testMetadata() throws IOException {
    final Document metadata = new Document("randomkey", 45);
    ref.putMetadata(metadata);
    streamManager.uploadFile(ref);
    ref.putMetadata(new Document());
    final GridFSFile file = streamManager.loadFileMetadata(ref);
    assertEquals(metadata, file.getMetadata());
}

public void deleteFile(Document mo) {
        String filename = mo.get(this);
        if (filename != null) {
            fileSystemProvider.getFileSystem().find(Filters.eq("filename", filename)).forEach(new Block<GridFSFile>() {
                @Override
                public void apply(GridFSFile file) {
                    fileSystemProvider.getFileSystem().delete(file.getObjectId());
                }
            });
            mo.getDataObject().remove(this.getName());
//            mo.writeToDatabase(false);
        }
    }

@Override
public Observable<GridFSFile> first() {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<GridFSFile>>(){
        @Override
        public void apply(final SingleResultCallback<GridFSFile> callback) {
            wrapped.first(callback);
        }
    }), observableAdapter);
}

@Override
public Observable<GridFSFile> getGridFSFile() {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<GridFSFile>>() {
        @Override
        public void apply(final SingleResultCallback<GridFSFile> callback) {
            wrapped.getGridFSFile(callback);
        }
    }), observableAdapter);
}

@Override
public Publisher<GridFSFile> getGridFSFile() {
    return new ObservableToPublisher<GridFSFile>(observe(new Block<SingleResultCallback<GridFSFile>>() {
        @Override
        public void apply(final SingleResultCallback<GridFSFile> callback) {
            wrapped.getGridFSFile(callback);
        }
    }));
}

@Override
public Publisher<GridFSFile> first() {
    return new ObservableToPublisher<GridFSFile>(observe(new Block<SingleResultCallback<GridFSFile>>(){
        @Override
        public void apply(final SingleResultCallback<GridFSFile> callback) {
            wrapped.first(callback);
        }
    }));
}

@Override
public OperationResult deleteFile(
        final Database db,
        final String dbName,
        final String bucketName,
        final BsonValue fileId,
        final String requestEtag,
        final boolean checkEtag) {
    final String bucket = extractBucketName(bucketName);

    GridFSBucket gridFSBucket = GridFSBuckets.create(
            db.getDatabase(dbName),
            bucket);

    GridFSFile file = gridFSBucket
            .find(eq("_id", fileId))
            .limit(1).iterator().tryNext();

    if (file == null) {
        return new OperationResult(HttpStatus.SC_NOT_FOUND);
    } else if (checkEtag) {
        Object oldEtag = file.getMetadata().get("_etag");

        if (oldEtag != null) {
            if (requestEtag == null) {
                return new OperationResult(HttpStatus.SC_CONFLICT, oldEtag);
            } else if (!Objects.equals(oldEtag.toString(), requestEtag)) {
                return new OperationResult(
                        HttpStatus.SC_PRECONDITION_FAILED, oldEtag);
            }
        }
    }

    gridFSBucket.delete(fileId);

    return new OperationResult(HttpStatus.SC_NO_CONTENT);
}

@Override
public void handleRequest(
        HttpServerExchange exchange,
        RequestContext context)
        throws Exception {
    if (context.isInError()) {
        next(exchange, context);
        return;
    }

    LOGGER.trace("GET " + exchange.getRequestURL());
    final String bucket = extractBucketName(context.getCollectionName());

    GridFSBucket gridFSBucket = GridFSBuckets.create(
            MongoDBClientSingleton.getInstance().getClient()
                    .getDatabase(context.getDBName()),
            bucket);

    GridFSFile dbsfile = gridFSBucket
            .find(eq("_id", context.getDocumentId()))
            .limit(1).iterator().tryNext();

    if (dbsfile == null) {
        fileNotFound(context, exchange);
    } else if (!checkEtag(exchange, dbsfile)) {
        sendBinaryContent(context, gridFSBucket, dbsfile, exchange);
    }

    next(exchange, context);
}

private boolean checkEtag(HttpServerExchange exchange, GridFSFile dbsfile) {
    if (dbsfile != null) {
        Object etag;

        if (dbsfile.getMetadata() != null
                && dbsfile.getMetadata().containsKey("_etag")) {
            etag = dbsfile.getMetadata().get("_etag");
        } else {
            etag = null;
        }

        if (etag != null && etag instanceof ObjectId) {
            ObjectId _etag = (ObjectId) etag;

            BsonObjectId __etag = new BsonObjectId(_etag);

            // in case the request contains the IF_NONE_MATCH header with the current etag value,
            // just return 304 NOT_MODIFIED code
            if (RequestHelper.checkReadEtag(exchange, __etag)) {
                exchange.setStatusCode(HttpStatus.SC_NOT_MODIFIED);
                exchange.endExchange();
                return true;
            }
        }
    }

    return false;
}

@Override
public List<AssociatedDocument> getAssociatedDocuments(String uniqueId, FetchType fetchType) throws Exception {
    GridFSBucket gridFS = createGridFSConnection();
    List<AssociatedDocument> assocDocs = new ArrayList<>();
    if (!FetchType.NONE.equals(fetchType)) {
        GridFSFindIterable files = gridFS.find(new Document(ASSOCIATED_METADATA + "." + DOCUMENT_UNIQUE_ID_KEY, uniqueId));
        for (GridFSFile file : files) {
            AssociatedDocument ad = loadGridFSToAssociatedDocument(gridFS, file, fetchType);
            assocDocs.add(ad);
        }

    }
    return assocDocs;
}

@Override
public AssociatedDocument getAssociatedDocument(String uniqueId, String fileName, FetchType fetchType) throws Exception {
    GridFSBucket gridFS = createGridFSConnection();
    if (!FetchType.NONE.equals(fetchType)) {
        GridFSFile file = gridFS.find(new Document(ASSOCIATED_METADATA + "." + FILE_UNIQUE_ID_KEY, getGridFsId(uniqueId, fileName))).first();
        if (null != file) {
            return loadGridFSToAssociatedDocument(gridFS, file, fetchType);
        }
    }
    return null;
}

private AssociatedDocument loadGridFSToAssociatedDocument(GridFSBucket gridFS, GridFSFile file, FetchType fetchType) throws IOException {
    AssociatedDocument.Builder aBuilder = AssociatedDocument.newBuilder();
    aBuilder.setFilename(file.getFilename());
    Document metadata = file.getMetadata();

    boolean compressed = false;
    if (metadata.containsKey(COMPRESSED_FLAG)) {
        compressed = (boolean) metadata.remove(COMPRESSED_FLAG);
    }

    long timestamp = (long) metadata.remove(TIMESTAMP);

    aBuilder.setCompressed(compressed);
    aBuilder.setTimestamp(timestamp);

    aBuilder.setDocumentUniqueId((String) metadata.remove(DOCUMENT_UNIQUE_ID_KEY));
    for (String field : metadata.keySet()) {
        aBuilder.addMetadata(Metadata.newBuilder().setKey(field).setValue((String) metadata.get(field)));
    }

    if (FetchType.FULL.equals(fetchType)) {

        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        gridFS.downloadToStream(file.getObjectId(), byteArrayOutputStream);
        byte[] bytes = byteArrayOutputStream.toByteArray();
        if (null != bytes) {
            if (compressed) {
                bytes = CommonCompression.uncompressZlib(bytes);
            }
            aBuilder.setDocument(ByteString.copyFrom(bytes));
        }
    }
    aBuilder.setIndexName(indexName);
    return aBuilder.build();
}

@Override
public List<String> getAssociatedFilenames(String uniqueId) throws Exception {
    GridFSBucket gridFS = createGridFSConnection();
    ArrayList<String> fileNames = new ArrayList<>();
    gridFS.find(new Document(ASSOCIATED_METADATA + "." + DOCUMENT_UNIQUE_ID_KEY, uniqueId))
            .forEach((Consumer<com.mongodb.client.gridfs.model.GridFSFile>) gridFSFile -> fileNames.add(gridFSFile.getFilename()));

    return fileNames;
}

@Override
public void deleteAssociatedDocument(String uniqueId, String fileName) {
    GridFSBucket gridFS = createGridFSConnection();
    gridFS.find(new Document(ASSOCIATED_METADATA + "." + FILE_UNIQUE_ID_KEY, getGridFsId(uniqueId, fileName)))
            .forEach((Block<com.mongodb.client.gridfs.model.GridFSFile>) gridFSFile -> gridFS.delete(gridFSFile.getObjectId()));

}

@Override
public long flush(long expiration) throws IOException {
    rwl.r.lock();
    try {
        final Bson filter = expiration == 0 ? Filters.exists("uri") : Filters.lt("_id",
                new ObjectId(new Date(expiration)));
        indexedCollection.deleteMany(filter);
        for (GridFSFile f : contentGrid.find(filter))
            contentGrid.delete(f.getObjectId());
        long l = metaCollection.deleteMany(filter).getDeletedCount();
        return l;
    } finally {
        rwl.r.unlock();
    }
}

@Override
public GridFSFile loadFileMetadata(SmofGridRef ref) {
    final GridFSBucket bucket = pool.getBucket(ref.getBucketName());
    return bucket.find(Filters.eq(Element.ID, ref.getId())).first();
}

@Override
   public GridFSFile loadMetadata(SmofGridRef ref) {
    return streamManager.loadFileMetadata(ref);
}

public GridFSFile gridFSFile() {
    return gridFSFile(database, bucket, objectId);
}

@BeanCodecKey(ignore = true)
public GridFSFile getGridFSFile(MongoDatabase mongoDatabase) {
    return GridFSBuckets.create(mongoDatabase, bucket).find(Filters.eq("_id", fileObjectId)).first();
}

@Override
public Observable<GridFSFile> toObservable() {
    return RxObservables.create(Observables.observe(wrapped), observableAdapter);
}

@Override
public Subscription subscribe(final Subscriber<? super GridFSFile> s) {
    return toObservable().subscribe(s);
}

@Override
public void subscribe(final Subscriber<? super GridFSFile> s) {
    new ObservableToPublisher<GridFSFile>(observe(wrapped)).subscribe(s);
}

private void sendBinaryContent(
        final RequestContext context,
        final GridFSBucket gridFSBucket,
        final GridFSFile file,
        final HttpServerExchange exchange)
        throws IOException {
    LOGGER.trace("Filename = {}", file.getFilename());
    LOGGER.trace("Content length = {}", file.getLength());

    if (file.getMetadata() != null
            && file.getMetadata().get("contentType") != null) {
        exchange.getResponseHeaders().put(Headers.CONTENT_TYPE,
                file.getMetadata().get("contentType").toString());
    } else if (file.getMetadata() != null
            && file.getMetadata().get("contentType") != null) {
        exchange.getResponseHeaders().put(Headers.CONTENT_TYPE,
                file.getMetadata().get("contentType").toString());
    } else {
        exchange.getResponseHeaders().put(
                Headers.CONTENT_TYPE,
                APPLICATION_OCTET_STREAM);
    }

    exchange.getResponseHeaders().put(
            Headers.CONTENT_LENGTH,
            file.getLength());

    exchange.getResponseHeaders().put(
            Headers.CONTENT_DISPOSITION,
            String.format("inline; filename=\"%s\"",
                    extractFilename(file)));

    exchange.getResponseHeaders().put(
            Headers.CONTENT_TRANSFER_ENCODING,
            CONTENT_TRANSFER_ENCODING_BINARY);

    ResponseHelper.injectEtagHeader(exchange, file.getMetadata());

    context.setResponseStatusCode(HttpStatus.SC_OK);

    gridFSBucket.downloadToStream(
            file.getId(),
            exchange.getOutputStream());
}

private String extractFilename(final GridFSFile dbsfile) {
    return dbsfile.getFilename() != null
            ? dbsfile.getFilename()
            : dbsfile.getId().toString();
}

public void getAssociatedDocuments(OutputStream outputstream, Document filter) throws IOException {
    Charset charset = Charset.forName("UTF-8");

    GridFSBucket gridFS = createGridFSConnection();
    GridFSFindIterable gridFSFiles = gridFS.find(filter);
    outputstream.write("{\n".getBytes(charset));
    outputstream.write(" \"associatedDocs\": [\n".getBytes(charset));

    boolean first = true;
    for (GridFSFile gridFSFile : gridFSFiles) {
        if (first) {
            first = false;
        }
        else {
            outputstream.write(",\n".getBytes(charset));
        }

        Document metadata = gridFSFile.getMetadata();

        String uniqueId = metadata.getString(DOCUMENT_UNIQUE_ID_KEY);
        String uniquieIdKeyValue = "  { \"uniqueId\": \"" + uniqueId + "\", ";
        outputstream.write(uniquieIdKeyValue.getBytes(charset));

        String filename = gridFSFile.getFilename();
        String filenameKeyValue = "\"filename\": \"" + filename + "\", ";
        outputstream.write(filenameKeyValue.getBytes(charset));

        Date uploadDate = gridFSFile.getUploadDate();
        String uploadDateKeyValue = "\"uploadDate\": {\"$date\":" + uploadDate.getTime() + "}";
        outputstream.write(uploadDateKeyValue.getBytes(charset));

        metadata.remove(TIMESTAMP);
        metadata.remove(COMPRESSED_FLAG);
        metadata.remove(DOCUMENT_UNIQUE_ID_KEY);
        metadata.remove(FILE_UNIQUE_ID_KEY);

        if (!metadata.isEmpty()) {
            String metaJson = metadata.toJson();
            String metaString = ", \"meta\": " + metaJson;
            outputstream.write(metaString.getBytes(charset));
        }

        outputstream.write(" }".getBytes(charset));

    }
    outputstream.write("\n ]\n}".getBytes(charset));
}

@Override
public void deleteAssociatedDocuments(String uniqueId) {
    GridFSBucket gridFS = createGridFSConnection();
    gridFS.find(new Document(ASSOCIATED_METADATA + "." + DOCUMENT_UNIQUE_ID_KEY, uniqueId))
            .forEach((Block<com.mongodb.client.gridfs.model.GridFSFile>) gridFSFile -> gridFS.delete(gridFSFile.getObjectId()));
}

public static GridFSFile gridFSFile(MongoDatabase database, String bucket, ObjectId objectId) {
    return  GridFSBuckets.create(database, bucket).find(Filters.eq("_id", objectId)).first();

}

/**
 * Gets the corresponding {@link GridFSFile} for the file being downloaded
 *
 * @return an observable with a single element, the corresponding GridFSFile for the file being downloaded
 */
Observable<GridFSFile> getGridFSFile();

/**
 * Helper to return an observable limited first from the query.
 *
 * @return an observable with a single element
 */
Observable<GridFSFile> first();

/**
 * Gets the corresponding {@link GridFSFile} for the file being downloaded
 *
 * @return a publisher with a single element, the corresponding GridFSFile for the file being downloaded
 */
Publisher<GridFSFile> getGridFSFile();

/**
 * Helper to return a publisher limited first from the query.
 *
 * @return a publisher with a single element
 */
Publisher<GridFSFile> first();

GridFSFile loadFileMetadata(SmofGridRef ref);

GridFSFile loadMetadata(SmofGridRef ref);