Java 类com.mongodb.WriteResult 实例源码

/**
 * 更改恢复次数
 *
 * @param id              事务id
 * @param retry           恢复次数
 * @param applicationName 应用名称
 * @return true 成功
 */
@Override
public Boolean updateRetry(String id, Integer retry, String applicationName) {
    if (StringUtils.isBlank(id) || StringUtils.isBlank(applicationName) || Objects.isNull(retry)) {
        return Boolean.FALSE;
    }
    final String mongoTableName = RepositoryPathUtils.buildMongoTableName(applicationName);

    Query query = new Query();
    query.addCriteria(new Criteria("transId").is(id));
    Update update = new Update();
    update.set("lastTime", DateUtils.getCurrentDateTime());
    update.set("retriedCount", retry);
    final WriteResult writeResult = mongoTemplate.updateFirst(query, update,
            MongoAdapter.class, mongoTableName);
    if (writeResult.getN() <= 0) {
        throw new TransactionRuntimeException("更新数据异常!");
    }
    return Boolean.TRUE;
}

/**
 * 更新 List<Participant>  只更新这一个字段数据
 *
 * @param tccTransaction 实体对象
 */
@Override
public int updateParticipant(TccTransaction tccTransaction) {
    Query query = new Query();
    query.addCriteria(new Criteria("transId").is(tccTransaction.getTransId()));
    Update update = new Update();
    try {
        update.set("contents", objectSerializer.serialize(tccTransaction.getParticipants()));
    } catch (TccException e) {
        e.printStackTrace();
    }
    final WriteResult writeResult = template.updateFirst(query, update, MongoAdapter.class, collectionName);
    if (writeResult.getN() <= 0) {
        throw new TccRuntimeException("更新数据异常!");
    }
    return 1;
}

/**
 * 更新事务失败日志
 *
 * @param mythTransaction 实体对象
 * @return rows 1 成功
 * @throws MythRuntimeException 异常信息
 */
@Override
public int updateFailTransaction(MythTransaction mythTransaction) throws MythRuntimeException {
    Query query = new Query();
    query.addCriteria(new Criteria("transId").is(mythTransaction.getTransId()));
    Update update = new Update();

    update.set("status", mythTransaction.getStatus());
    update.set("errorMsg", mythTransaction.getErrorMsg());
    update.set("lastTime", new Date());
    update.set("retriedCount", mythTransaction.getRetriedCount());

    final WriteResult writeResult = template.updateFirst(query, update, MongoAdapter.class, collectionName);
    if (writeResult.getN() <= 0) {
        throw new MythRuntimeException("更新数据异常!");
    }
    return CommonConstant.SUCCESS;
}

/**
 * 更新 List<Participant>  只更新这一个字段数据
 *
 * @param mythTransaction 实体对象
 */
@Override
public int updateParticipant(MythTransaction mythTransaction) throws MythRuntimeException {
    Query query = new Query();
    query.addCriteria(new Criteria("transId").is(mythTransaction.getTransId()));
    Update update = new Update();
    try {
        update.set("contents", objectSerializer.serialize(mythTransaction.getMythParticipants()));
    } catch (MythException e) {
        e.printStackTrace();
    }
    final WriteResult writeResult = template.updateFirst(query, update, MongoAdapter.class, collectionName);
    if (writeResult.getN() <= 0) {
        throw new MythRuntimeException("更新数据异常!");
    }
    return CommonConstant.SUCCESS;
}

@Override
public boolean saveResourcePlan(ResourcePlan resourcePlan) {
    boolean result = false;
    if(resourcePlan.getAddTime() == 0) { //insert
        resourcePlan.setAddTime(new Date().getTime());
        resourcePlan.setModTime(new Date().getTime());

        mongoTemplate.save(resourcePlan, Constant.COL_NAME_RESOURCE_PLAN);
        result = Preconditions.isNotBlank(resourcePlan.getId());

    } else {                            //update
        Query query = new Query().addCriteria(Criteria.where("_id").is(resourcePlan.getId()));
        Update update = new Update();
        update.set("startPageNum", resourcePlan.getStartPageNum());
        update.set("endPageNum", resourcePlan.getEndPageNum());
        update.set("modTime", new Date().getTime());

        WriteResult writeResult = mongoTemplate.updateFirst(query, update, Constant.COL_NAME_RESOURCE_PLAN);
        result = writeResult!=null && writeResult.getN() > 0;
    }

    return result;
}

