private static Source<Success, NotUsed> repeatableCreateCappedCollectionSource(
        final MongoDatabase database,
        final String collectionName,
        final long cappedCollectionSizeInBytes) {

    final CreateCollectionOptions collectionOptions = new CreateCollectionOptions()
            .capped(true)
            .sizeInBytes(cappedCollectionSizeInBytes)
            .maxDocuments(1);

    return Source.lazily(
            () -> Source.fromPublisher(database.createCollection(collectionName, collectionOptions)))
            .mapMaterializedValue(whatever -> NotUsed.getInstance())
            .withAttributes(Attributes.inputBuffer(1, 1))
            .recoverWithRetries(1, new PFBuilder<Throwable, Source<Success, NotUsed>>()
                    .match(MongoCommandException.class,
                            MongoTimestampPersistence::isCollectionAlreadyExistsError,
                            error -> Source.single(Success.SUCCESS))
                    .build());

}
 

@Override
public void initialize(final RyaDetails details) throws AlreadyInitializedException, RyaDetailsRepositoryException {
    // Preconditions.
    requireNonNull( details );

    if(!details.getRyaInstanceName().equals( instanceName )) {
        throw new RyaDetailsRepositoryException("The instance name that was in the provided 'details' does not match " +
                "the instance name that this repository is connected to. Make sure you're connected to the" +
                "correct Rya instance.");
    }

    if(isInitialized()) {
        throw new AlreadyInitializedException("The repository has already been initialized for the Rya instance named '" +
                instanceName + "'.");
    }

    // Create the document that hosts the details if it has not been created yet.
    db.createCollection(INSTANCE_DETAILS_COLLECTION_NAME, new CreateCollectionOptions());
    final MongoCollection<Document> col = db.getCollection(INSTANCE_DETAILS_COLLECTION_NAME);

    // Write the details to the collection.
    col.insertOne(MongoDetailsAdapter.toDocument(details), new InsertOneOptions());
}
 

void enableValidation(final MappedClass mc, final Validation validation) {
    if (validation != null) {
        String collectionName = mc.getCollectionName();
        try {
            getDatabase().runCommand(new Document("collMod", collectionName)
                                         .append("validator", parse(validation.value()))
                                         .append("validationLevel", validation.level().getValue())
                                         .append("validationAction", validation.action().getValue()));
        } catch (MongoCommandException e) {
            if (e.getCode() == 26) {
                getDatabase().createCollection(collectionName,
                    new CreateCollectionOptions()
                        .validationOptions(new ValidationOptions()
                                               .validator(parse(validation.value()))
                                               .validationLevel(validation.level())
                                               .validationAction(validation.action())));
            } else {
                throw e;
            }
        }
    }
}
 

private static MongoCollection<Document> getOrCreateMongoCollection(final MongoDatabase database,
        final String collectionName, final boolean isCapped, final Integer sizeInBytes) {
    try {
        LOGGER.debug("Gettting collection '{}'...", collectionName);
        // throws IllegalArgumentException if collectionName is invalid
        final MongoCollection<Document> found = database.getCollection(collectionName);
        LOGGER.debug("Got collection {}", found);
        return found;
    } catch (final IllegalStateException e) {
        LOGGER.debug("Collection '{}' does not exist.", collectionName);
        final CreateCollectionOptions options = new CreateCollectionOptions().capped(isCapped)
                .sizeInBytes(sizeInBytes);
        LOGGER.debug("Creating collection '{}' with options {}...", collectionName, options);
        database.createCollection(collectionName, options);
        LOGGER.debug("Created collection.");
        final MongoCollection<Document> created = database.getCollection(collectionName);
        LOGGER.debug("Got created collection {}", created);
        return created;
    }

}
 

private static MongoCollection<Document> getOrCreateMongoCollection(final MongoDatabase database,
        final String collectionName, final boolean isCapped, final Integer sizeInBytes) {
    try {
        LOGGER.debug("Gettting collection '{}'...", collectionName);
        // throws IllegalArgumentException if collectionName is invalid
        return database.getCollection(collectionName);
    } catch (final IllegalStateException e) {
        LOGGER.debug("Collection '{}' does not exist.", collectionName);
        final CreateCollectionOptions options = new CreateCollectionOptions()
        // @formatter:off
                .capped(isCapped)
                .sizeInBytes(sizeInBytes);
        // @formatter:on
        LOGGER.debug("Creating collection {} (capped = {}, sizeInBytes = {})", collectionName, isCapped,
                sizeInBytes);
        database.createCollection(collectionName, options);
        return database.getCollection(collectionName);
    }

}
 

