@Test
void testInsertionFailedWhenDocumentExist() {
    String collection = randomCollection();
    ReactiveMongoDatabase database = client.getDatabase(DATABASE);
    ReactiveMongoCollection<Document> myCollection = database.getCollection(collection);
    Document doc = createDoc();
    ObjectId value = new ObjectId();
    doc.put("_id", value);
    myCollection.insertOne(doc).await().indefinitely();
    try {
        myCollection.insertOne(doc).await().indefinitely();
        fail("Write Exception expected");
    } catch (Exception e) {
        assertThat(e).isInstanceOf(MongoWriteException.class);
    }
}
 

/**
 * Saves an entry to file
 * @param entry
 * @param dbName usually scrapig
 * @return true if success
 */
public static boolean saveEntry(DBEntry entry, String dbName){

    if(entry == null || !entry.isValid())
        return false;

    Logger log = Logger.getLogger(DAO.class);

    MongoDatabase db = MongoDB.INSTANCE.getDatabase(dbName);

    String collectionName = getCollectionName(entry);


    MongoCollection collection = db.getCollection(collectionName,BasicDBObject.class);

    try {
        collection.insertOne(entry);
        return true;
    }
    catch (MongoWriteException ex){
        if (ex.getCode() != 11000) // Ignore errors about duplicates
            log.error(ex.getError().getMessage());
        return false;
    }

}
 

public boolean acquireLock(MongoDatabase db) {

    Document insertObj = new Document(KEY_PROP_NAME, LOCK_ENTRY_KEY_VAL).append("status", "LOCK_HELD");

    // acquire lock by attempting to insert the same value in the collection - if it already exists (i.e. lock held)
    // there will be an exception
    try {
      db.getCollection(lockCollectionName).insertOne(insertObj);
    } catch (MongoWriteException ex) {
      if (ex.getError().getCategory() == ErrorCategory.DUPLICATE_KEY) {
        logger.warn("Duplicate key exception while acquireLock. Probably the lock has been already acquired.");
      }
      return false;
    }
    return true;
  }
 

/**
 * Un-assigns a key/value property from the element. The object value of the
 * removed property is returned.
 *
 * @param key the key of the property to remove from the element
 * @return the object value associated with that key prior to removal. Should be
 *         instance of BsonValue
 */
@Override
public <T> T removeProperty(final String key) {
	try {
		BsonValue value = getProperty(key);
		BsonDocument filter = new BsonDocument();
		filter.put(Tokens.ID, new BsonString(this.id));
		BsonDocument update = new BsonDocument();
		update.put("$unset", new BsonDocument(key, new BsonNull()));
		if (this instanceof ChronoVertex) {
			graph.getVertexCollection().updateOne(filter, update, new UpdateOptions().upsert(true));
			return (T) value;
		} else {
			graph.getEdgeCollection().updateOne(filter, update, new UpdateOptions().upsert(true));
			return (T) value;
		}
	} catch (MongoWriteException e) {
		throw e;
	}
}
 

@Test
public void createValidation() {
    getMapper().map(DocumentValidation.class);
    getDs().enableDocumentValidation();
    assertEquals(Document.parse(DocumentValidation.class.getAnnotation(Validation.class).value()), getValidator());

    try {
        getDs().save(new DocumentValidation("John", 1, new Date()));
        fail("Document should have failed validation");
    } catch (MongoWriteException e) {
        assertTrue(e.getMessage().contains("Document failed validation"));
    }

    getDs().save(new DocumentValidation("Harold", 100, new Date()));

}
 

@Test
public void testBypassDocumentValidation() {
    getMapper().map(User.class);
    getDs().enableDocumentValidation();

    final User user = new User("Jim Halpert", new Date());
    user.age = 5;

    try {
        getDs().save(user);
        fail("Document validation should have rejected the document");
    } catch (MongoWriteException ignored) {
    }

    getDs().save(user, new InsertOneOptions().bypassDocumentValidation(true));

    Assert.assertEquals(1, getDs().find(User.class).count());
}
 

@Test(expected = MongoWriteException.class)
public void noDuplicatesAllowed() {
    String data = "this is a blog post !";
    String url = "http://www.columbia.edu";
    BlogPost post = new BlogPost(data,"tst",null,"source",url,"GUID231253423","");
    posts.insertOne(post);
    String data2 = "this is a blog post !";
    String url2 = "http://www.columbia.edu";
    BlogPost post2 = new BlogPost(data,"tst",null,"source",url,"GUID231253423","");
    posts.insertOne(post2);
}
 

@Test(expected = MongoWriteException.class)
public void noDuplicatesAllowed() {
        String data = "this is a tweet !";
        String url = "http://www.columbia.edu";
        Tweet tweet = new Tweet(data,data,"tst","test",null,"topsy",url,"123456","filename");
        tweets.insertOne(tweet);
        String data2 = "this is a tweet !";
        String url2 = "http://www.columbia.edu";
        Tweet tweet2 = new Tweet(data2,data,"tst","test",null,"topsy",url2,"123456","filename");
        tweets.insertOne(tweet2);
    }
 