public SmartiUser createPasswordRecoveryToken(String login) {
    final SmartiUser mongoUser = findUser(login);
    if (mongoUser == null) {
        return null;
    }

    final Date now = new Date(),
            expiry = DateUtils.addHours(now, 24);
    final String token = HashUtils.sha256(UUID.randomUUID() + mongoUser.getLogin());
    final SmartiUser.PasswordRecovery recovery = new SmartiUser.PasswordRecovery(token, now, expiry);

    final WriteResult result = updateMongoUser(mongoUser.getLogin(), Update.update(SmartiUser.FIELD_RECOVERY, recovery));
    if (result.getN() == 1) {
        return getSmaritUser(mongoUser.getLogin());
    } else {
        return null;
    }
}

@Override
public Conversation updateMessage(ObjectId conversationId, Message message) {
    final Query query = new Query(Criteria.where("_id").is(conversationId))
            .addCriteria(Criteria.where("messages._id").is(message.getId()));

    final Update update = new Update()
            .set("messages.$", message)
            .currentDate("lastModified");

    final WriteResult writeResult = mongoTemplate.updateFirst(query, update, Conversation.class);
    if (writeResult.getN() == 1) {
        return mongoTemplate.findById(conversationId, Conversation.class);
    } else {
        return null;
    }
}

@Override
public Conversation saveIfNotLastModifiedAfter(Conversation conversation, Date lastModified) {

    final Query query = new Query();
    query.addCriteria(Criteria.where("_id").is(conversation.getId()));
    query.addCriteria(Criteria.where("lastModified").lte(lastModified));

    BasicDBObject data = new BasicDBObject();
    mongoTemplate.getConverter().write(conversation, data);
    final Update update = new Update();
    data.entrySet().stream()
        .filter(e -> !Objects.equals("lastModified", e.getKey()))
        .forEach(e -> update.set(e.getKey(), e.getValue()));
    update.currentDate("lastModified");

    final WriteResult writeResult = mongoTemplate.updateFirst(query, update, Conversation.class);
    if (writeResult.getN() == 1) {
        return mongoTemplate.findById(conversation.getId(), Conversation.class);
    } else {
        throw new ConcurrentModificationException(
                String.format("Conversation %s has been modified after %tF_%<tT.%<tS (%tF_%<tT.%<tS)", conversation.getId(), lastModified, conversation.getLastModified()));
    }
}

@Test
public void shouldInsertOneHeroWithAutomaticObjectId() {
    //GIVEN
    Address castleWinterfell = new Address("Winterfell", "Westeros", Region.THE_NORTH);

    Set<Human> children = Sets.newHashSet();
    children.add(Hero.createHeroWithoutChildrenAndNoBeasts("Robb", "Stark", castleWinterfell));
    children.add(Heroine.createHeroineWithoutChildrenAndNoBeasts("Sansa", "Stark", castleWinterfell));
    children.add(Heroine.createHeroineWithoutChildrenAndNoBeasts("Arya", "Stark", castleWinterfell));
    children.add(Hero.createHeroWithoutChildrenAndNoBeasts("Bran", "Stark", castleWinterfell));
    children.add(Hero.createHeroWithoutChildrenAndNoBeasts("Rickon", "Stark", castleWinterfell));
    children.add(Hero.createHeroWithoutChildrenAndNoBeasts("Jon", "Snow", castleWinterfell));

    Hero eddardStark = Hero.createHeroWithoutBeasts("Eddard", "Stark", castleWinterfell, children);

    //WHEN
    WriteResult insert = heroes.insert(eddardStark);

    //THEN
    Assertions.assertThat(insert.getError()).isNull();
}

@Test
public void shouldAddFieldToTheLightbringer() {
    //GIVEN
    WeaponDetails details = new WeaponDetails("The one who pulls out this sword from fire will be named Lord's Chosen ...", "Azor Ahai");

    //WHEN
    WriteResult lightbringer = weapons.update("{_id: #}", "Lightbringer").with("{$set: {details: #}}", details);

    //THEN
    assertThat(lightbringer.getError()).isNull();

    //AND WHEN
    Sword sword = weapons.findOne("{_id: 'Lightbringer'}").as(Sword.class);


    //THEN
    assertThat(sword).isNotNull();
}

/**
 * Add the specified mvc to the specified database
 *
 * @param dbSpecPath
 * @param mvc
 * @return
 */
