entitysyncservices.java

国外的一套开源CRM

            long perSplitMinItems = Long.MAX_VALUE;
            long perSplitMaxItems = 0;

            long startingTimeMillis = System.currentTimeMillis();
            
            // increment starting time to run until now
            while (currentRunStartTime.before(syncEndStamp)) {
                long splitStartTime = System.currentTimeMillis();
                
                Timestamp currentRunEndTime = new Timestamp(currentRunStartTime.getTime() + splitMillis);
                if (currentRunEndTime.after(syncEndStamp)) {
                    currentRunEndTime = syncEndStamp;
                }
                
                // make sure the following message is commented out before commit:
                //Debug.logInfo("Doing runEntitySync split, currentRunStartTime=" + currentRunStartTime + ", currentRunEndTime=" + currentRunEndTime, module);
                
                totalSplits++;
                
                // tx times are indexed
                // keep track of how long these sync runs take and store that info on the history table
                // saves info about removed, all entities that don't have no-auto-stamp set, this will be done in the GenericDAO like the stamp sets
                
                // ===== INSERTS =====
                
                // first grab all values inserted in the date range, then get the updates (leaving out all values inserted in the data range)
                ArrayList valuesToCreate = new ArrayList(); // make it an ArrayList to easily merge in sorted lists

                // iterate through entities, get all records with tx stamp in the current time range, put all in a single list
                Iterator entityModelToUseCreateIter = entityModelToUseList.iterator();
                while (entityModelToUseCreateIter.hasNext()) {
                    int insertBefore = 0;
                    ModelEntity modelEntity = (ModelEntity) entityModelToUseCreateIter.next();
                    // get the values created within the current time range
                    EntityCondition findValCondition = new EntityConditionList(UtilMisc.toList(
                            new EntityExpr(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.GREATER_THAN_EQUAL_TO, currentRunStartTime), 
                            new EntityExpr(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.LESS_THAN, currentRunEndTime)), EntityOperator.AND);
                    EntityListIterator eli = delegator.findListIteratorByCondition(modelEntity.getEntityName(), findValCondition, null, UtilMisc.toList(ModelEntity.CREATE_STAMP_TX_FIELD, ModelEntity.CREATE_STAMP_FIELD));
                    GenericValue nextValue = null;
                    //long valuesPerEntity = 0;
                    while ((nextValue = (GenericValue) eli.next()) != null) {
                        // sort by the tx stamp and then the record stamp 
                        // find first value in valuesToStore list, starting with the current insertBefore value, that has a CREATE_STAMP_TX_FIELD after the nextValue.CREATE_STAMP_TX_FIELD, then do the same with CREATE_STAMP_FIELD
                        while (insertBefore < valuesToCreate.size() && ((GenericValue) valuesToCreate.get(insertBefore)).getTimestamp(ModelEntity.CREATE_STAMP_TX_FIELD).before(nextValue.getTimestamp(ModelEntity.CREATE_STAMP_TX_FIELD))) {
                            insertBefore++;
                        }
                        while (insertBefore < valuesToCreate.size() && ((GenericValue) valuesToCreate.get(insertBefore)).getTimestamp(ModelEntity.CREATE_STAMP_FIELD).before(nextValue.getTimestamp(ModelEntity.CREATE_STAMP_FIELD))) {
                            insertBefore++;
                        }
                        valuesToCreate.add(insertBefore, nextValue);
                        //valuesPerEntity++;
                    }
                    eli.close();
                    
                    // definately remove this message and related data gathering
                    //long preCount = delegator.findCountByCondition(modelEntity.getEntityName(), findValCondition, null);
                    //long entityTotalCount = delegator.findCountByCondition(modelEntity.getEntityName(), null, null);
                    //if (entityTotalCount > 0 || preCount > 0 || valuesPerEntity > 0) Debug.logInfo("Got " + valuesPerEntity + "/" + preCount + "/" + entityTotalCount + " values for entity " + modelEntity.getEntityName(), module);
                }
                
                // ===== UPDATES =====
                
                // simulate two ordered lists and merge them on-the-fly for faster combined sorting
                ArrayList valuesToStore = new ArrayList(); // make it an ArrayList to easily merge in sorted lists

                // iterate through entities, get all records with tx stamp in the current time range, put all in a single list
                Iterator entityModelToUseUpdateIter = entityModelToUseList.iterator();
                while (entityModelToUseUpdateIter.hasNext()) {
                    int insertBefore = 0;
                    ModelEntity modelEntity = (ModelEntity) entityModelToUseUpdateIter.next();
                    // get all values that were updated, but NOT created in the current time range; if no info on created stamp, that's okay we'll include it here because it won't have been included in the valuesToCreate list
                    EntityCondition createdBeforeStartCond = new EntityExpr(
                            new EntityExpr(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.EQUALS, null), 
                            EntityOperator.OR, 
                            new EntityExpr(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.LESS_THAN, currentRunStartTime));
                    EntityCondition findValCondition = new EntityConditionList(UtilMisc.toList(
                            new EntityExpr(ModelEntity.STAMP_TX_FIELD, EntityOperator.GREATER_THAN_EQUAL_TO, currentRunStartTime), 
                            new EntityExpr(ModelEntity.STAMP_TX_FIELD, EntityOperator.LESS_THAN, currentRunEndTime), 
                            createdBeforeStartCond), 
                            EntityOperator.AND);
                    EntityListIterator eli = delegator.findListIteratorByCondition(modelEntity.getEntityName(), findValCondition, null, UtilMisc.toList(ModelEntity.STAMP_TX_FIELD, ModelEntity.STAMP_FIELD));
                    GenericValue nextValue = null;
                    //long valuesPerEntity = 0;
                    while ((nextValue = (GenericValue) eli.next()) != null) {
                        // sort by the tx stamp and then the record stamp 
                        // find first value in valuesToStore list, starting with the current insertBefore value, that has a STAMP_TX_FIELD after the nextValue.STAMP_TX_FIELD, then do the same with STAMP_FIELD
                        while (insertBefore < valuesToStore.size() && ((GenericValue) valuesToStore.get(insertBefore)).getTimestamp(ModelEntity.STAMP_TX_FIELD).before(nextValue.getTimestamp(ModelEntity.STAMP_TX_FIELD))) {
                            insertBefore++;
                        }
                        while (insertBefore < valuesToStore.size() && ((GenericValue) valuesToStore.get(insertBefore)).getTimestamp(ModelEntity.STAMP_FIELD).before(nextValue.getTimestamp(ModelEntity.STAMP_FIELD))) {
                            insertBefore++;
                        }
                        valuesToStore.add(insertBefore, nextValue);
                        //valuesPerEntity++;
                    }
                    eli.close();
                    
                    // definately remove this message and related data gathering
                    //long preCount = delegator.findCountByCondition(modelEntity.getEntityName(), findValCondition, null);
                    //long entityTotalCount = delegator.findCountByCondition(modelEntity.getEntityName(), null, null);
                    //if (entityTotalCount > 0 || preCount > 0 || valuesPerEntity > 0) Debug.logInfo("Got " + valuesPerEntity + "/" + preCount + "/" + entityTotalCount + " values for entity " + modelEntity.getEntityName(), module);
                }
                
                // ===== DELETES =====
                
                // get all removed items from the given time range, add to list for those
                List keysToRemove = new LinkedList();
                // find all instances of this entity with the STAMP_TX_FIELD != null, sort ascending to get lowest/oldest value first, then grab first and consider as candidate currentRunStartTime
                EntityCondition findValCondition = new EntityConditionList(UtilMisc.toList(
                        new EntityExpr(ModelEntity.STAMP_TX_FIELD, EntityOperator.GREATER_THAN_EQUAL_TO, currentRunStartTime), 
                        new EntityExpr(ModelEntity.STAMP_TX_FIELD, EntityOperator.LESS_THAN, currentRunEndTime)), EntityOperator.AND);
                EntityListIterator removeEli = delegator.findListIteratorByCondition("EntitySyncRemove", findValCondition, null, UtilMisc.toList(ModelEntity.STAMP_FIELD));
                GenericValue nextValue = null;
                while ((nextValue = (GenericValue) removeEli.next()) != null) {
                    keysToRemove.add(nextValue);
                }
                removeEli.close();
                
                // grab some totals before calling...
                long totalRowsToCreateCur = valuesToCreate.size();
                long totalRowsToStoreCur = valuesToStore.size();
                long totalRowsToRemoveCur = keysToRemove.size();

                long totalRowsPerSplit = totalRowsToCreateCur + totalRowsToStoreCur + totalRowsToRemoveCur;
                
                if (totalRowsPerSplit < perSplitMinItems) {
                    perSplitMinItems = totalRowsPerSplit;
                }
                if (totalRowsPerSplit > perSplitMaxItems) {
                    perSplitMaxItems = totalRowsPerSplit;
                }

                totalRowsToCreate += totalRowsToCreateCur;
                totalRowsToStore += totalRowsToStoreCur;
                totalRowsToRemove += totalRowsToRemoveCur;
                
                // call service named on EntitySync, IFF there is actually data to send over
                if (totalRowsPerSplit > 0) {
                    Map targetServiceMap = UtilMisc.toMap("entitySyncId", entitySyncId, "valuesToCreate", valuesToCreate, "valuesToStore", valuesToStore, "keysToRemove", keysToRemove, "userLogin", userLogin);
                    if (UtilValidate.isNotEmpty(targetDelegatorName)) {
                        targetServiceMap.put("delegatorName", targetDelegatorName);
                    }
                    Map remoteStoreResult = dispatcher.runSync(targetServiceName, targetServiceMap);
                    if (ServiceUtil.isError(remoteStoreResult)) {
                        String errorMsg = "Error running EntitySync [" + entitySyncId + "], call to store service [" + targetServiceName + "] failed.";
                        List errorList = new LinkedList();
                        saveSyncErrorInfo(entitySyncId, startDate, "ESR_OTHER_ERROR", errorList, dispatcher, userLogin);
                        return ServiceUtil.returnError(errorMsg, errorList, null, remoteStoreResult);
                    }
                    
                    totalStoreCalls++;
                    
                    long toCreateInsertedCur = remoteStoreResult.get("toCreateInserted") == null ? 0 : ((Long) remoteStoreResult.get("toCreateInserted")).longValue();
                    long toCreateUpdatedCur = remoteStoreResult.get("toCreateUpdated") == null ? 0 : ((Long) remoteStoreResult.get("toCreateUpdated")).longValue();
                    long toCreateNotUpdatedCur = remoteStoreResult.get("toCreateNotUpdated") == null ? 0 : ((Long) remoteStoreResult.get("toCreateNotUpdated")).longValue();
                    long toStoreInsertedCur = remoteStoreResult.get("toStoreInserted") == null ? 0 : ((Long) remoteStoreResult.get("toStoreInserted")).longValue();
                    long toStoreUpdatedCur = remoteStoreResult.get("toStoreUpdated") == null ? 0 : ((Long) remoteStoreResult.get("toStoreUpdated")).longValue();
                    long toStoreNotUpdatedCur = remoteStoreResult.get("toStoreNotUpdated") == null ? 0 : ((Long) remoteStoreResult.get("toStoreNotUpdated")).longValue();
                    long toRemoveDeletedCur = remoteStoreResult.get("toRemoveDeleted") == null ? 0 : ((Long) remoteStoreResult.get("toRemoveDeleted")).longValue();
                    long toRemoveAlreadyDeletedCur = remoteStoreResult.get("toRemoveAlreadyDeleted") == null ? 0 : ((Long) remoteStoreResult.get("toRemoveAlreadyDeleted")).longValue();
                    
                    toCreateInserted += toCreateInsertedCur;
                    toCreateUpdated += toCreateUpdatedCur;
                    toCreateNotUpdated += toCreateNotUpdatedCur;
                    toStoreInserted += toStoreInsertedCur;
                    toStoreUpdated += toStoreUpdatedCur;
                    toStoreNotUpdated += toStoreNotUpdatedCur;
                    toRemoveDeleted += toRemoveDeletedCur;
                    toRemoveAlreadyDeleted += toRemoveAlreadyDeletedCur;
                }

                long splitTotalTime = System.currentTimeMillis() - splitStartTime;
                if (splitTotalTime < perSplitMinMillis) {
                    perSplitMinMillis = splitTotalTime;
                }
                if (splitTotalTime > perSplitMaxMillis) {
                    perSplitMaxMillis = splitTotalTime;
                }
                
                long runningTimeMillis = System.currentTimeMillis() - startingTimeMillis;
                
                // store latest result on EntitySync, ie update lastSuccessfulSynchTime, should run in own tx
                Map updateEsRunResult = dispatcher.runSync("updateEntitySyncRunning", UtilMisc.toMap("entitySyncId", entitySyncId, "lastSuccessfulSynchTime", currentRunEndTime, "userLogin", userLogin));