/**
 * 更新 TM中的消息状态
 *
 * @param transactionMsg
 */
@Override
public Boolean updateTxTransactionMsgStatus(TransactionMsg transactionMsg) {
    try {
        Query query = new Query();
        query.addCriteria(new Criteria("groupId").is(transactionMsg.getGroupId()).and("subTaskId").is(transactionMsg.getSubTaskId()));
        Update update = Update.update("consumed", transactionMsg.getConsumed());
        String message = transactionMsg.getMessage();
        if (StringUtils.isNotBlank(message)) {
            update.set("message", message);
        }
        update.set("updateTime", Timestamp.valueOf(DateUtils.getCurrentDateTime()).getTime());
        final WriteResult writeResult = mongoTemplate.updateFirst(query, update, TransactionMsg.class, CollectionNameEnum.TransactionMsg.name());
        return writeResult.getN() > 0;
    } catch (Exception e) {
        //TODO 处理异常
        LogUtil.error(LOGGER, e::getLocalizedMessage);
        return false;
    }
}
 

@Override
public Boolean updateRetry(final String id, final Integer retry, final 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;
}
 

@Override
public Boolean updateRetry(final String id, final Integer retry, final String appName) {
    if (StringUtils.isBlank(id) || StringUtils.isBlank(appName) || Objects.isNull(retry)) {
        return Boolean.FALSE;
    }
    final String mongoTableName = RepositoryPathUtils.buildMongoTableName(appName);
    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 RuntimeException("更新数据异常!");
    }
    return Boolean.TRUE;
}
 

@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;
}
 

@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;
}
 

/**
 * Save an entity to mongoDB. 
 * If no id in it, then insert the entity.
 * Else, check the id type, to confirm do save or insert.
 * @param t 
 * @return 
 */
public WriteResult save(T t){
    WriteResult wr;
    BuguEntity ent = (BuguEntity)t;
    if(StringUtil.isEmpty(ent.getId())){
        wr = insert(t);
    }
    else{
        Field idField = FieldsCache.getInstance().getIdField(clazz);
        Id idAnnotation = idField.getAnnotation(Id.class);
        if(idAnnotation.type()==IdType.USER_DEFINE){
            if(this.exists(Operator.ID, ent.getId())){
                wr = doSave(ent);
            }else{
                wr = insert(t);
            }
        }
        else{
            wr = doSave(ent);
        }
    }
    return wr;
}
 

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, upsert, 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);
}
 

/**
 * set the state of balancer.
 *
 * @param dbConn
 * @param state
 * @return Error description, or null if no errors
 */
private static String setBalancerState(DB dbConn, boolean state) {
    DBObject balancer = new BasicDBObject(ID, "balancer");
    DBObject updateObj = new BasicDBObject();
    String stopped = state ? "false" : "true";
    updateObj.put("$set", new BasicDBObject("stopped", stopped));
    WriteResult wresult = dbConn.getSisterDB("config").getCollection("settings").update(balancer, updateObj, true, false);
    if (wresult != null) {
        CommandResult result = wresult.getLastError();
        if (!result.ok()) {
            LOG.error("Error setting balancer state to {}: {}", state, result.getErrorMessage());
            return result.getErrorMessage();
        }
    }
    return null;
}
 

/**
 * This method deletes the entity from the collection for a given key.
 *
 * @param tableName Name of the table
 * @param entity    Entity
 * @throws ODataServiceFault
 */
@Override
public boolean deleteEntityInTable(String tableName, ODataEntry entity) throws ODataServiceFault {
    String documentId = entity.getValue(DOCUMENT_ID);
    WriteResult delete = jongo.getCollection(tableName).remove(new ObjectId(documentId));
    int wasDeleted = delete.getN();
    if (wasDeleted == 1) {
        return delete.wasAcknowledged();
    } else {
        throw new ODataServiceFault("Document ID: " + documentId + " does not exist in "
                + "collection: " + tableName + ".");
    }
}
 

/**
 * This method updates the entity in table when transactional update is necessary.
 *
 * @param tableName     Table Name
 * @param oldProperties Old Properties
 * @param newProperties New Properties
 * @throws ODataServiceFault
 */
@Override
public boolean updateEntityInTableTransactional(String tableName, ODataEntry oldProperties,
                                                ODataEntry newProperties) throws ODataServiceFault {
    String oldPropertyObjectKeyValue = oldProperties.getValue(DOCUMENT_ID);
    StringBuilder updateNewProperties = new StringBuilder();
    updateNewProperties.append("{$set: {");
    boolean propertyMatch = false;
    for (String column : newProperties.getData().keySet()) {
        if (propertyMatch) {
            updateNewProperties.append("', ");
        }
        String propertyValue = newProperties.getValue(column);
        updateNewProperties.append(column).append(": '").append(propertyValue);
        propertyMatch = true;
    }
    updateNewProperties.append("'}}");
    String query = updateNewProperties.toString();
    WriteResult update = jongo.getCollection(tableName).update(new ObjectId(oldPropertyObjectKeyValue)).with(query);
    int wasUpdated = update.getN();
    if (wasUpdated == 1) {
        return update.wasAcknowledged();
    } else {
        throw new ODataServiceFault("Error occured while updating the entity to collection :"
                + tableName + ".");
    }
}
 