static String addCall(String dbSpecPath, MongoVariantContext mvc) {

    NA12878DBArgumentCollection args = new NA12878DBArgumentCollection(dbSpecPath);

    String errorMessage = null;
    NA12878KnowledgeBase kb = null;
    try {
        kb = new NA12878KnowledgeBase(null, args);
        WriteResult wr = kb.addCall(mvc);
        errorMessage = wr.getError();
    } catch (Exception ex) {
        errorMessage = ex.getMessage();
        if (errorMessage == null) errorMessage = "" + ex;
    } finally {
        if (kb != null) kb.close();
    }

    return errorMessage;
}

/**
 * Perform a single 'put' operation on the local object store.
 *
 * @param ref  Object reference string of the object to be written.
 * @param obj  JSON string encoding the object to be written.
 * @param collection  Collection to put to.
 *
 * @return a ResultDesc object describing the success or failure of the
 *    operation.
 */
private ResultDesc doPut(String ref, String obj, DBCollection collection,
                         boolean requireNew)
{
    String failure = null;
    if (obj == null) {
        failure = "no object data given";
    } else {
        try {
            DBObject objectToWrite = jsonLiteralToDBObject(obj, ref);
            if (requireNew) {
                WriteResult wr = collection.insert(objectToWrite);
            } else {
                DBObject query = new BasicDBObject();
                query.put("ref", ref);
                collection.update(query, objectToWrite, true, false);
            }
        } catch (Exception e) {
            failure = e.getMessage();
        }
    }
    return new ResultDesc(ref, failure);
}

/**
 * Perform a single 'update' operation on the local object store.
 *
 * @param ref  Object reference string of the object to be written.
 * @param version  Expected version number of object before updating.
 * @param obj  JSON string encoding the object to be written.
 * @param collection  Collection to put to.
 *
 * @return an UpdateResultDesc object describing the success or failure of
 *    the operation.
 */
private UpdateResultDesc doUpdate(String ref, int version, String obj,
                                  DBCollection collection)
{
    String failure = null;
    boolean atomicFailure = false;
    if (obj == null) {
        failure = "no object data given";
    } else {
        try {
            DBObject objectToWrite = jsonLiteralToDBObject(obj, ref);
            DBObject query = new BasicDBObject();
            query.put("ref", ref);
            query.put("version", version);
            WriteResult result =
                collection.update(query, objectToWrite, false, false);
            if (result.getN() != 1) {
                failure = "stale version number on update";
                atomicFailure = true;
            }
        } catch (Exception e) {
            failure = e.getMessage();
        }
    }
    return new UpdateResultDesc(ref, failure, atomicFailure);
}

@Override
public boolean updateOrCreate(final Collection<MongoApproval> mongoApprovals) {
    boolean result = true;
    for (MongoApproval mongoApproval : mongoApprovals) {
        final Update update = Update
                .update("expiresAt", mongoApproval.getExpiresAt())
                .addToSet("status", mongoApproval.getStatus())
                .addToSet("lastModifiedAt",
                        mongoApproval.getLastUpdatedAt());

        final WriteResult writeResult = mongoTemplate.upsert(
                byUserIdAndClientIdAndScope(mongoApproval), update,
                MongoApproval.class);

        if (writeResult.getN() != 1) {
            result = false;
        }
    }
    return result;
}

/**
 * This method extracts database collection name and inserts passet object in 
 * database.
 * @param object
 *        Object to insert in database
 * @return inserted object
 * @throws ObjectNotStoredException
 */
private FfmaDomainObject storeToMongoDb(FfmaDomainObject object)
    throws ObjectNotStoredException {

    // TODO: check if exists? last time ? etc
    DBCollection mongoCollection = db.getCollectionFromString(object
            .getClass().getSimpleName());
    WriteResult res = mongoCollection.insert((BasicDBObject) object);
    log.debug("storeToMongoDb() coll:" + mongoCollection + ", res: " + res.toString());
    try {
        return retrieveObject(object);
    } catch (Exception e) {
        throw new ObjectNotStoredException(
                "Cannot store and retreive object from db after creation!",
                e);
    }
}

/**
 * 
 * mongodb，解析 更新操作是否成功
 * 
 * 返回更新数据库的结果
 * 
 * 小于零：更新出现异常 等于零：成功执行了更新的SQL，但是没有影响到任何数据 大于零：成功执行了更新的SQL，影响到多条数据，条数就是返回值
 * 
 * @param result
 * @return
 */