@BeforeEach
public void setUp() {
    mongoClient = mongoDBService.getClient();

    MongoDatabase database = mongoClient.getDatabase("testDatabase");

    /*
     The consume operation needs taliable cursors which require capped
     collections
     */
    CreateCollectionOptions options = new CreateCollectionOptions();
    options.capped(true);
    options.sizeInBytes(1024 * 1024);

    database.createCollection("testCollection", options);

    MongoCollection<Document> collection = database.getCollection("testCollection");

    List<Document> documents = new ArrayList<>(expect);
    for (int i = 0; i < expect; i++) {
        Document doc = new Document();

        doc.append("name", "test");
        doc.append("value", "value " + i);

        documents.add(doc);
    }

    collection.insertMany(documents);
}
 

@Test
void testCollectionCreation() {
    ReactiveMongoDatabase database = client.getDatabase(DATABASE);
    database.createCollection("cappedCollection",
            new CreateCollectionOptions().capped(true).sizeInBytes(0x100000)).await().indefinitely();
    assertThat(database.listCollectionNames()
            .collectItems().asList().await().indefinitely()).hasSize(1).containsExactly("cappedCollection");
    assertThat(database.listCollections().map(doc -> doc.getString("name"))
            .collectItems().asList().await().indefinitely()).hasSize(1).containsExactly("cappedCollection");
    assertThat(database.listCollections(Document.class).map(doc -> doc.getString("name"))
            .collectItems().asList().await().indefinitely()).hasSize(1).containsExactly("cappedCollection");

    assertThat(database.getCollection("cappedCollection").getNamespace().getDatabaseName()).isEqualTo(DATABASE);
    assertThat(database.getCollection("cappedCollection").getDocumentClass()).isEqualTo(Document.class);
}
 

@Test
void testCollectionCreationWithOptions() {
    ReactiveMongoDatabase database = client.getDatabase(DATABASE);
    database.createCollection("cappedCollection",
            new CreateCollectionOptions().capped(true).sizeInBytes(0x100000)).await().indefinitely();
    assertThat(database.listCollections().map(doc -> doc.getString("name"))
            .collectItems().asList().await().indefinitely()).hasSize(1).containsExactly("cappedCollection");
}
 

@BeforeClass
public static void doCollectionSetup() {
    // The feature only works with capped collections!
    CreateCollectionOptions opts = new CreateCollectionOptions().capped(true).sizeInBytes(1024 * 1024);
    EmbedMongoConfiguration.getDB().createCollection(COLLECTION, opts);
    LOG.debug("Created a capped collection named {}", COLLECTION);
    EmbedMongoConfiguration.getDB().createCollection(COLLECTION_TRACKING);
    LOG.debug("Created a tracking collection named {}", COLLECTION_TRACKING);
}
 

@Override
public Observable<Success> createCollection(final String collectionName, final CreateCollectionOptions options) {
    return RxObservables.create(Observables.observe(new Block<SingleResultCallback<Success>>() {
        @Override
        public void apply(final SingleResultCallback<Success> callback) {
            wrapped.createCollection(collectionName, options, voidToSuccessCallback(callback));
        }
    }), observableAdapter);
}
 

private MongoCollection<Document> getCollection(String collectionName) {
    // first check if collection exists, if not then create a new collection with cappedSize
    if (!isCollectionExists(collectionName)) {
        CreateCollectionOptions option = new CreateCollectionOptions();
        option.capped(true);
        option.maxDocuments(cappedMaxDocuments);
        option.sizeInBytes(cappedMaxSize);
        db.createCollection(collectionName, option);
    }

    return db.getCollection(collectionName);

}
 

@Override
public Publisher<Success> createCollection(final String collectionName, final CreateCollectionOptions options) {
    return new ObservableToPublisher<Success>(com.mongodb.async.client.Observables.observe(
            new Block<com.mongodb.async.SingleResultCallback<Success>>() {
                @Override
                public void apply(final com.mongodb.async.SingleResultCallback<Success> callback) {
                    wrapped.createCollection(collectionName, options, voidToSuccessCallback(callback));
                }
            }));
}
 

@Override
public Publisher<Success> createCollection(final ClientSession clientSession, final String collectionName,
                                           final CreateCollectionOptions options) {
    return new ObservableToPublisher<Success>(com.mongodb.async.client.Observables.observe(
            new Block<com.mongodb.async.SingleResultCallback<Success>>() {
                @Override
                public void apply(final com.mongodb.async.SingleResultCallback<Success> callback) {
                    wrapped.createCollection(clientSession.getWrapped(), collectionName, options, voidToSuccessCallback(callback));
                }
            }));
}
 

private MongoCollection<Document> createCappedCollection() {
    db.createCollection(COLLECTION_NAME, new CreateCollectionOptions()
      .capped(true)
      .sizeInBytes(100000)
      .maxDocuments(MAX_DOCUMENTS_IN_COLLECTION));
    return db.getCollection(COLLECTION_NAME);
}
 