@Override
public void updateFailTransaction(final 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(ERROR);
    }
}
 

@Override
public void updateParticipant(final 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(ERROR);
    }
}
 

@Override
public int updateStatus(final String id, final Integer status) throws MythRuntimeException {
    Query query = new Query();
    query.addCriteria(new Criteria("transId").is(id));
    Update update = new Update();
    update.set("status", status);
    final WriteResult writeResult = template.updateFirst(query, update, MongoAdapter.class, collectionName);
    if (writeResult.getN() <= 0) {
        throw new MythRuntimeException(ERROR);
    }
    return CommonConstant.SUCCESS;
}
 

@Override
public WriteResult changePassword( final String username, final String password) {
    Query query = new Query();
    query.addCriteria(new Criteria("username").is(username));
    Update update = new Update();
    update.set("password", password);
    update.set("lastPasswordResetDate", new Date());
    return this.mongoTemplate.updateFirst(query, update, UserDO.class);
}
 

@Override
public WriteResult changeUserInfo(UserDO userDO) {
    Query query = new Query();
    query.addCriteria(new Criteria("username").is(userDO.getUsername()));
    Update update = new Update();
    if (userDO.getBlogLink()!=null){
        update.set("blogLink", userDO.getBlogLink());
    }
    if (userDO.getAvatar()!=null){
        update.set("avatar", userDO.getAvatar());
    }
    return this.mongoTemplate.upsert(query, update, UserDO.class);
}
 

/**
 * 更改密码
 * @param code
 * @param password
 * @param responseMsgVO
 */
public void changePwd(String code, String password, ResponseMsgVO responseMsgVO) {
    String username=  UserContextHandler.getUsername();


    UserDO userDO = userRepository.findByUsername(username);
    log.info("用户:{}更改密码",userDO.getUsername());
    String key = CODE_PREFIX + CodeTypeEnum.CHANGE_PWD.getValue() + userDO.getEmail();
    String value = redisTemplate.opsForValue().get(key);
    if (!StringUtils.equals(code, value)) {
        responseMsgVO.buildWithMsgAndStatus(PosCodeEnum.PARAM_ERROR, "验证码错误");
        return;
    }

    if (!checkPassWordLegal(password, responseMsgVO)){
        return;
    }

    // 更新到mongo数据库
    BCryptPasswordEncoder passwordEncoder = new BCryptPasswordEncoder();
    WriteResult result = userRepository.changePassword(username, passwordEncoder.encode(password));
    if (result.getN() == 0) {
        responseMsgVO.buildWithMsgAndStatus(PosCodeEnum.PARAM_ERROR, "更新失败");
        return;
    }
    responseMsgVO.buildOK();
}
 

private Boolean changeUserInfo(UserDO userDO){
    WriteResult result= userRepository.changeUserInfo(userDO);
    //转换DTO对象
    UserDTO userDTO = UserConvert.doToDto(userDO);
    Boolean eschange=  elasticSearchService.userAdd(userDTO);
    if (eschange!=null && result.getN()!=0){
        return true;
    }else {
        return false;
    }
}
 

@Override
public boolean updateProxyById(String id) {

    Query query = new Query(Criteria.where("id").is(id));
    Update update = new Update();
    update.set("lastSuccessfulTime", new Date().getTime());        //最近一次验证成功的时间
    WriteResult writeResult = mongoTemplate.updateFirst(query, update, ProxyData.class,Constant.COL_NAME_PROXY);
    return writeResult!=null && writeResult.getN() > 0;
}
 

@Override
public boolean deleteProxyById(String id) {

    Query query = new Query(Criteria.where("id").is(id));
    WriteResult writeResult = mongoTemplate.remove(query, ProxyData.class, Constant.COL_NAME_PROXY);
    return writeResult!=null && writeResult.getN() > 0;
}
 