@Override
public int getUpdateResult(WriteResult result) {
    if (result == null) {
        return FAIL_CODE_ONE;
    }
    @SuppressWarnings("deprecation")
    CommandResult cr = result.getLastError();
    if (cr == null) {
        return FAIL_CODE_TWO;
    }
    boolean error_flag = false;
    error_flag = cr.ok();
    if (!error_flag) {// 获取上次操作结果是否有错误.
        return FAIL_CODE_THREE;
    }
    int affect_count = result.getN();// 操作影响的对象个数

    if (affect_count < 0) {
        return FAIL_CODE_FOUR;
    } else {
        return affect_count;
    }
}

/**
 * Removes the selected algorithm's metadata.
 *
 * @param id the Id of the selected algorithm's metadata.
 */
public void deleteAlgorithm(String id) {
    log.debug(MSG_DAO_DELETE + id);

    if (!ObjectId.isValid(id)) {
        log.error(MSG_ERR_NOT_VALID_ID);
        throw new BadRequestException(MSG_ERR_NOT_VALID_ID);
    }

    algorithmsCollection = INSTANCE.getDatasource().getDbCollection(ALGORITHMS_COLLECTION_NAME);
    BasicDBObject query = new BasicDBObject();
    query.put("_id", new ObjectId(id));
    WriteResult wRes = algorithmsCollection.remove(query);

    if (wRes.getN() == 0) {
        log.error(MSG_ERR_NOT_FOUND);
        throw new ResourceNotFoundException();
    }
}

/**
 * Remove the selected user.
 *
 * @param id the Id of the selected user.
 */
public void deleteUser(String id) {
    log.debug(MSG_DAO_DELETE + id + ".");

    if (!ObjectId.isValid(id)) {
        log.error(MSG_ERR_NOT_VALID_ID);
        throw new BadRequestException(MSG_ERR_NOT_VALID_ID);
    }

    usersCollection = INSTANCE.getDatasource().getDbCollection(USERS_COLLECTION_NAME);
    BasicDBObject query = new BasicDBObject();
    query.put("_id", new ObjectId(id));
    WriteResult wRes = usersCollection.remove(query);

    if (wRes.getN() == 0) {
        log.error(MSG_ERR_NOT_FOUND);
        throw new ResourceNotFoundException();
    }
}

/**
 * Removes the selected dataset's metadata.
 *
 * @param id the Id of the selected dataset's metadata.
 */
public void deleteDataset(String id) {
    log.debug(MSG_DAO_DELETE);

    if (!ObjectId.isValid(id)) {
        log.error(MSG_ERR_NOT_VALID_ID);
        throw new BadRequestException(MSG_ERR_NOT_VALID_ID);
    }

    datasetsCollection = INSTANCE.getDatasource().getDbCollection(DATASETS_COLLECTION_NAME);
    BasicDBObject query = new BasicDBObject();
    query.put("_id", new ObjectId(id));
    WriteResult wRes = datasetsCollection.remove(query);

    if (wRes.getN() == 0) {
        log.error(MSG_ERR_NOT_FOUND);
        throw new ResourceNotFoundException();
    }
}

/**
 * Remove the selected department.
 *
 * @param id the Id of the selected department.
 */
public void deleteDepartment(String id) {
    log.debug(MSG_DAO_DELETE + id + ".");

    if (!ObjectId.isValid(id)) {
        log.error(MSG_ERR_NOT_VALID_ID);
        throw new BadRequestException(MSG_ERR_NOT_VALID_ID);
    }

    departmentsCollection = INSTANCE.getDatasource().
            getDbCollection(DEPARTMENTS_COLLECTION_NAME);
    BasicDBObject query = new BasicDBObject();
    query.put("_id", new ObjectId(id));
    WriteResult wRes = departmentsCollection.remove(query);

    if (wRes.getN() == 0) {
        log.error(MSG_ERR_NOT_FOUND);
        throw new ResourceNotFoundException();
    }
}

/**
 * Remove the selected company.
 *
 * @param id the Id of the selected company.
 */