private MongoDatabase addValidation(final Document validator) {
    ValidationOptions options = new ValidationOptions()
                                    .validator(validator)
                                    .validationLevel(ValidationLevel.MODERATE)
                                    .validationAction(ValidationAction.ERROR);
    MongoDatabase database = getMongoClient().getDatabase(TEST_DB_NAME);
    database.getCollection("validation").drop();
    database.createCollection("validation", new CreateCollectionOptions().validationOptions(options));
    return database;
}
 

@Override
public Uni<Void> createCollection(String collectionName, CreateCollectionOptions options) {
    return Wrappers.toUni(database.createCollection(collectionName, options));
}
 

@Override
public Uni<Void> createCollection(ClientSession clientSession, String collectionName,
        CreateCollectionOptions options) {
    return Wrappers.toUni(database.createCollection(clientSession, collectionName, options));
}
 

@BeforeClass
public static void doCollectionSetup() {
    // The feature only works with capped collections!
    CreateCollectionOptions opts = new CreateCollectionOptions().capped(true).sizeInBytes(1024 * 1024);
    EmbedMongoConfiguration.getDB().createCollection(COLLECTION, opts);
}
 

@Test
public void verifyMetadataJsonSchemaExpected() throws IOException {
    // Given
    String collection = "validSchema";
    Map<String, Object> properties = new HashMap<>();
    properties.put("database", DATABASE);
    properties.put("collection", collection);
    properties.put("host", String.format("%s:%s", EmbedMongoConfiguration.HOST, EmbedMongoConfiguration.PORT));
    properties.put("user", EmbedMongoConfiguration.USER);
    properties.put("password", EmbedMongoConfiguration.PASSWORD);
    properties.put("adminDB", EmbedMongoConfiguration.ADMIN_DB);
    // When
    Document jsonSchema = Document.parse("{ \n"
        + "      bsonType: \"object\", \n"
        + "      required: [ \"name\", \"surname\", \"email\" ], \n"
        + "      properties: { \n"
        + "         name: { \n"
        + "            bsonType: \"string\", \n"
        + "            description: \"required and must be a string\" }, \n"
        + "         surname: { \n"
        + "            bsonType: \"string\", \n"
        + "            description: \"required and must be a string\" }, \n"
        + "         email: { \n"
        + "            bsonType: \"string\", \n"
        + "            pattern: \"^[email protected]+$\", \n"
        + "            description: \"required and must be a valid email address\" }, \n"
        + "         year_of_birth: { \n"
        + "            bsonType: \"int\", \n"
        + "            minimum: 1900, \n"
        + "            maximum: 2018,\n"
        + "            description: \"the value must be in the range 1900-2018\" }, \n"
        + "         gender: { \n"
        + "            enum: [ \"M\", \"F\" ], \n"
        + "            description: \"can be only M or F\" } \n"
        + "      }}");
    ValidationOptions collOptions = new ValidationOptions().validator(Filters.eq("$jsonSchema",jsonSchema));
    client.getDatabase(DATABASE).createCollection(collection,
        new CreateCollectionOptions().validationOptions(collOptions));
    // Then
    MongoDBMetadataRetrieval metaBridge = new MongoDBMetadataRetrieval();
    SyndesisMetadata metadata = metaBridge.fetch(
        context,
        SCHEME,
        CONNECTOR_ID,
        properties
    );
    // format as json the datashape
    JsonNode json = OBJECT_MAPPER.readTree(metadata.outputShape.getSpecification());
    Assertions.assertThat(json.get("$schema").asText()).isEqualTo("http://json-schema.org/schema#");
    Assertions.assertThat(json.get("required").isArray()).isTrue();
    Assertions.assertThat(json.get("properties").isObject()).isTrue();
    Assertions.assertThat(json.get("properties").get("name")).isNotNull();
    Assertions.assertThat(json.get("properties").get("surname")).isNotNull();
    Assertions.assertThat(json.get("properties").get("email")).isNotNull();
    Assertions.assertThat(json.get("properties").get("year_of_birth")).isNotNull();
    Assertions.assertThat(json.get("properties").get("gender")).isNotNull();
}
 

@Override
public Observable<Success> createCollection(final String collectionName) {
    return createCollection(collectionName, new CreateCollectionOptions());
}
 

/**
 * Run this main method to see the output of this quick example.
 *
 * @param args takes an optional single argument for the connection string
 * @throws Throwable if an operation fails
 */
