@Override
public Source<Long, NotUsed> count(final Query query,
        @Nullable final List<String> authorizationSubjectIds) {

    checkNotNull(query, "query");

    final BsonDocument queryFilter = getMongoFilter(query, authorizationSubjectIds);
    log.debug("count with query filter <{}>.", queryFilter);

    final CountOptions countOptions = new CountOptions()
            .skip(query.getSkip())
            .limit(query.getLimit())
            .maxTime(maxQueryTime.getSeconds(), TimeUnit.SECONDS);

    return Source.fromPublisher(collection.count(queryFilter, countOptions))
            .mapError(handleMongoExecutionTimeExceededException())
            .log("count");
}
 

/**
 * Updates the document if the document's ETAG is matching the given etag (conditional put).
 * <p>
 * Using this method requires that the document contains an "etag" field that is updated if
 * the document is changed.
 * </p>
 *
 * @param value    the new value
 * @param eTag     the etag used for conditional update
 * @param maxTime  max time for the update
 * @param timeUnit the time unit for the maxTime value
 * @return {@link UpdateIfMatchResult}
 */
public UpdateIfMatchResult updateIfMatch(final V value,
                                         final String eTag,
                                         final long maxTime,
                                         final TimeUnit timeUnit) {
    final K key = keyOf(value);
    if (key != null) {
        final Bson query = and(eq(AbstractMongoRepository.ID, key), eq(ETAG, eTag));

        final Document updatedETaggable = collectionWithWriteTimeout(maxTime, timeUnit).findOneAndReplace(query, encode(value), new FindOneAndReplaceOptions().returnDocument(AFTER));
        if (isNull(updatedETaggable)) {
            final boolean documentExists = collection()
                    .countDocuments(eq(AbstractMongoRepository.ID, key), new CountOptions().maxTime(maxTime, timeUnit)) != 0;
            if (documentExists) {
                return CONCURRENTLY_MODIFIED;
            }

            return NOT_FOUND;
        }

        return OK;
    } else {
        throw new IllegalArgumentException("Key must not be null");
    }
}
 

@Test
void testInsertionOfManyDocumentsAndQueries() {
    ReactiveMongoDatabase database = client.getDatabase(DATABASE);
    ReactiveMongoCollection<Document> collection = database.getCollection(randomAlphaString(8));

    List<Document> documents = new ArrayList<>();
    for (int i = 0; i < 100; i++) {
        documents.add(new Document("i", i));
    }

    collection.insertMany(documents).await().indefinitely();
    Long count = collection.countDocuments().await().indefinitely();
    Long countWithOption = collection.countDocuments(new Document(), new CountOptions().limit(10))
            .await().indefinitely();
    Long estimated = collection.estimatedDocumentCount().await().indefinitely();
    Long estimatedWithOptions = collection
            .estimatedDocumentCount(new EstimatedDocumentCountOptions().maxTime(10, TimeUnit.SECONDS))
            .await().indefinitely();
    assertThat(count).isEqualTo(100);
    assertThat(countWithOption).isEqualTo(10);
    assertThat(estimated).isEqualTo(100);
    assertThat(estimatedWithOptions).isEqualTo(100);

    count = collection.countDocuments(eq("i", 10)).await().indefinitely();
    assertThat(count).isEqualTo(1);

    Optional<Document> document = collection.find().collectItems().first().await().asOptional().indefinitely();
    assertThat(document).isNotEmpty().hasValueSatisfying(doc -> assertThat(doc.get("i", 0)));

    document = collection.find(eq("i", 20)).collectItems().first().await().asOptional().indefinitely();
    assertThat(document).isNotEmpty().hasValueSatisfying(doc -> assertThat(doc.get("i", 20)));

    List<Document> list = collection.find().collectItems().asList().await().indefinitely();
    assertThat(list).hasSize(100);

    list = collection.find(gt("i", 50)).collectItems().asList().await().indefinitely();
    assertThat(list).hasSize(49);
}
 

/**
 * Counts the number of documents in the collection according to the given options.
 *
 * @param filter  the query filter
 * @param options the options describing the count
 * @return the number of documents in the collection
 */
long count(final MongoNamespace namespace, final Bson filter, final CountOptions options) {
  this.waitUntilInitialized();

  try {
    ongoingOperationsGroup.enter();
    return getLocalCollection(namespace).countDocuments(filter, options);
  } finally {
    ongoingOperationsGroup.exit();
  }
}
 

CountOperation(
    final MongoNamespace namespace,
    final DataSynchronizer dataSynchronizer,
    final Bson filter,
    final CountOptions countOptions
) {
  this.namespace = namespace;
  this.dataSynchronizer = dataSynchronizer;;
  this.filter = filter;
  this.countOptions = countOptions;
}
 

CountOperation count(final Bson filter, final SyncCountOptions countOptions) {
  return new CountOperation(
      namespace,
      dataSynchronizer,
      filter,
      new CountOptions().limit(countOptions.getLimit()));
}
 

@Override
public Observable<Long> count(final Bson filter, final CountOptions options) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<Long>>() {
        @Override
        public void apply(final SingleResultCallback<Long> callback) {
            wrapped.count(filter, options, callback);
        }
    }), observableAdapter);
}
 

@Override
@Deprecated
@SuppressWarnings("deprecation")
public Publisher<Long> count(final Bson filter, final CountOptions options) {
    return new ObservableToPublisher<Long>(com.mongodb.async.client.Observables.observe(
            new Block<com.mongodb.async.SingleResultCallback<Long>>() {
                @Override
                public void apply(final com.mongodb.async.SingleResultCallback<Long> callback) {
                    wrapped.count(filter, options, callback);
                }
            }));
}
 