public void deleteCompany(String id) {
    log.debug(MSG_DAO_DELETE + id + ".");

    if (!ObjectId.isValid(id)) {
        log.error(MSG_ERR_NOT_VALID_ID);
        throw new BadRequestException(MSG_ERR_NOT_VALID_ID);
    }

    companiesCollection = INSTANCE.getDatasource().getDbCollection(COMPANIES_COLLECTION_NAME);
    BasicDBObject query = new BasicDBObject();
    query.put("_id", new ObjectId(id));
    WriteResult wRes = companiesCollection.remove(query);

    if (wRes.getN() == 0) {
        log.error(MSG_ERR_NOT_FOUND);
        throw new ResourceNotFoundException();
    }
}

/**
 * Removes the selected process' metadata object.
 *
 * @param id the Id of the selected process' metadata object.
 */
public void deleteProcess(String id) {
    log.debug(MSG_DAO_DELETE + id);

    // Check passed Id
    if (!ObjectId.isValid(id)) {
        log.error(MSG_ERR_NOT_VALID_ID);
        throw new BadRequestException(MSG_ERR_NOT_VALID_ID);
    }

    processesCollection = INSTANCE.getDatasource().getDbCollection(PROCESSES_COLLECTION_NAME);
    BasicDBObject query = new BasicDBObject();
    query.put("_id", new ObjectId(id));
    WriteResult wRes = processesCollection.remove(query);

    // Check the number of deleted objects
    if (wRes.getN() == 0) { // if 0 then the query found nothing
        log.error(MSG_ERR_NOT_FOUND);
        throw new ResourceNotFoundException(MSG_ERR_NOT_FOUND);
    }
}

/**
 * Removes the selected data-source's metadata object.
 *
 * @param id the Id of the selected data-source's metadata object.
 */
public void deleteDataSource(String id) {
    log.debug(MSG_DAO_DELETE + id);

    if (!ObjectId.isValid(id)) {
        log.error(MSG_ERR_NOT_VALID_ID);
        throw new BadRequestException(MSG_ERR_NOT_VALID_ID);
    }

    datasourcesCollection
            = INSTANCE.getDatasource().getDbCollection(DATASOURCES_COLLECTION_NAME);
    BasicDBObject query = new BasicDBObject("_id", new ObjectId(id));
    WriteResult wRes = datasourcesCollection.remove(query);

    if (wRes.getN() == 0) {
        log.error(MSG_ERR_NOT_FOUND);
        throw new ResourceNotFoundException();
    }
}

/**
 * UPLOAD TO DB
 */
public static void insertJetStreamConfiguration(BasicDBObject dbObject,
        MongoLogConnection mongoLogConnection) {
    JetStreamBeanConfigurationLogDo beanConfig = null;
    DBCollection dbCol = mongoLogConnection.getDBCollection();

    if (dbCol == null) {
        throw new MongoConfigRuntimeException(
                "jetstreamconfig collection is unknown");
    }

    WriteResult result = dbCol.insert(dbObject);
    if (result.getError() != null) {
        throw new MongoConfigRuntimeException(result.getError());
    }
}

public static boolean removeConfigurationByQuery(BasicDBObject query, MongoConnection mongoConnection) {

    DBCollection dbCol = mongoConnection.getDBCollection();

    if (dbCol == null) {
        throw new MongoConfigRuntimeException("jetstreamconfig collection is unknown");
    }

    try {
        if(query ==null) {
            return false;
        }

        WriteResult result = dbCol.remove(query, WriteConcern.SAFE);

        if(result.getLastError().ok()) {
            return true;
        }

    } catch (Exception err) {
        throw new MongoConfigRuntimeException(err);
    } 

    return true;
}

@Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
    Class<?>[] argumentsTypes, Object ret) throws Throwable {
    AbstractSpan activeSpan = ContextManager.activeSpan();
    CommandResult cresult = null;
    if (ret instanceof WriteResult) {
        WriteResult wresult = (WriteResult)ret;
        cresult = wresult.getCachedLastError();
    } else if (ret instanceof AggregationOutput) {
        AggregationOutput aresult = (AggregationOutput)ret;
        cresult = aresult.getCommandResult();
    }
    if (null != cresult && !cresult.ok()) {
        activeSpan.log(cresult.getException());
    }
    ContextManager.stopSpan();
    return ret;
}

/**
 * Update amenityIds for service(s). The List<JSONObject>  should in the below format
 * [{
 *     "serviceId":"1234", "amenityIds":["2323","33423","33523"]
 * },{
 *     "serviceId":"1434", "amenityIds":["233433","3333423"]
 * }]
 *
 * @return
 */