public static void main(final String[] args) throws Throwable {
    MongoClient mongoClient;

    if (args.length == 0) {
        // connect to the local database server
        mongoClient = MongoClients.create();
    } else {
        mongoClient = MongoClients.create(args[0]);
    }

    // get handle to "mydb" database
    MongoDatabase database = mongoClient.getDatabase("mydb");


    // get a handle to the "test" collection
    MongoCollection<Document> collection = database.getCollection("test");
    ObservableSubscriber subscriber = new ObservableSubscriber<Success>();
    collection.drop().subscribe(subscriber);
    subscriber.await();

    // getting a list of databases
    mongoClient.listDatabaseNames().subscribe(new PrintSubscriber<String>("Database Names: %s"));

    // drop a database
    subscriber = new ObservableSubscriber<Success>();
    mongoClient.getDatabase("databaseToBeDropped").drop().subscribe(subscriber);
    subscriber.await();

    // create a collection
    database.createCollection("cappedCollection", new CreateCollectionOptions().capped(true).sizeInBytes(0x100000))
                .subscribe(new PrintSubscriber<Success>("Creation Created!"));


    database.listCollectionNames().subscribe(new PrintSubscriber<String>("Collection Names: %s"));

    // drop a collection:
    subscriber = new ObservableSubscriber<Success>();
    collection.drop().subscribe(subscriber);
    subscriber.await();

    // create an ascending index on the "i" field
    collection.createIndex(new Document("i", 1)).subscribe(new PrintSubscriber<String>("Created an index named: %s"));

    // list the indexes on the collection
    collection.listIndexes().subscribe(new PrintDocumentSubscriber());


    // create a text index on the "content" field
    subscriber = new PrintSubscriber<String>("Created an index named: %s");
    collection.createIndex(new Document("content", "text")).subscribe(subscriber);
    subscriber.await();

    subscriber = new OperationSubscriber();
    collection.insertMany(asList(new Document("_id", 0).append("content", "textual content"),
            new Document("_id", 1).append("content", "additional content"),
            new Document("_id", 2).append("content", "irrelevant content"))).subscribe(subscriber);
    subscriber.await();

    // Find using the text index
    subscriber = new PrintSubscriber("Text search matches: %s");
    collection.countDocuments(text("textual content -irrelevant")).subscribe(subscriber);
    subscriber.await();

    // Find using the $language operator
    subscriber = new PrintSubscriber("Text search matches (english): %s");
    Bson textSearch = text("textual content -irrelevant", new TextSearchOptions().language("english"));
    collection.countDocuments(textSearch).subscribe(subscriber);
    subscriber.await();

    // Find the highest scoring match
    System.out.print("Highest scoring document: ");
    Document projection = new Document("score", new Document("$meta", "textScore"));
    collection.find(textSearch).projection(projection).first().subscribe(new PrintDocumentSubscriber());


    // Run a command
    database.runCommand(new Document("buildInfo", 1)).subscribe(new PrintDocumentSubscriber());

    // release resources
    subscriber = new OperationSubscriber();
    database.drop().subscribe(subscriber);
    subscriber.await();
    mongoClient.close();
}
 

@Override
public Publisher<Success> createCollection(final String collectionName) {
    return createCollection(collectionName, new CreateCollectionOptions());
}
 

@Override
public Publisher<Success> createCollection(final ClientSession clientSession, final String collectionName) {
    return createCollection(clientSession, collectionName, new CreateCollectionOptions());
}
 

/**
 * Create a new collection with the selected options
 *
 * @param collectionName the name for the new collection to create
 * @param options various options for creating the collection
 * @return a {@link Uni} emitting {@code null} when the operation has completed
 */
Uni<Void> createCollection(String collectionName, CreateCollectionOptions options);
 

/**
 * Create a new collection with the selected options
 *
 * @param clientSession the client session with which to associate this operation
 * @param collectionName the name for the new collection to create
 * @param options various options for creating the collection
 * @return a {@link Uni} emitting {@code null} when the operation has completed
 */
Uni<Void> createCollection(ClientSession clientSession, String collectionName,
        CreateCollectionOptions options);
 

/**
 * Create a new collection with the selected options
 *
 * @param collectionName the name for the new collection to create
 * @param options        various options for creating the collection
 * @return an observable identifying when the collection has been created
 * @mongodb.driver.manual reference/commands/create Create Command
 */
Observable<Success> createCollection(String collectionName, CreateCollectionOptions options);
 

/**
 * Create a new collection with the selected options
 *
 * @param collectionName the name for the new collection to create
 * @param options        various options for creating the collection
 * @return a publisher identifying when the collection has been created
 * @mongodb.driver.manual reference/commands/create Create Command
 */
Publisher<Success> createCollection(String collectionName, CreateCollectionOptions options);
 

/**
 * Create a new collection with the selected options
 *
 * @param clientSession the client session with which to associate this operation
 * @param collectionName the name for the new collection to create
 * @param options        various options for creating the collection
 * @return a publisher identifying when the collection has been created
 * @mongodb.driver.manual reference/commands/create Create Command
 * @mongodb.server.release 3.6
 * @since 1.7
 */
Publisher<Success> createCollection(ClientSession clientSession, String collectionName, CreateCollectionOptions options);