@Override
@Deprecated
@SuppressWarnings("deprecation")
public Publisher<Long> count(final ClientSession clientSession, final Bson filter, final CountOptions options) {
    return new ObservableToPublisher<Long>(com.mongodb.async.client.Observables.observe(
            new Block<com.mongodb.async.SingleResultCallback<Long>>() {
                @Override
                public void apply(final com.mongodb.async.SingleResultCallback<Long> callback) {
                    wrapped.count(clientSession.getWrapped(), filter, options, callback);
                }
            }));
}
 

@Override
public Publisher<Long> countDocuments(final Bson filter, final CountOptions options) {
    return new ObservableToPublisher<Long>(com.mongodb.async.client.Observables.observe(
            new Block<com.mongodb.async.SingleResultCallback<Long>>() {
                @Override
                public void apply(final com.mongodb.async.SingleResultCallback<Long> callback) {
                    wrapped.countDocuments(filter, options, callback);
                }
            }));
}
 

@Override
public Publisher<Long> countDocuments(final ClientSession clientSession, final Bson filter, final CountOptions options) {
    return new ObservableToPublisher<Long>(com.mongodb.async.client.Observables.observe(
            new Block<com.mongodb.async.SingleResultCallback<Long>>() {
                @Override
                public void apply(final com.mongodb.async.SingleResultCallback<Long> callback) {
                    wrapped.countDocuments(clientSession.getWrapped(), filter, options, callback);
                }
            }));
}
 

@Override
public CountOptions convert(Environment env, TraceInfo trace, Memory arg) throws Throwable {
    if (arg.isNull()) return null;

    ArrayMemory arr = arg.toValue(ArrayMemory.class);
    CountOptions options = new CountOptions();

    if (arr.containsKey("skip")) options.skip(arg.valueOfIndex("skip").toInteger());
    if (arr.containsKey("limit")) options.limit(arg.valueOfIndex("limit").toInteger());
    if (arr.containsKey("maxTime")) {
        options.maxTime(WrapTimer.parsePeriod(arg.valueOfIndex("maxTime").toString()), TimeUnit.MILLISECONDS);
    }

    return options;
}
 

@Override
public Memory unconvert(Environment env, TraceInfo trace, CountOptions arg) throws Throwable {
    if (arg == null) return Memory.NULL;

    ArrayMemory options = ArrayMemory.createHashed(12);

    options.put("skip", arg.getSkip());
    options.put("limit", arg.getLimit());
    options.put("maxTime", arg.getMaxTime(TimeUnit.MILLISECONDS));

    return options;
}
 

@Override
public Uni<Long> countDocuments(Bson filter, CountOptions options) {
    return Wrappers.toUni(collection.countDocuments(filter, options));
}
 

@Override
public Uni<Long> countDocuments(ClientSession clientSession, Bson filter, CountOptions options) {
    return Wrappers.toUni(collection.countDocuments(clientSession, filter, options));
}
 

@Override
public Observable<Long> count() {
    return count(new BsonDocument(), new CountOptions());
}
 

@Override
public Observable<Long> count(final Bson filter) {
    return count(filter, new CountOptions());
}
 

public long size(final long maxTime, final TimeUnit timeUnit) {
    return collection().countDocuments(new BsonDocument(), new CountOptions().maxTime(maxTime, timeUnit));
}
 

@Override
@Deprecated
public Publisher<Long> count() {
    return count(new BsonDocument(), new CountOptions());
}
 

@Override
@Deprecated
public Publisher<Long> count(final Bson filter) {
    return count(filter, new CountOptions());
}
 

@Override
@Deprecated
public Publisher<Long> count(final ClientSession clientSession) {
    return count(clientSession, new BsonDocument(), new CountOptions());
}
 

@Override
@Deprecated
public Publisher<Long> count(final ClientSession clientSession, final Bson filter) {
    return count(clientSession, filter, new CountOptions());
}
 

@Override
public Publisher<Long> countDocuments(final Bson filter) {
    return countDocuments(filter, new CountOptions());
}
 

@Override
public Publisher<Long> countDocuments(final ClientSession clientSession, final Bson filter) {
    return countDocuments(clientSession, filter, new CountOptions());
}
 

@Signature
public long count(BasicDBObject filter, CountOptions options) {
    return getWrappedObject().count(filter, options);
}
 

@Override
public Class<?>[] getOperationClasses() {
    return new Class[] { CountOptions.class };
}
 

/**
 * Counts the number of documents in the collection according to the given options.
 *
 * @param filter the query filter
 * @param options the options describing the count
 * @return a {@link Uni} completed with the number of documents
 */
Uni<Long> countDocuments(Bson filter, CountOptions options);
 

/**
 * Counts the number of documents in the collection according to the given options.
 *
 * @param clientSession the client session with which to associate this operation
 * @param filter the query filter
 * @param options the options describing the count
 * @return a {@link Uni} completed with the number of documents
 */
Uni<Long> countDocuments(ClientSession clientSession, Bson filter, CountOptions options);
 

/**
 * Counts the number of documents in the collection according to the given options.
 *
 * @param filter the query filter
 * @return the number of documents in the collection
 */
long count(final MongoNamespace namespace, final Bson filter) {
  return count(namespace, filter, new CountOptions());
}
 

/**
 * Counts the number of documents in the collection according to the given options.
 *
 * @param filter  the query filter
 * @param options the options describing the count
 * @return an Observable with a single element indicating the number of documents
 */
Observable<Long> count(Bson filter, CountOptions options);