public boolean updateServiceAmenities(List<JSONObject> services) {
    /*
    BulkOperations ops = mongoTemplate.bulkOps(BulkOperations.BulkMode.UNORDERED, BusService.class);
    for (JSONObject service: services) {
        Query query = new Query(where(AbstractDocument.KEY_ID).is(service.get("serviceId").toString()));
        Update updateOp = new Update();
        updateOp.set("amenityIds", service.get("amenityIds"));
        ops.updateOne(query, updateOp);
    }
    BulkWriteResult result = ops.execute();
    return result.getModifiedCount() == services.size(); */

    for (JSONObject jsonObject : services) {
        Update updateOp = new Update();
        updateOp.set("amenityIds", jsonObject.get("amenityIds"));
        final Query query = new Query();
        query.addCriteria(where("_id").is(jsonObject.get("serviceId")));
        WriteResult writeResult =  mongoTemplate.updateMulti(query, updateOp, BusService.class);
        if(writeResult.getN() != 1) {
            return false;
        }
    }
    return true;
}

public boolean updateCashBalance(String userId, double cashBalance) {
    Update updateOp = new Update();
    updateOp.inc("amountToBePaid", cashBalance);
    final Query query = new Query();
    query.addCriteria(where("_id").is(userId));
    WriteResult writeResult =  mongoTemplate.updateMulti(query, updateOp, User.class);
    return writeResult.getN() == 1;
}

private ShipmentSequence nextSequeceNumber(ShipmentType shipmentType) {
    ShipmentSequence shipmentSequence = null;
    if(shipmentSequenceDAO.findByShipmentCode(shipmentType.getKey()) == null){
        shipmentSequence = shipmentSequenceDAO.save(new ShipmentSequence(shipmentType));
    } else {
        Update updateOp = new Update();
        updateOp.inc("nextNumber", 1);
        final Query query = new Query();
        query.addCriteria(where("shipmentCode").is(shipmentType.getKey()));
        WriteResult writeResult =  mongoTemplate.updateMulti(query, updateOp, ShipmentSequence.class);
        if(writeResult.getN() == 1){
            shipmentSequence = shipmentSequenceDAO.findByShipmentCode(shipmentType.getKey());
        } else {
            throw new IllegalStateException("next number failed");
        }
    }
    return shipmentSequence;
}

@Override
public boolean deleteChunks(List<String> chunkIds, long maxLastModifiedTime) throws Exception {
    DBCollection collection = getBlobCollection();
    QueryBuilder queryBuilder = new QueryBuilder();
    if (chunkIds != null) {
        queryBuilder = queryBuilder.and(MongoBlob.KEY_ID).in(chunkIds.toArray(new String[0]));
        if (maxLastModifiedTime > 0) {
            queryBuilder = queryBuilder.and(MongoBlob.KEY_LAST_MOD)
                                .lessThan(maxLastModifiedTime);
        }
    }

    WriteResult result = collection.remove(queryBuilder.get());
    if (result.getN() == chunkIds.size()) {
        return true;
    }

    return false;
}

public final static boolean upsert(DBCollection coll, DBObject q, DBObject dbo, boolean ensureId) {

        WriteResult wr = coll.update(q, dbo, true, false, WriteConcern.ACKNOWLEDGED);
        boolean updatedExisting = wr.isUpdateOfExisting();

        if (ensureId) {
            if (updatedExisting) {

                BasicDBObject f = new BasicDBObject();
                f.put(MongoUtils._id, 1);

                DBObject o = coll.findOne(q, f);
                dbo.put(MongoUtils._id, o.get(MongoUtils._id));

            } else {
                ObjectId upserted = (ObjectId) wr.getUpsertedId();
                dbo.put(MongoUtils._id, upserted);
            }
        }

        return updatedExisting;     
    }

