entitysynccontext.java

Sequoia ERP是一个真正的企业级开源ERP解决方案。它提供的模块包括：电子商务应用(e-commerce), POS系统(point of sales),知识管理,存货与仓库管理

        if (entitiesSkippedForKnownNext > 0) {
            if (Debug.infoOn()) Debug.logInfo("In assembleValuesToCreate skipped [" + entitiesSkippedForKnownNext + "/" + entityModelToUseList + "] entities for the time period ending at [" + currentRunEndTime + "] because of next known create times", module);
        }
        
        // TEST SECTION: leave false for normal use
        boolean logValues = false;
        if (logValues && valuesToCreate.size() > 0) {
            StringBuffer toCreateInfo = new StringBuffer();
            Iterator valuesToCreateIter = valuesToCreate.iterator();
            while (valuesToCreateIter.hasNext()) {
                GenericValue valueToCreate = (GenericValue) valuesToCreateIter.next();
                toCreateInfo.append("\n-->[");
                toCreateInfo.append(valueToCreate.get(ModelEntity.CREATE_STAMP_TX_FIELD));
                toCreateInfo.append(":");
                toCreateInfo.append(valueToCreate.get(ModelEntity.CREATE_STAMP_FIELD));
                toCreateInfo.append("] ");
                toCreateInfo.append(valueToCreate.getPrimaryKey());
            }
            Debug.logInfo(toCreateInfo.toString(), module);
        }
        
        return valuesToCreate;
    }

    public ArrayList assembleValuesToStore() throws SyncDataErrorException {
        // 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

        if (this.nextUpdateTxTime != null && (this.nextUpdateTxTime.equals(currentRunEndTime) || this.nextUpdateTxTime.after(currentRunEndTime))) {
            // this means that for all entities in this pack we found on the last pass that there would be nothing for this one, so just return nothing...
            return valuesToStore;
        }

        // Debug.logInfo("Getting values to store; currentRunStartTime=" + currentRunStartTime + ", currentRunEndTime=" + currentRunEndTime, module);
        
        int entitiesSkippedForKnownNext = 0;
        
        // 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();
            
            // first test to see if we know that there are no records for this entity in this time period...
            Timestamp knownNextUpdateTime = (Timestamp) this.nextEntityUpdateTxTime.get(modelEntity.getEntityName());
            if (knownNextUpdateTime != null && (knownNextUpdateTime.equals(currentRunEndTime) || knownNextUpdateTime.after(currentRunEndTime))) {
                entitiesSkippedForKnownNext++;
                continue;
            }

            boolean beganTransaction = false;
            try {
                beganTransaction = TransactionUtil.begin(7200);
            } catch (GenericTransactionException e) {
                throw new SyncDataErrorException("Unable to begin JTA transaction", e);
            }

            try {
                // 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);

                // if we didn't find anything for this entity, find the next value's Timestamp and keep track of it
                if (valuesPerEntity == 0) {
                    Timestamp startCheckStamp = new Timestamp(System.currentTimeMillis() - syncEndBufferMillis);
                    
                    EntityCondition findNextCondition = new EntityConditionList(UtilMisc.toList(
                            new EntityExpr(ModelEntity.STAMP_TX_FIELD, EntityOperator.NOT_EQUAL, null), 
                            new EntityExpr(ModelEntity.STAMP_TX_FIELD, EntityOperator.GREATER_THAN_EQUAL_TO, currentRunEndTime),
                            new EntityExpr(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.NOT_EQUAL, null),
                            new EntityExpr(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.LESS_THAN, currentRunEndTime)), 
                            EntityOperator.AND);
                    EntityListIterator eliNext = delegator.findListIteratorByCondition(modelEntity.getEntityName(), findNextCondition, null, UtilMisc.toList(ModelEntity.STAMP_TX_FIELD));
                    // get the first element and it's tx time value...
                    GenericValue firstVal = (GenericValue) eliNext.next();
                    eliNext.close();
                    Timestamp nextTxTime;
                    if (firstVal != null) {
                        nextTxTime = firstVal.getTimestamp(ModelEntity.CREATE_STAMP_TX_FIELD);
                    } else {
                        // no results? well, then it's safe to say that up to the pre-querytime (minus the buffer, as usual) we are okay
                        nextTxTime = startCheckStamp;
                    }
                    if (this.nextUpdateTxTime == null || nextTxTime.before(this.nextUpdateTxTime)) {
                        this.nextUpdateTxTime = nextTxTime;
                        Debug.logInfo("EntitySync: Set nextUpdateTxTime to [" + nextTxTime + "]", module);
                    }
                    Timestamp curEntityNextTxTime = (Timestamp) this.nextEntityUpdateTxTime.get(modelEntity.getEntityName());
                    if (curEntityNextTxTime == null || nextTxTime.before(curEntityNextTxTime)) {
                        this.nextEntityUpdateTxTime.put(modelEntity.getEntityName(), nextTxTime);
                        Debug.logInfo("EntitySync: Set nextEntityUpdateTxTime to [" + nextTxTime + "] for the entity [" + modelEntity.getEntityName() + "]", module);
                    }
                }
            } catch (GenericEntityException e) {
                try {
                    TransactionUtil.rollback(beganTransaction, "Entity Engine error in assembleValuesToStore", e);
                } catch (GenericTransactionException e2) {
                    Debug.logWarning(e2, "Unable to call rollback()", module);
                }
                throw new SyncDataErrorException("Error getting values to store from the datasource", e);
            } catch (Throwable t) {
                try {
                    TransactionUtil.rollback(beganTransaction, "General error in assembleValuesToStore", t);
                } catch (GenericTransactionException e2) {
                    Debug.logWarning(e2, "Unable to call rollback()", module);
                }
                throw new SyncDataErrorException("Caught runtime error while getting values to store", t);
            }

            try {
                TransactionUtil.commit(beganTransaction);
            } catch (GenericTransactionException e) {
                throw new SyncDataErrorException("Commit transaction failed", e);
            }
        }

        if (entitiesSkippedForKnownNext > 0) {
            if (Debug.infoOn()) Debug.logInfo("In assembleValuesToStore skipped [" + entitiesSkippedForKnownNext + "/" + entityModelToUseList + "] entities for the time period ending at [" + currentRunEndTime + "] because of next known update times", module);
        }
        
        // TEST SECTION: leave false for normal use
        boolean logValues = false;
        if (logValues && valuesToStore.size() > 0) {
            StringBuffer toStoreInfo = new StringBuffer();
            Iterator valuesToStoreIter = valuesToStore.iterator();
            while (valuesToStoreIter.hasNext()) {
                GenericValue valueToStore = (GenericValue) valuesToStoreIter.next();
                toStoreInfo.append("\n-->[");
                toStoreInfo.append(valueToStore.get(ModelEntity.STAMP_TX_FIELD));
                toStoreInfo.append(":");
                toStoreInfo.append(valueToStore.get(ModelEntity.STAMP_FIELD));
                toStoreInfo.append("] ");
                toStoreInfo.append(valueToStore.getPrimaryKey());
            }
            Debug.logInfo(toStoreInfo.toString(), module);
        }
        
        return valuesToStore;
    }

    public LinkedList assembleKeysToRemove() throws SyncDataErrorException {
        // get all removed items from the given time range, add to list for those
        LinkedList keysToRemove = new LinkedList();

        if (this.nextRemoveTxTime != null && (this.nextRemoveTxTime.equals(currentRunEndTime) || this.nextRemoveTxTime.after(currentRunEndTime))) {
            // this means that for all entities in this pack we found on the last pass that there would be nothing for this one, so just return nothing...
            return keysToRemove;
        }

        //Debug.logInfo("Getting keys to remove; currentRunStartTime=" + currentRunStartTime + ", currentRunEndTime=" + currentRunEndTime, module);

        boolean beganTransaction = false;
        try {
            beganTransaction = TransactionUtil.begin(7200);
        } catch (GenericTransactionException e) {
            throw new SyncDataErrorException("Unable to begin JTA transaction", e);
        }

        try {
            // 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_TX_FIELD, ModelEntity.STAMP_FIELD));
            GenericValue entitySyncRemove = null;
            while ((entitySyncRemove = (GenericValue) removeEli.next()) != null) {
                // pull the PK from the EntitySyncRemove in the primaryKeyRemoved field, de-XML-serialize it 
                String primaryKeyRemoved = entitySyncRemove.getString("primaryKeyRemoved");
                GenericEntity pkToRemove = null;
                try {
                    pkToRemove = (GenericEntity) XmlSerializer.deserialize(primaryKeyRemoved, delegator);
                } catch (IOException e) {
                    String errorMsg = "Error deserializing GenericPK to remove in Entity Sync Data for entitySyncId [" + entitySyncId + "] and entitySyncRemoveId [" + entitySyncRemove.getString("entitySyncRemoveId") + "]: " + e.toString();
                    Debug.logError(e, errorMsg, module);
                    throw new SyncDataErrorException(errorMsg, e);
                } catch (SAXException e) {
                    String errorMsg = "Error deserializing GenericPK to remove in Entity Sync Data for entitySyncId [" + entitySyncId + "] and entitySyncRemoveId [" + entitySyncRemove.getString("entitySyncRemoveId") + "]: " + e.toString();
                    Debug.logError(e, errorMsg, module);
                    throw new SyncDataErrorException(errorMsg, e);
                } catch (ParserConfigurationException e) {
                    String errorMsg = "Error deserializing GenericPK to remove in Entity Sync Data for entitySyncId [" + entitySyncId + "] and entitySyncRemoveId [" + entitySyncRemove.getString("entitySyncRemoveId") + "]: " + e.toString();
                    Debug.logError(e, errorMsg, module);
                    throw new SyncDataErrorException(errorMsg, e);
                } catch (SerializeException e) {
                    String errorMsg = "Error deserializing GenericPK to remove in Entity Sync Data for entitySyncId [" + entitySyncId + "] and entitySyncRemoveId [" + entitySyncRemove.getString("entitySyncRemoveId") + "]: " + e.toString();
                    Debug.logError(e, errorMsg, module);
                    throw new SyncDataErrorException(errorMsg, e);
                }
                
                // set the stamp fields for future reference