@Override
public boolean saveProxyResource(ProxyResource proxyResource) {
    boolean result = false;
    if(Preconditions.isBlank(proxyResource.getResId())) {            //insert
        proxyResource.setResId(commonDao.getNextSequence(Constant.COL_NAME_PROXY_RESOURCE).getSequence());
        proxyResource.setAddTime(new Date().getTime());
        proxyResource.setModTime(new Date().getTime());
        mongoTemplate.save(proxyResource, Constant.COL_NAME_PROXY_RESOURCE);

        result = Preconditions.isNotBlank(proxyResource.getId());
    } else {                                                        //update
        Query query = new Query().addCriteria(Criteria.where("resId").is(proxyResource.getResId()));
        Update update = new Update();
        update.set("webName", proxyResource.getWebName());
        update.set("webUrl", proxyResource.getWebUrl());
        update.set("pageCount", proxyResource.getPageCount());
        update.set("prefix", proxyResource.getPrefix());
        update.set("suffix", proxyResource.getSuffix());
        update.set("parser", proxyResource.getParser());
        update.set("modTime", new Date().getTime());

        WriteResult writeResult = mongoTemplate.updateFirst(query, update, Constant.COL_NAME_PROXY_RESOURCE);

        result = writeResult!=null && writeResult.getN() > 0;
    }
    return result;
}
 

public int delete(DBCollection collection, WriteConcern writeConcern) {
	DBObject query = new BasicDBObject();
	if (null != condition) {
		this.condition.setQuery(query, null);
	}

	log(query);

	// WriteResult result = collection.remove(query, WriteConcern.ACKNOWLEDGED);
	WriteResult result = collection.remove(query, writeConcern);
	// collection.remove(query)
	// System.out.println(query.toString());
	return result.getN();
}
 

public void submitToPubmedDB(PubmedEntry entry) {
  List<String> xPath = new ArrayList<String>();
  xPath.add("MedlineCitation"); xPath.add("PMID");
  int pmid = Integer.parseInt(entry.getXPathString(xPath));
  if(this.dbPubmed != null) {
    WriteResult result;
    if (alreadyEntered(entry, pmid))
      return;
    DBObject doc = (DBObject)JSON.parse(entry.toJSON());
    doc.put("_id", pmid);
    this.dbPubmed.insert(doc);
  } else
    Logger.printf(0, "Pubmed Entry [%d]: %s\n", pmid, entry); // human readable...
}
 

public void sendJSON(Object obj, HttpServletRequest request, HttpServletResponse response) throws IOException {
	boolean gzip = false;
	if (request != null) {
		String acceptedEncodings = request.getHeader("accept-encoding");
		gzip = acceptedEncodings != null && acceptedEncodings.indexOf("gzip") != -1;
	}
	response.setCharacterEncoding("UTF-8");
	response.setContentType("application/json");
	if (obj instanceof WriteResult) {
		String error = ((WriteResult) obj).getError();
		if (error != null) {
			obj = error;
		} else {
			obj = "OK";
		}
	}
	OutputStream os = null;
	try {
		byte data[] = JSON.serialize(obj).getBytes("UTF-8");
		os = response.getOutputStream();
		if (gzip && data.length >= 860) {
			response.setHeader("Content-Encoding", "gzip");
			GZIPOutputStream gzos = new GZIPOutputStream(os);
			gzos.write(data);
			gzos.finish();
			gzos.close();
		} else {
			os.write(data);
		}
	} catch (Exception e) {
		e.printStackTrace();
		response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, e.getMessage());
	} finally {
		if (os != null) {
			os.close();
		}
	}
}
 

@Override
public void endWindow()
{
  logger.debug("mongo datalist size: " + dataList.size());
  if (dataList.size() > 0) {
    WriteResult result = dbCollection.insert(dataList, writeConcern);
    logger.debug("Result for MongoDB insert: " + result);
    dataList.clear();
  }
}
 

@Override
public boolean remove(String jobId) {
    Query<JobPo> query = template.createQuery(JobPo.class);
    query.field("jobId").equal(jobId);
    WriteResult wr = template.delete(query);
    return wr.getN() == 1;
}
 

@Override
public boolean remove(String jobClientNodeGroup, String id) {
    Query<JobFeedbackPo> query = createQuery(jobClientNodeGroup);
    query.field("id").equal(id);
    WriteResult wr = template.delete(query);
    return wr.getN() == 1;
}
 

@Override
public boolean remove(String jobId) {
    Query<JobPo> query = template.createQuery(JobPo.class);
    query.field("jobId").equal(jobId);
    WriteResult wr = template.delete(query);
    return wr.getN() == 1;
}
 

@Override
public boolean remove(String jobId) {
    Query<JobPo> query = template.createQuery(JobPo.class);
    query.field("jobId").equal(jobId);
    WriteResult wr = template.delete(query);
    return wr.getN() == 1;
}
 

@Override
public boolean remove(String taskTrackerNodeGroup, String jobId) {
    String tableName = JobQueueUtils.getExecutableQueueName(taskTrackerNodeGroup);
    Query<JobPo> query = template.createQuery(tableName, JobPo.class);
    query.field("jobId").equal(jobId);
    WriteResult wr = template.delete(query);
    return wr.getN() == 1;
}