public void ping(String callerId, String resourceId) {
    Date now = new Date();
    BasicDBObject q = new BasicDBObject().
            append(CALLERID, callerId).
            append(RESOURCEID, resourceId).
            append(EXPIRATION, new BasicDBObject("$gt", now)).
            append(COUNT, new BasicDBObject("$gt", 0));
    DBObject lock = coll.findOne(q,null,ReadPreference.primary());
    if (lock != null) {
        Date expiration = new Date(now.getTime() + ((Number) lock.get(TTL)).longValue());
        int ver = ((Number) lock.get(VERSION)).intValue();
        BasicDBObject update = new BasicDBObject().
                append("$set", new BasicDBObject(TIMESTAMP, now).
                        append(EXPIRATION, expiration)).
                append("$inc", new BasicDBObject(VERSION, 1));
        q = q.append(VERSION, ver);
        WriteResult wr = coll.update(q, update, false, false, WriteConcern.ACKNOWLEDGED);
        if (wr.getN() != 1) {
            throw new InvalidLockException(resourceId);
        }
        LOGGER.debug("{}/{} pinged", callerId, resourceId);
    } else {
        throw new InvalidLockException(resourceId);
    }
}

private WriteResult execute(DBObject condition){
    List ids = null;
    if(dao.hasCustomListener){
        ids = dao.getCollection().distinct(Operator.ID, condition);
    }
    if(isolated){
        condition.put(Operator.ISOLATED, 1);
    }
    WriteResult wr = dao.getCollection().update(condition, modifier, false, true);  //update multi
    if(dao.hasCustomListener && ids != null){
        DBObject in = new BasicDBObject(Operator.IN, ids);
        DBCursor cursor = dao.getCollection().find(new BasicDBObject(Operator.ID, in));
        List<T> list = MapperUtil.toList(dao.getEntityClass(), cursor);
        for(T t : list){
            dao.notifyUpdated((BuguEntity)t);
        }
    }
    return wr;
}

@Test
public void testPersist() {
    DBCollection coll = createMockCollection();
    ToDo td = new ToDo();
    td.setTitle("This is a test");
    td.setId("aaaaaaaaaaaaaaaaaaaaaaa1");
    expect(coll.insert(isA(DBObject.class))).andAnswer(new IAnswer<WriteResult>() {
        @Override
        public WriteResult answer() throws Throwable {
            DBObject obj = (DBObject)getCurrentArguments()[0];
            obj.put("_id", new ObjectId("aaaaaaaaaaaaaaaaaaaaaaa1"));
            return null;
        }
    });
    replay(coll);
    MongoStore store = new MongoStore(coll);
    assertEquals(td, store.persist(td));
    verify(coll);
}

@Override
public boolean updateOrCreate(final Collection<MongoApproval> mongoApprovals) {
    boolean result = true;
    for (MongoApproval mongoApproval : mongoApprovals) {
        final Update update = Update.update("expiresAt", mongoApproval.getExpiresAt())
                .set("status", mongoApproval.getStatus())
                .set("lastUpdatedAt", mongoApproval.getLastUpdatedAt());

        final WriteResult writeResult = mongoTemplate.upsert(byUserIdAndClientIdAndScope(mongoApproval), update, MongoApproval.class);

        if (writeResult.getN() != 1) {
            result = false;
        }
    }
    return result;
}

@Override
public boolean insertUserFans(String girlId, String boyId) {
    // TODO Auto-generated method stub

    DB db = MongoDBUtil.getDB();
    BasicDBObject girl = new BasicDBObject();
    girl.put("_id", girlId);

    DBRef boyDbRef = new DBRef(db, "user", boyId);

    BasicDBObject update = new BasicDBObject();
    update.put("user", boyDbRef);
    update.put("isPass", 0);

    DBCollection users = db.getCollection(userCollection);

    WriteResult res = users.update(girl, new BasicDBObject("$addToSet",
            new BasicDBObject("userFans", update)), false, true);

    return res.getN() > 0 ? true : false;
}

@Override
public boolean insertUserBoyMatch(String boyId, String girlId) {
    // TODO Auto-generated method stub

    DB db = MongoDBUtil.getDB();
    BasicDBObject boy = new BasicDBObject();
    boy.put("_id", boyId);
    DBRef girlDbRef = new DBRef(db, "user", girlId);

    BasicDBObject update = new BasicDBObject();
    update.put("user", girlDbRef);
    update.put("isPass", 0);

    DBCollection users = db.getCollection(userCollection);

    WriteResult res = users.update(boy, new BasicDBObject("$addToSet",
            new BasicDBObject("userMatch", update)), false, true);

    return res.getN() > 0 ? true : false;
}