@Test(expected = MongoWriteException.class)
public void testUniqueIndexedEntity() {
    getMapper().map(List.of(UniqueIndexOnValue.class));
    getDs().ensureIndexes();
    assertThat(getIndexInfo(UniqueIndexOnValue.class), hasIndexNamed("l_ascending"));
    getDs().save(new UniqueIndexOnValue("a"));

    // this should throw...
    getDs().save(new UniqueIndexOnValue("v"));
}
 

@Override
public CompletableFuture<List<DomainEventStream>> queryAggregateEventsAsync(String aggregateRootId, String aggregateRootTypeName, int minVersion, int maxVersion) {
    return IOHelper.tryIOFuncAsync(() -> {
        CompletableFuture<List<DomainEventStream>> future = new CompletableFuture<>();

        Bson filter = Filters.and(Filters.eq("aggregateRootId", aggregateRootId),
                Filters.gte("version", minVersion),
                Filters.lte("version", maxVersion));
        Bson sort = Sorts.ascending("version");
        mongoClient.getDatabase(mongoConfiguration.getDatabaseName()).getCollection(mongoConfiguration.getEventCollectionName())
                .find(filter).sort(sort).subscribe(new Subscriber<Document>() {
            final List<DomainEventStream> streams = Lists.newArrayList();

            @Override
            public void onSubscribe(Subscription s) {
                s.request(Long.MAX_VALUE);
            }

            @Override
            public void onNext(Document document) {
                DomainEventStream eventStream = new DomainEventStream(
                        document.getString("commandId"),
                        document.getString("aggregateRootId"),
                        document.getString("aggregateRootTypeName"),
                        document.get("gmtCreate", Date.class),
                        eventSerializer.deserialize(JsonTool.deserialize(document.getString("events"), Map.class), IDomainEvent.class),
                        Maps.newHashMap());
                streams.add(eventStream);
            }

            @Override
            public void onError(Throwable t) {
                future.completeExceptionally(t);
            }

            @Override
            public void onComplete() {
                streams.sort(Comparator.comparingInt(DomainEventStream::getVersion));
                future.complete(streams);
            }
        });
        return future.exceptionally(throwable -> {
            if (throwable instanceof MongoWriteException) {
                MongoWriteException ex = (MongoWriteException) throwable;
                String errorMessage = String.format("Failed to query aggregate events async, aggregateRootId: %s, aggregateRootType: %s", aggregateRootId, aggregateRootTypeName);
                logger.error(errorMessage, ex);
                throw new IORuntimeException(throwable);
            }
            logger.error("Failed to query aggregate events async, aggregateRootId: {}, aggregateRootType: {}", aggregateRootId, aggregateRootTypeName, throwable);
            throw new EnodeRuntimeException(throwable);
        });
    }, "QueryAggregateEventsAsync");
}
 

private <T> boolean tryVersionedUpdate(final T entity, final MongoCollection collection, final InsertOneOptions options) {
    final MappedClass mc = mapper.getMappedClass(entity.getClass());
    if (mc.getVersionField() == null) {
        return false;
    }

    MappedField idField = mc.getIdField();
    final Object idValue = idField.getFieldValue(entity);
    final MappedField versionField = mc.getVersionField();

    Long oldVersion = (Long) versionField.getFieldValue(entity);
    long newVersion = oldVersion == null ? 1L : oldVersion + 1;
    ClientSession session = findSession(options);

    if (newVersion == 1) {
        try {
            updateVersion(entity, versionField, newVersion);
            if (session == null) {
                options.prepare(collection).insertOne(entity, options.getOptions());
            } else {
                options.prepare(collection).insertOne(session, entity, options.getOptions());
            }
        } catch (MongoWriteException e) {
            updateVersion(entity, versionField, oldVersion);
            throw new ConcurrentModificationException(Sofia.concurrentModification(entity.getClass().getName(), idValue));
        }
    } else if (idValue != null) {
        final UpdateResult res = find(collection.getNamespace().getCollectionName())
                                     .filter(eq("_id", idValue),
                                         eq(versionField.getMappedFieldName(), oldVersion))
                                     .update(UpdateOperators.set(entity))
                                     .execute(new UpdateOptions()
                                                  .bypassDocumentValidation(options.getBypassDocumentValidation())
                                                  .clientSession(session)
                                                  .writeConcern(options.writeConcern()));

        if (res.getModifiedCount() != 1) {
            throw new ConcurrentModificationException(Sofia.concurrentModification(entity.getClass().getName(), idValue));
        }
        updateVersion(entity, versionField, newVersion);
    }

    return true;
}