private void updateActivityRef(String activityId, long time, boolean isHidden) {
  DBCollection activityColl = CollectionName.ACTIVITY_COLLECTION.getCollection(this.abstractMongoStorage);
  //
   BasicDBObject update = new BasicDBObject();
   Map<String, Object> fields = new HashMap<String, Object>();
   fields.put(ActivityMongoEntity.lastUpdated.getName(), time);
   fields.put(ActivityMongoEntity.hidable.getName(), isHidden);
   BasicDBObject set = new BasicDBObject(fields);
   //
   update.append("$set", set);
   BasicDBObject query = new BasicDBObject(ActivityMongoEntity.id.getName(), new ObjectId(activityId));

   WriteResult result = activityColl.update(query, update);
   LOG.debug("UPDATED TIME ACTIVITY: " + result.toString());
   //update refs
   DBCollection streamCol = CollectionName.STREAM_ITEM_COLLECTION.getCollection(this.abstractMongoStorage);
   query = new BasicDBObject(StreamItemMongoEntity.activityId.getName(), activityId);
   fields = new HashMap<String, Object>();
   fields.put(StreamItemMongoEntity.time.getName(), time);
   fields.put(StreamItemMongoEntity.hiable.getName(), isHidden);
   set = new BasicDBObject(fields);
   update = new BasicDBObject("$set", set);
   result = streamCol.updateMulti(query, update);
   LOG.debug("UPDATED ACTIVITY Reference: " + result.toString());
}

/**
 * Reset list of pending content items to process
 * 
 * @param queueId ID of processing queue
 * @param inProcessOnly only reset items marked as "in_process"
 * @return number of reset items
 * @throws EumssiException with a specific StatusType, if one of the following scenarios occurs:
 *  <br>
 *  <br><code>StatusType.ERROR_INVALID_QUEUE_ID</code> (Error 102) if the specified queue id does not correspond to a valid queue.
 *  <br>
 *  <br><code>StatusType.ERROR_DB_CONNECT</code> (Error 400) if an unhandled error occurred during acquisition of the database connection.
 *  <br><code>StatusType.ERROR_DB_QUERY</code> (Error 401) if an unhandled error occurred during the query execution.
 *  <br><code>StatusType.ERROR_UNKNOWN</code> (Error 999) if an unhandled exception is thrown.
 */
public Integer resetQueue(String queueId, Boolean inProcessOnly, String filters) throws EumssiException {
    DBObject query = null;
    if (this.queues.containsKey(queueId)) {
        query = (DBObject) JSON.parse(this.queues.getProperty(queueId));
        // check that item is marked as in_process
        String testReset = String.format("{\"processing.queues.%s\":\"in_process\"}", queueId);
        if (!inProcessOnly) { // reset all items, even if already processed
            testReset = String.format("{\"processing.queues.%s\":{\"$in\":[\"in_process\",\"processed\"]}}", queueId);
        }
        query.putAll((BSONObject) JSON.parse(testReset));
        query.putAll((BSONObject) JSON.parse(filters)); // apply user-provided filters
    } else {
        throw new EumssiException(StatusType.ERROR_INVALID_QUEUE_ID);
    }
    try {
        log.info("performing query "+query.toString()+" on collection "+this.collection.toString());
        WriteResult r = collection.update(query, new BasicDBObject("$set", new BasicDBObject("processing.results."+queueId,"pending")));
        Integer updatedCount = r.getN();
        return updatedCount;}
    catch (Exception e) {
        // TODO: better handle possible failures
        throw new EumssiException(StatusType.ERROR_UNKNOWN);
    }
}

/**
 * TODO: document method 
 * @param queueId
 * @param data
 * @return
 */
public Integer putResults(String queueId, String data) throws EumssiException {
    Integer updatedCount = 0;
    BasicDBList jsonData = (BasicDBList) JSON.parse(data);
    for (Object item : jsonData) {
        try {
            String itemId = (String) ((DBObject)item).get("content_id");
            WriteResult r = collection.update(new BasicDBObject("_id", UUID.fromString(itemId)), new BasicDBObject("$set", new BasicDBObject("processing.results."+queueId,((DBObject) item).get("result"))));
            collection.update(
                    new BasicDBObject("_id", UUID.fromString(itemId)), 
                    new BasicDBObject()
                        .append("$set", new BasicDBObject("processing.queues."+queueId,"processed"))
                        .append("$pull", new BasicDBObject("processing.available_data",queueId))
                    );
            updatedCount += r.getN();
        } catch (Exception e) { //TODO: better exception handling
            log.error(String.format("couldn't insert data in document %s", item), e);               
        }
    }
    return updatedCount;
}