/*******************************************************************************
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 *******************************************************************************/
package org.apache.ofbiz.entityext.synchronization;

import java.io.IOException;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import javax.xml.parsers.ParserConfigurationException;

import org.apache.ofbiz.base.util.Debug;
import org.apache.ofbiz.base.util.UtilDateTime;
import org.apache.ofbiz.base.util.UtilMisc;
import org.apache.ofbiz.base.util.UtilValidate;
import org.apache.ofbiz.entity.Delegator;
import org.apache.ofbiz.entity.DelegatorFactory;
import org.apache.ofbiz.entity.GenericEntity;
import org.apache.ofbiz.entity.GenericEntityException;
import org.apache.ofbiz.entity.GenericValue;
import org.apache.ofbiz.entity.condition.EntityCondition;
import org.apache.ofbiz.entity.condition.EntityOperator;
import org.apache.ofbiz.entity.model.ModelEntity;
import org.apache.ofbiz.entity.serialize.SerializeException;
import org.apache.ofbiz.entity.serialize.XmlSerializer;
import org.apache.ofbiz.entity.transaction.GenericTransactionException;
import org.apache.ofbiz.entity.transaction.TransactionUtil;
import org.apache.ofbiz.entity.util.EntityListIterator;
import org.apache.ofbiz.entity.util.EntityQuery;
import org.apache.ofbiz.entityext.EntityGroupUtil;
import org.apache.ofbiz.service.DispatchContext;
import org.apache.ofbiz.service.GeneralServiceException;
import org.apache.ofbiz.service.GenericServiceException;
import org.apache.ofbiz.service.LocalDispatcher;
import org.apache.ofbiz.service.ModelService;
import org.apache.ofbiz.service.ServiceUtil;
import org.xml.sax.SAXException;


/**
 * Entity Engine Sync Services
 */
public class EntitySyncContext {

    public static final String module = EntitySyncContext.class.getName();

    // set default split to 10 seconds, ie try not to get too much data moving over at once
    public static final long defaultSyncSplitMillis = 10000;

    // default offline split is 30 minutes
    public static final long defaultOfflineSyncSplitMillis = 1800000;

    // default to 5 minutes
    public static final long defaultSyncEndBufferMillis = 300000;

    // default to 2 hours, 120m, 7200s
    public static final long defaultMaxRunningNoUpdateMillis = 7200000;

    public Delegator delegator;
    public LocalDispatcher dispatcher;
    public Map<String, ? extends Object> context;

    public GenericValue userLogin;
    public boolean isOfflineSync = false;

    public String entitySyncId;
    public GenericValue entitySync;

    public String targetServiceName;
    public String targetDelegatorName;

    public Timestamp syncEndStamp;
    public long offlineSyncSplitMillis = defaultOfflineSyncSplitMillis;
    public long syncSplitMillis = defaultSyncSplitMillis;
    public long syncEndBufferMillis = defaultSyncEndBufferMillis;
    public long maxRunningNoUpdateMillis = defaultMaxRunningNoUpdateMillis;

    public Timestamp lastSuccessfulSynchTime;
    public List<ModelEntity> entityModelToUseList;
    public Set<String> entityNameToUseSet;
    public Timestamp currentRunStartTime;
    public Timestamp currentRunEndTime;

    // these values are used to make this more efficient; if we run into an entity that has 0
    //results for a given time block, we will do a query to find the next create/update/remove
    //time for that entity, and also keep track of a global next with the lowest future next value;
    //using these we can skip a lot of queries and speed this up significantly
    public Map<String, Timestamp> nextEntityCreateTxTime = new HashMap<String, Timestamp>();
    public Map<String, Timestamp> nextEntityUpdateTxTime = new HashMap<String, Timestamp>();
    public Timestamp nextCreateTxTime = null;
    public Timestamp nextUpdateTxTime = null;
    public Timestamp nextRemoveTxTime = null;

    // this is the other part of the history PK, leave null until we create the history object
    public Timestamp startDate = null;

    long toCreateInserted = 0;
    long toCreateUpdated = 0;
    long toCreateNotUpdated = 0;
    long toStoreInserted = 0;
    long toStoreUpdated = 0;
    long toStoreNotUpdated = 0;
    long toRemoveDeleted = 0;
    long toRemoveAlreadyDeleted = 0;

    long totalRowsExported = 0;
    long totalRowsToCreate = 0;
    long totalRowsToStore = 0;
    long totalRowsToRemove = 0;

    long totalRowsPerSplit = 0;
    long totalStoreCalls = 0;
    long totalSplits = 0;
    long perSplitMinMillis = Long.MAX_VALUE;
    long perSplitMaxMillis = 0;
    long perSplitMinItems = Long.MAX_VALUE;
    long perSplitMaxItems = 0;
    long splitStartTime = 0;

    public EntitySyncContext(DispatchContext dctx, Map<String, ? extends Object> context) throws SyncDataErrorException, SyncAbortException {
        this.context = context;
        this.dispatcher = dctx.getDispatcher();

        this.delegator = dctx.getDelegator();
        // what to do with the delegatorName? this is the delegatorName to use in this service...
        String delegatorName = (String) context.get("delegatorName");
        if (UtilValidate.isNotEmpty(delegatorName)) {
            this.delegator = DelegatorFactory.getDelegator(delegatorName);
        }


        this.userLogin = (GenericValue) context.get("userLogin");

        this.entitySyncId = (String) context.get("entitySyncId");
        Debug.logInfo("Creating EntitySyncContext with entitySyncId=" + entitySyncId, module);

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

        try {
            this.entitySync = EntityQuery.use(delegator).from("EntitySync").where("entitySyncId", this.entitySyncId).queryOne();
            if (this.entitySync == null) {
                throw new SyncAbortException("Not running EntitySync [" + entitySyncId + "], no record found with that ID.");
            }

            targetServiceName = entitySync.getString("targetServiceName");
            targetDelegatorName = entitySync.getString("targetDelegatorName");

            // make the last time to sync X minutes before the current time so that if this machines clock is up to that amount of time
            //ahead of another machine writing to the DB it will still work fine and not lose any data
            syncEndStamp = new Timestamp(System.currentTimeMillis() - syncEndBufferMillis);

            this.offlineSyncSplitMillis = getOfflineSyncSplitMillis(entitySync);
            this.syncSplitMillis = getSyncSplitMillis(entitySync);
            this.syncEndBufferMillis = getSyncEndBufferMillis(entitySync);
            this.maxRunningNoUpdateMillis = getMaxRunningNoUpdateMillis(entitySync);

            this.lastSuccessfulSynchTime = entitySync.getTimestamp("lastSuccessfulSynchTime");
            this.entityModelToUseList = this.makeEntityModelToUseList();
            this.entityNameToUseSet = this.makeEntityNameToUseSet();

            // set start and end times for the first/current pass
            this.currentRunStartTime = getCurrentRunStartTime(lastSuccessfulSynchTime, entityModelToUseList, delegator);
            this.setCurrentRunEndTime();

            // this is mostly for the pull side... will always be null for at the beginning of a push process, to be filled in later
            this.startDate = (Timestamp) context.get("startDate");
        } catch (GenericEntityException e) {
            try {
                TransactionUtil.rollback(beganTransaction, "Entity Engine error while getting Entity Sync init information", e);
            } catch (GenericTransactionException e2) {
                Debug.logWarning(e2, "Unable to call rollback()", module);
            }
            throw new SyncDataErrorException("Error initializing EntitySync Context", e);
        }

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

    /**
     * To see if it is running check:
     *  - in the running status
     *  - AND when the entitySync was last updated, and if it was more than maxRunningNoUpdateMillis ago, then don't consider it to be running
     * @return boolean representing if the EntitySync should be considered running
     */
    public boolean isEntitySyncRunning() {
        boolean isInRunning = ("ESR_RUNNING".equals(this.entitySync.getString("runStatusId")) ||
                "ESR_PENDING".equals(this.entitySync.getString("runStatusId")));

        if (!isInRunning) {
            return false;
        }

        Timestamp esLastUpdated = this.entitySync.getTimestamp(ModelEntity.STAMP_FIELD);
        if (esLastUpdated == null) {
            // shouldn't ever happen, but just in case; assume is running if we don't know when it was last updated
            return true;
        }
        long esLastUpdatedMillis = esLastUpdated.getTime();
        long nowTimestampMillis = UtilDateTime.nowTimestamp().getTime();
        long timeSinceUpdated = nowTimestampMillis - esLastUpdatedMillis;
        if (timeSinceUpdated > this.maxRunningNoUpdateMillis) {
            // it has been longer than the maxRunningNoUpdateMillis, so don't consider it running
            return false;
        }

        return true;
    }

    public boolean hasMoreTimeToSync() {
        return currentRunStartTime.before(syncEndStamp);
    }

    protected void setCurrentRunEndTime() {
        this.currentRunEndTime = getNextRunEndTime();
    }

    protected Timestamp getNextRunEndTime() {
        long syncSplit = this.isOfflineSync ? offlineSyncSplitMillis : syncSplitMillis;
        Timestamp nextRunEndTime = new Timestamp(this.currentRunStartTime.getTime() + syncSplit);
        if (nextRunEndTime.after(this.syncEndStamp)) {
            nextRunEndTime = this.syncEndStamp;
        }
        return nextRunEndTime;
    }

    public void advanceRunTimes() {
        this.currentRunStartTime = this.currentRunEndTime;
        this.setCurrentRunEndTime();
    }

    public void setSplitStartTime() {
        this.splitStartTime = System.currentTimeMillis();
    }

    protected static long getSyncSplitMillis(GenericValue entitySync) {
        long splitMillis = defaultSyncSplitMillis;
        Long syncSplitMillis = entitySync.getLong("syncSplitMillis");
        if (syncSplitMillis != null) {
            splitMillis = syncSplitMillis.longValue();
        }
        return splitMillis;
    }

    protected static long getOfflineSyncSplitMillis(GenericValue entitySync) {
        long splitMillis = defaultOfflineSyncSplitMillis;
        Long syncSplitMillis = entitySync.getLong("offlineSyncSplitMillis");
        if (syncSplitMillis != null) {
            splitMillis = syncSplitMillis.longValue();
        }
        return splitMillis;
    }

    protected static long getSyncEndBufferMillis(GenericValue entitySync) {
        long syncEndBufferMillis = defaultSyncEndBufferMillis;
        Long syncEndBufferMillisLong = entitySync.getLong("syncEndBufferMillis");
        if (syncEndBufferMillisLong != null) {
            syncEndBufferMillis = syncEndBufferMillisLong.longValue();
        }
        return syncEndBufferMillis;
    }

    protected static long getMaxRunningNoUpdateMillis(GenericValue entitySync) {
        long maxRunningNoUpdateMillis = defaultMaxRunningNoUpdateMillis;
        Long maxRunningNoUpdateMillisLong = entitySync.getLong("maxRunningNoUpdateMillis");
        if (maxRunningNoUpdateMillisLong != null) {
            maxRunningNoUpdateMillis = maxRunningNoUpdateMillisLong.longValue();
        }
        return maxRunningNoUpdateMillis;
    }

    /** create history record, target service should run in own tx */
    public void createInitialHistory() throws SyncDataErrorException, SyncServiceErrorException {
        String errorMsg = "Not running EntitySync [" + entitySyncId + "], could not create EntitySyncHistory";
        try {
            Map<String, Object> initialHistoryRes = dispatcher.runSync("createEntitySyncHistory", UtilMisc.toMap("entitySyncId", entitySyncId, "runStatusId", "ESR_RUNNING", "beginningSynchTime", this.currentRunStartTime, "lastCandidateEndTime", this.currentRunEndTime, "userLogin", userLogin));
            if (ServiceUtil.isError(initialHistoryRes)) {
                throw new SyncDataErrorException(errorMsg, null, null, initialHistoryRes, null);
            }
            this.startDate = (Timestamp) initialHistoryRes.get("startDate");
        } catch (GenericServiceException e) {
            throw new SyncServiceErrorException(errorMsg, e);
        }
    }

    public ArrayList<GenericValue> assembleValuesToCreate() throws SyncDataErrorException {
        // first grab all values inserted in the date range, then get the updates (leaving out all values inserted in the data range)
        ArrayList<GenericValue> valuesToCreate = new ArrayList<GenericValue>(); // make it an ArrayList to easily merge in sorted lists

        if (this.nextCreateTxTime != null && (this.nextCreateTxTime.equals(currentRunEndTime) || this.nextCreateTxTime.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 valuesToCreate;
        }

        //Debug.logInfo("Getting values to create; 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
        for (ModelEntity modelEntity: entityModelToUseList) {
            int insertBefore = 0;

            // first test to see if we know that there are no records for this entity in this time period...
            Timestamp knownNextCreateTime = this.nextEntityCreateTxTime.get(modelEntity.getEntityName());
            if (knownNextCreateTime != null && (knownNextCreateTime.equals(currentRunEndTime) || knownNextCreateTime.after(currentRunEndTime))) {
                //Debug.logInfo("In assembleValuesToCreate found knownNextCreateTime [" + knownNextCreateTime + "] after currentRunEndTime [" + currentRunEndTime + "], so skipping time per period for entity [" + modelEntity.getEntityName() + "]", module);
                entitiesSkippedForKnownNext++;
                continue;
            }

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

            try {
                // get the values created within the current time range
                EntityCondition findValCondition = EntityCondition.makeCondition(
                        EntityCondition.makeCondition(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.GREATER_THAN_EQUAL_TO, currentRunStartTime),
                        EntityCondition.makeCondition(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.LESS_THAN, currentRunEndTime));
                EntityListIterator eli = EntityQuery.use(delegator)
                                                    .from(modelEntity.getEntityName())
                                                    .where(findValCondition)
                                                    .orderBy(ModelEntity.CREATE_STAMP_TX_FIELD, ModelEntity.CREATE_STAMP_FIELD)
                                                    .queryIterator();
                GenericValue nextValue = null;
                long valuesPerEntity = 0;
                while ((nextValue = eli.next()) != null) {
                    // sort by the tx stamp and then the record stamp
                    // find first value in valuesToCreate 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() && valuesToCreate.get(insertBefore).getTimestamp(ModelEntity.CREATE_STAMP_TX_FIELD).before(nextValue.getTimestamp(ModelEntity.CREATE_STAMP_TX_FIELD))) {
                        insertBefore++;
                    }
                    while (insertBefore < valuesToCreate.size() && 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);

                // 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 = EntityCondition.makeCondition(
                            EntityCondition.makeCondition(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.NOT_EQUAL, null),
                            EntityCondition.makeCondition(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.GREATER_THAN_EQUAL_TO, currentRunEndTime));
                    EntityListIterator eliNext = EntityQuery.use(delegator)
                                                            .from(modelEntity.getEntityName())
                                                            .where(findNextCondition)
                                                            .orderBy(ModelEntity.CREATE_STAMP_TX_FIELD)
                                                            .queryIterator();
                    // get the first element and it's tx time value...
                    GenericValue firstVal = 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.nextCreateTxTime == null || nextTxTime.before(this.nextCreateTxTime)) {
                        this.nextCreateTxTime = nextTxTime;
                        Debug.logInfo("EntitySync: Set nextCreateTxTime to [" + nextTxTime + "]", module);
                    }
                    Timestamp curEntityNextTxTime = this.nextEntityCreateTxTime.get(modelEntity.getEntityName());
                    if (curEntityNextTxTime == null || nextTxTime.before(curEntityNextTxTime)) {
                        this.nextEntityCreateTxTime.put(modelEntity.getEntityName(), nextTxTime);
                        Debug.logInfo("EntitySync: Set nextEntityCreateTxTime to [" + nextTxTime + "] for the entity [" + modelEntity.getEntityName() + "]", module);
                    }
                }
            } catch (GenericEntityException e) {
                try {
                    TransactionUtil.rollback(beganTransaction, "Entity Engine error in assembleValuesToCreate", e);

                } catch (GenericTransactionException e2) {
                    Debug.logWarning(e2, "Unable to call rollback()", module);
                }
                throw new SyncDataErrorException("Error getting values to create from the datasource", e);
            } catch (Throwable t) {
                try {
                    TransactionUtil.rollback(beganTransaction, "Throwable error in assembleValuesToCreate", t);
                } catch (GenericTransactionException e2) {
                    Debug.logWarning(e2, "Unable to call rollback()", module);
                }
                throw new SyncDataErrorException("Caught runtime error while getting values to create", t);
            }

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

        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) {
            StringBuilder toCreateInfo = new StringBuilder();
            for (GenericValue valueToCreate: valuesToCreate) {
                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);
        }
        
        // As the this.nextCreateTxTime calculation is only based on entities without values to create, if there at least one value to create returned
        // this calculation is false, so it needs to be nullified
        if (valuesToCreate.size() > 0) {
            this.nextCreateTxTime = null;
        }
        
        return valuesToCreate;
    }

    public ArrayList<GenericValue> assembleValuesToStore() throws SyncDataErrorException {
        // simulate two ordered lists and merge them on-the-fly for faster combined sorting
        ArrayList<GenericValue> valuesToStore = new ArrayList<GenericValue>(); // 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
        for (ModelEntity modelEntity: entityModelToUseList) {
            int insertBefore = 0;

            // first test to see if we know that there are no records for this entity in this time period...
            Timestamp knownNextUpdateTime = 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 = EntityCondition.makeCondition(
                        EntityCondition.makeCondition(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.EQUALS, null),
                        EntityOperator.OR,
                        EntityCondition.makeCondition(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.LESS_THAN, currentRunStartTime));
                EntityCondition findValCondition = EntityCondition.makeCondition(
                        EntityCondition.makeCondition(ModelEntity.STAMP_TX_FIELD, EntityOperator.GREATER_THAN_EQUAL_TO, currentRunStartTime),
                        EntityCondition.makeCondition(ModelEntity.STAMP_TX_FIELD, EntityOperator.LESS_THAN, currentRunEndTime),
                        createdBeforeStartCond);
                EntityListIterator eli = EntityQuery.use(delegator)
                                                    .from(modelEntity.getEntityName())
                                                    .where(findValCondition)
                                                    .orderBy(ModelEntity.STAMP_TX_FIELD, ModelEntity.STAMP_FIELD)
                                                    .queryIterator();
                GenericValue nextValue = null;
                long valuesPerEntity = 0;
                while ((nextValue = 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() && valuesToStore.get(insertBefore).getTimestamp(ModelEntity.STAMP_TX_FIELD).before(nextValue.getTimestamp(ModelEntity.STAMP_TX_FIELD))) {
                        insertBefore++;
                    }
                    while (insertBefore < valuesToStore.size() && 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 = EntityCondition.makeCondition(
                            EntityCondition.makeCondition(ModelEntity.STAMP_TX_FIELD, EntityOperator.NOT_EQUAL, null),
                            EntityCondition.makeCondition(ModelEntity.STAMP_TX_FIELD, EntityOperator.GREATER_THAN_EQUAL_TO, currentRunEndTime),
                            EntityCondition.makeCondition(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.NOT_EQUAL, null),
                            EntityCondition.makeCondition(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.LESS_THAN, currentRunEndTime));
                    EntityListIterator eliNext = EntityQuery.use(delegator)
                                                            .from(modelEntity.getEntityName())
                                                            .where(findNextCondition)
                                                            .orderBy(ModelEntity.STAMP_TX_FIELD)
                                                            .queryIterator();
                    // get the first element and it's tx time value...
                    GenericValue firstVal = 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 = 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) {
            StringBuilder toStoreInfo = new StringBuilder();
            for (GenericValue valueToStore: valuesToStore) {
                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);
        }
        
        // As the this.nextUpdateTxTime calculation is only based on entities without values to store, if there at least one value to store returned
        // this calculation is false, so it needs to be nullified
        if (valuesToStore.size() > 0) {
            this.nextUpdateTxTime = null;
        }        

        return valuesToStore;
    }

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

        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 = EntityCondition.makeCondition(
                    EntityCondition.makeCondition(ModelEntity.STAMP_TX_FIELD, EntityOperator.GREATER_THAN_EQUAL_TO, currentRunStartTime),
                    EntityCondition.makeCondition(ModelEntity.STAMP_TX_FIELD, EntityOperator.LESS_THAN, currentRunEndTime));
            EntityListIterator removeEli = EntityQuery.use(delegator)
                                                      .from("EntitySyncRemove")
                                                      .where(findValCondition)
                                                      .orderBy(ModelEntity.STAMP_TX_FIELD, ModelEntity.STAMP_FIELD)
                                                      .queryIterator();
            GenericValue entitySyncRemove = null;
            while ((entitySyncRemove = 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
                pkToRemove.set(ModelEntity.STAMP_TX_FIELD, entitySyncRemove.get(ModelEntity.STAMP_TX_FIELD));
                pkToRemove.set(ModelEntity.STAMP_FIELD, entitySyncRemove.get(ModelEntity.STAMP_FIELD));
                pkToRemove.set(ModelEntity.CREATE_STAMP_TX_FIELD, entitySyncRemove.get(ModelEntity.CREATE_STAMP_TX_FIELD));
                pkToRemove.set(ModelEntity.CREATE_STAMP_FIELD, entitySyncRemove.get(ModelEntity.CREATE_STAMP_FIELD));

                if (this.entityNameToUseSet.contains(pkToRemove.getEntityName())) {
                    keysToRemove.add(pkToRemove);
                }
            }
            removeEli.close();

            // if we didn't find anything for this entity, find the next value's Timestamp and keep track of it
            if (keysToRemove.size() == 0) {
                EntityCondition findNextCondition = EntityCondition.makeCondition(ModelEntity.STAMP_TX_FIELD, EntityOperator.GREATER_THAN_EQUAL_TO, currentRunEndTime);
                EntityListIterator eliNext = EntityQuery.use(delegator)
                                                        .from("EntitySyncRemove")
                                                        .where(findNextCondition)
                                                        .orderBy(ModelEntity.STAMP_TX_FIELD)
                                                        .queryIterator();
                // get the first element and it's tx time value...
                GenericValue firstVal = eliNext.next();
                eliNext.close();
                if (firstVal != null) {
                    Timestamp nextTxTime = firstVal.getTimestamp(ModelEntity.STAMP_TX_FIELD);
                    if (this.nextRemoveTxTime == null || nextTxTime.before(this.nextRemoveTxTime)) {
                        this.nextRemoveTxTime = nextTxTime;
                    }
                }
            }
        } catch (GenericEntityException e) {
            try {
                TransactionUtil.rollback(beganTransaction, "Entity Engine error in assembleKeysToRemove", e);
            } catch (GenericTransactionException e2) {
                Debug.logWarning(e2, "Unable to call rollback()", module);
            }
            throw new SyncDataErrorException("Error getting keys to remove from the datasource", e);
        } catch (Throwable t) {
            try {
                TransactionUtil.rollback(beganTransaction, "General error in assembleKeysToRemove", t);
            } catch (GenericTransactionException e2) {
                Debug.logWarning(e2, "Unable to call rollback()", module);
            }
            throw new SyncDataErrorException("Caught runtime error while getting keys to remove", t);
        }

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

        // TEST SECTION: leave false for normal use
        boolean logValues = false;
        if (logValues && keysToRemove.size() > 0) {
            StringBuilder toRemoveInfo = new StringBuilder();
            for (GenericEntity keyToRemove: keysToRemove) {
                toRemoveInfo.append("\n-->[");
                toRemoveInfo.append(keyToRemove.get(ModelEntity.STAMP_TX_FIELD));
                toRemoveInfo.append(":");
                toRemoveInfo.append(keyToRemove.get(ModelEntity.STAMP_FIELD));
                toRemoveInfo.append("] ");
                toRemoveInfo.append(keyToRemove);
            }
            Debug.logInfo(toRemoveInfo.toString(), module);
        }

        // As this.nextRemoveTxTime calculation is only based on entities without keys to remove, if there at least one key to remove returned
        // this calculation is false, so it needs to be nullified
        if (keysToRemove.size() > 0) {
            this.nextRemoveTxTime = null;
        }
        
        return keysToRemove;
    }

    public void saveResultsReportedFromDataStore() throws SyncDataErrorException, SyncServiceErrorException {
        try {
            long runningTimeMillis = System.currentTimeMillis() - startDate.getTime();

            // get the total for this split
            long splitTotalTime = System.currentTimeMillis() - this.splitStartTime;
            if (splitTotalTime < this.perSplitMinMillis) {
                this.perSplitMinMillis = splitTotalTime;
            }
            if (splitTotalTime > this.perSplitMaxMillis) {
                this.perSplitMaxMillis = splitTotalTime;
            }

            // start the timer for the next split
            setSplitStartTime();

            // total the rows saved so far, and gather some info about them before saving
            this.totalRowsPerSplit = this.toCreateInserted + this.toCreateNotUpdated + this.toCreateUpdated +
                    this.toStoreInserted + this.toStoreNotUpdated + this.toStoreUpdated +
                    this.toRemoveAlreadyDeleted + this.toRemoveDeleted;
            if (this.totalRowsPerSplit < this.perSplitMinItems) {
                this.perSplitMinItems = this.totalRowsPerSplit;
            }
            if (this.totalRowsPerSplit > this.perSplitMaxItems) {
                this.perSplitMaxItems = this.totalRowsPerSplit;
            }
            this.totalRowsToCreate += this.toCreateInserted + this.toCreateNotUpdated + this.toCreateUpdated;
            this.totalRowsToStore += this.toStoreInserted + this.toStoreNotUpdated + this.toStoreUpdated;
            this.totalRowsToRemove += this.toRemoveAlreadyDeleted + this.toRemoveDeleted;

            // store latest result on EntitySync, ie update lastSuccessfulSynchTime, should run in own tx
            Map<String, Object> updateEsRunResult = dispatcher.runSync("updateEntitySyncRunning", UtilMisc.toMap("entitySyncId", entitySyncId, "lastSuccessfulSynchTime", this.currentRunEndTime, "userLogin", userLogin));

            // store result of service call on history with results so far, should run in own tx
            Map<String, Object> updateHistoryMap = UtilMisc.toMap("entitySyncId", entitySyncId, "startDate", startDate,
                    "lastSuccessfulSynchTime", this.currentRunEndTime, "lastCandidateEndTime", this.getNextRunEndTime(),
                    "lastSplitStartTime", Long.valueOf(this.splitStartTime));
            updateHistoryMap.put("toCreateInserted", Long.valueOf(toCreateInserted));
            updateHistoryMap.put("toCreateUpdated", Long.valueOf(toCreateUpdated));
            updateHistoryMap.put("toCreateNotUpdated", Long.valueOf(toCreateNotUpdated));
            updateHistoryMap.put("toStoreInserted", Long.valueOf(toStoreInserted));
            updateHistoryMap.put("toStoreUpdated", Long.valueOf(toStoreUpdated));
            updateHistoryMap.put("toStoreNotUpdated", Long.valueOf(toStoreNotUpdated));
            updateHistoryMap.put("toRemoveDeleted", Long.valueOf(toRemoveDeleted));
            updateHistoryMap.put("toRemoveAlreadyDeleted", Long.valueOf(toRemoveAlreadyDeleted));
            updateHistoryMap.put("runningTimeMillis", Long.valueOf(runningTimeMillis));
            updateHistoryMap.put("totalStoreCalls", Long.valueOf(totalStoreCalls));
            updateHistoryMap.put("totalSplits", Long.valueOf(totalSplits));
            updateHistoryMap.put("totalRowsExported", Long.valueOf(totalRowsExported));
            updateHistoryMap.put("totalRowsToCreate", Long.valueOf(totalRowsToCreate));
            updateHistoryMap.put("totalRowsToStore", Long.valueOf(totalRowsToStore));
            updateHistoryMap.put("totalRowsToRemove", Long.valueOf(totalRowsToRemove));
            updateHistoryMap.put("perSplitMinMillis", Long.valueOf(perSplitMinMillis));
            updateHistoryMap.put("perSplitMaxMillis", Long.valueOf(perSplitMaxMillis));
            updateHistoryMap.put("perSplitMinItems", Long.valueOf(perSplitMinItems));
            updateHistoryMap.put("perSplitMaxItems", Long.valueOf(perSplitMaxItems));
            updateHistoryMap.put("userLogin", userLogin);
            Map<String, Object> updateEsHistRunResult = dispatcher.runSync("updateEntitySyncHistory", updateHistoryMap);

            // now we have updated EntitySync and EntitySyncHistory, check both ops for errors...
            if (ServiceUtil.isError(updateEsRunResult)) {
                String errorMsg = "Error running EntitySync [" + entitySyncId + "], update of EntitySync record with lastSuccessfulSynchTime failed.";
                throw new SyncDataErrorException(errorMsg, null, null, updateEsRunResult, null);
            }

            if (ServiceUtil.isError(updateEsHistRunResult)) {
                String errorMsg = "Error running EntitySync [" + entitySyncId + "], update of EntitySyncHistory (startDate:[" + startDate + "]) record with lastSuccessfulSynchTime and result stats failed.";
                throw new SyncDataErrorException(errorMsg, null, null, updateEsHistRunResult, null);
            }
        } catch (GenericServiceException e) {
            throw new SyncServiceErrorException("Error saving results reported from data store", e);
        }
    }

    public void saveFinalSyncResults() throws SyncDataErrorException, SyncServiceErrorException {
        String newStatusId = "ESR_COMPLETE";
        if (this.isOfflineSync && totalRowsExported > 0) {
            newStatusId = "ESR_PENDING";
        }

        // the lastSuccessfulSynchTime on EntitySync will already be set, so just set status as completed
        String esErrMsg = "Could not mark Entity Sync as complete, but all synchronization was successful";
        try {
            Map<String, Object> completeEntitySyncRes = dispatcher.runSync("updateEntitySyncRunning", UtilMisc.toMap("entitySyncId", entitySyncId, "runStatusId", newStatusId, "userLogin", userLogin));
            if (ServiceUtil.isError(completeEntitySyncRes)) {
                // what to do here? try again?
                throw new SyncDataErrorException(esErrMsg, null, null, completeEntitySyncRes, null);
            }
        } catch (GenericServiceException e) {
            throw new SyncServiceErrorException(esErrMsg, e);
        }

        // if nothing moved over, remove the history record, otherwise store status
        long totalRows = totalRowsToCreate + totalRowsToStore + totalRowsToRemove;
        if (totalRows == 0) {
            String eshRemoveErrMsg = "Could not remove Entity Sync History (done because nothing was synced in this call), but all synchronization was successful";
            try {
                Map<String, Object> deleteEntitySyncHistRes = dispatcher.runSync("deleteEntitySyncHistory", UtilMisc.toMap("entitySyncId", entitySyncId, "startDate", startDate, "userLogin", userLogin));
                if (ServiceUtil.isError(deleteEntitySyncHistRes)) {
                    throw new SyncDataErrorException(eshRemoveErrMsg, null, null, deleteEntitySyncHistRes, null);
                }
            } catch (GenericServiceException e) {
                throw new SyncServiceErrorException(eshRemoveErrMsg, e);
            }
        } else {
            // the lastSuccessfulSynchTime on EntitySync will already be set, so just set status as completed
            String eshCompleteErrMsg = "Could not mark Entity Sync History as complete, but all synchronization was successful";
            try {
                Map<String, Object> completeEntitySyncHistRes = dispatcher.runSync("updateEntitySyncHistory", UtilMisc.toMap("entitySyncId", entitySyncId, "startDate", startDate, "runStatusId", "ESR_COMPLETE", "userLogin", userLogin));
                if (ServiceUtil.isError(completeEntitySyncHistRes)) {
                    // what to do here? try again?
                    throw new SyncDataErrorException(eshCompleteErrMsg, null, null, completeEntitySyncHistRes, null);
                }
            } catch (GenericServiceException e) {
                throw new SyncServiceErrorException(eshCompleteErrMsg, e);
            }
        }

        if (Debug.infoOn()) Debug.logInfo("Finished saveFinalSyncResults [" + entitySyncId + "]: totalRows=" + totalRows + ", totalRowsToCreate=" + totalRowsToCreate + ", totalRowsToStore=" + totalRowsToStore + ", totalRowsToRemove=" + totalRowsToRemove, module);
    }

    public Set<String> makeEntityNameToUseSet() {
        Set<String> entityNameToUseSet = new HashSet<String>();
        for (ModelEntity modelEntity: this.entityModelToUseList) {
            entityNameToUseSet.add(modelEntity.getEntityName());
        }
        return entityNameToUseSet;
    }

    /** prepare a list of all entities we want to synchronize: remove all view-entities and all entities that don't match the patterns attached to this EntitySync */
    protected List<ModelEntity> makeEntityModelToUseList() throws GenericEntityException {
        List<GenericValue> entitySyncIncludes = entitySync.getRelated("EntitySyncInclude", null, null, false);
        // get these ones as well, and just add them to the main list, it will have an extra field but that shouldn't hurt anything in the code below
        List<GenericValue> entitySyncGroupIncludes = entitySync.getRelated("EntitySyncInclGrpDetailView", null, null, false);
        entitySyncIncludes.addAll(entitySyncGroupIncludes);

        List<ModelEntity> entityModelToUseList = EntityGroupUtil.getModelEntitiesFromRecords(entitySyncIncludes, delegator, true);

        if (Debug.infoOn()) Debug.logInfo("In makeEntityModelToUseList for EntitySync with ID [" + entitySync.get("entitySyncId") + "] syncing " + entityModelToUseList.size() + " entities", module);
        return entityModelToUseList;
    }

    protected static Timestamp getCurrentRunStartTime(Timestamp lastSuccessfulSynchTime, List<ModelEntity> entityModelToUseList, Delegator delegator) throws GenericEntityException {
        // if currentRunStartTime is null, what to do? I guess iterate through all entities and find earliest tx stamp
        if (lastSuccessfulSynchTime == null) {
            Timestamp currentRunStartTime = null;
            for (ModelEntity modelEntity: entityModelToUseList) {
                // fields to select will be PK and the STAMP_TX_FIELD, slimmed down so we don't get a ton of data back
                Set<String> fieldsToSelect = UtilMisc.toSet(modelEntity.getPkFieldNames());
                // 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
                fieldsToSelect.add(ModelEntity.STAMP_TX_FIELD);
                EntityListIterator eli = EntityQuery.use(delegator)
                                                    .select(fieldsToSelect)
                                                    .from(modelEntity.getEntityName())
                                                    .where(EntityCondition.makeCondition(ModelEntity.STAMP_TX_FIELD, EntityOperator.NOT_EQUAL, null))
                                                    .orderBy(ModelEntity.STAMP_TX_FIELD)
                                                    .queryIterator();
                GenericValue nextValue = eli.next();
                eli.close();
                if (nextValue != null) {
                    Timestamp candidateTime = nextValue.getTimestamp(ModelEntity.STAMP_TX_FIELD);
                    if (currentRunStartTime == null || candidateTime.before(currentRunStartTime)) {
                        currentRunStartTime = candidateTime;
                    }
                }
            }
            if (Debug.infoOn()) Debug.logInfo("No currentRunStartTime was stored on the EntitySync record, so searched for the earliest value and got: " + currentRunStartTime, module);
            return currentRunStartTime;
        } else {
            return lastSuccessfulSynchTime;
        }
    }

    public void saveSyncErrorInfo(String runStatusId, List<Object> errorMessages) {
        // set error statuses on the EntitySync and EntitySyncHistory entities
        try {
            Map<String, Object> errorEntitySyncRes = dispatcher.runSync("updateEntitySyncRunning", UtilMisc.toMap("entitySyncId", entitySyncId, "runStatusId", runStatusId, "userLogin", userLogin));
            if (ServiceUtil.isError(errorEntitySyncRes)) {
                errorMessages.add("Could not save error run status [" + runStatusId + "] on EntitySync with ID [" + entitySyncId + "]: " + errorEntitySyncRes.get(ModelService.ERROR_MESSAGE));
            }
        } catch (GenericServiceException e) {
            errorMessages.add("Could not save error run status [" + runStatusId + "] on EntitySync with ID [" + entitySyncId + "]: " + e.toString());
        }
        if (startDate != null) {
            try {
                Map<String, Object> errorEntitySyncHistoryRes = dispatcher.runSync("updateEntitySyncHistory", UtilMisc.toMap("entitySyncId", entitySyncId, "startDate", startDate, "runStatusId", runStatusId, "userLogin", userLogin));
                if (ServiceUtil.isError(errorEntitySyncHistoryRes)) {
                    errorMessages.add("Could not save error run status [" + runStatusId + "] on EntitySyncHistory with ID [" + entitySyncId + "]: " + errorEntitySyncHistoryRes.get(ModelService.ERROR_MESSAGE));
                }
            } catch (GenericServiceException e) {
                errorMessages.add("Could not save error run status [" + runStatusId + "] on EntitySyncHistory with ID [" + entitySyncId + ":" + startDate + "]: " + e.toString());
            }
        }
    }

    // ======================== PUSH Methods ========================
    public void runPushStartRunning() throws SyncDataErrorException, SyncServiceErrorException, SyncAbortException {
        if (UtilValidate.isEmpty(targetServiceName)) {
            throw new SyncAbortException("Not running EntitySync [" + entitySyncId + "], no targetServiceName is specified, where do we send the data?");
        }

        // check to see if this sync is already running, if so return error
        if (this.isEntitySyncRunning()) {
            throw new SyncAbortException("Not running EntitySync [" + entitySyncId + "], an instance is already running.");
        }

        String markErrorMsg = "Could not start Entity Sync service, could not mark as running";
        try {
            // not running, get started NOW
            // set running status on entity sync, run in its own tx
            Map<String, Object> startEntitySyncRes = dispatcher.runSync("updateEntitySyncRunning", UtilMisc.toMap("entitySyncId", entitySyncId, "runStatusId", "ESR_RUNNING", "userLogin", userLogin));
            if (ModelService.RESPOND_ERROR.equals(startEntitySyncRes.get(ModelService.RESPONSE_MESSAGE))) {
                throw new SyncDataErrorException(markErrorMsg, null, null, startEntitySyncRes, null);
            }
        } catch (GenericServiceException e) {
            throw new SyncServiceErrorException(markErrorMsg, e);
        }

        // finally create the initial history record
        this.createInitialHistory();
    }

    public long setTotalRowCounts(ArrayList<GenericValue> valuesToCreate, ArrayList<GenericValue> valuesToStore, List<GenericEntity> keysToRemove) {
        this.totalRowsToCreate = valuesToCreate.size();
        this.totalRowsToStore = valuesToStore.size();
        this.totalRowsToRemove = keysToRemove.size();
        this.totalRowsPerSplit = this.totalRowsToCreate + this.totalRowsToStore + this.totalRowsToRemove;
        return this.totalRowsPerSplit;
    }

    public void runPushSendData(ArrayList<GenericValue> valuesToCreate, ArrayList<GenericValue> valuesToStore, List<GenericEntity> keysToRemove) throws SyncOtherErrorException, SyncServiceErrorException {
        // grab the totals for this data
        this.setTotalRowCounts(valuesToCreate, valuesToStore, keysToRemove);

        // call service named on EntitySync, IFF there is actually data to send over
        if (this.totalRowsPerSplit > 0) {
            Map<String, Object> targetServiceMap = UtilMisc.toMap("entitySyncId", entitySyncId, "valuesToCreate", valuesToCreate, "valuesToStore", valuesToStore, "keysToRemove", keysToRemove, "userLogin", userLogin);
            if (UtilValidate.isNotEmpty(targetDelegatorName)) {
                targetServiceMap.put("delegatorName", targetDelegatorName);
            }
            String serviceErrorMsg = "Error running EntitySync [" + entitySyncId + "], call to store service [" + targetServiceName + "] failed.";
            try {
                Map<String, Object> remoteStoreResult = dispatcher.runSync(targetServiceName, targetServiceMap);
                if (ServiceUtil.isError(remoteStoreResult)) {
                    throw new SyncOtherErrorException(serviceErrorMsg, null, null, remoteStoreResult, null);
                }

                this.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();

                this.toCreateInserted += toCreateInsertedCur;
                this.toCreateUpdated += toCreateUpdatedCur;
                this.toCreateNotUpdated += toCreateNotUpdatedCur;
                this.toStoreInserted += toStoreInsertedCur;
                this.toStoreUpdated += toStoreUpdatedCur;
                this.toStoreNotUpdated += toStoreNotUpdatedCur;
                this.toRemoveDeleted += toRemoveDeletedCur;
                this.toRemoveAlreadyDeleted += toRemoveAlreadyDeletedCur;
            } catch (GenericServiceException e) {
                throw new SyncServiceErrorException(serviceErrorMsg, e);
            }
        }
    }

    // ======================== PULL Methods ========================
    public void runPullStartOrRestoreSavedResults() throws SyncDataErrorException, SyncServiceErrorException, SyncAbortException {
        // if EntitySync.statusId is ESR_RUNNING, make sure startDate matches EntitySync.lastHistoryStartDate; or return error
        if (isEntitySyncRunning() && this.startDate == null) {
            throw new SyncAbortException("Not running EntitySync [" + entitySyncId + "], an instance is already running and no startDate for the current run was passed.");
        }

        if (this.startDate == null) {
            // get it started!
            String markErrorMsg = "Could not start Entity Sync service, could not mark as running";
            try {
                // not running, get started NOW
                // set running status on entity sync, run in its own tx
                Map<String, Object> startEntitySyncRes = dispatcher.runSync("updateEntitySyncRunning", UtilMisc.toMap("entitySyncId", entitySyncId, "runStatusId", "ESR_RUNNING", "userLogin", userLogin));
                if (ModelService.RESPOND_ERROR.equals(startEntitySyncRes.get(ModelService.RESPONSE_MESSAGE))) {
                    throw new SyncDataErrorException(markErrorMsg, null, null, startEntitySyncRes, null);
                }
            } catch (GenericServiceException e) {
                throw new SyncServiceErrorException(markErrorMsg, e);
            }

            // finally create the initial history record
            this.createInitialHistory();
            this.setSplitStartTime();
        } else {
            try {
                // set the latest values from the EntitySyncHistory, based on the values on the EntitySync
                GenericValue entitySyncHistory = EntityQuery.use(delegator).from("EntitySyncHistory").where("entitySyncId", entitySyncId, "startDate", startDate).queryOne();
                this.toCreateInserted = UtilMisc.toLong(entitySyncHistory.getLong("toCreateInserted"));
                this.toCreateUpdated = UtilMisc.toLong(entitySyncHistory.getLong("toCreateUpdated"));
                this.toCreateNotUpdated = UtilMisc.toLong(entitySyncHistory.getLong("toCreateNotUpdated"));

                this.toStoreInserted = UtilMisc.toLong(entitySyncHistory.getLong("toStoreInserted"));
                this.toStoreUpdated = UtilMisc.toLong(entitySyncHistory.getLong("toStoreUpdated"));
                this.toStoreNotUpdated = UtilMisc.toLong(entitySyncHistory.getLong("toStoreNotUpdated"));

                this.toRemoveDeleted = UtilMisc.toLong(entitySyncHistory.getLong("toRemoveDeleted"));
                this.toRemoveAlreadyDeleted = UtilMisc.toLong(entitySyncHistory.getLong("toRemoveAlreadyDeleted"));

                this.totalStoreCalls = UtilMisc.toLong(entitySyncHistory.getLong("totalStoreCalls"));
                this.totalSplits = UtilMisc.toLong(entitySyncHistory.getLong("totalSplits"));
                this.totalRowsToCreate = UtilMisc.toLong(entitySyncHistory.getLong("totalRowsToCreate"));
                this.totalRowsToStore = UtilMisc.toLong(entitySyncHistory.getLong("totalRowsToStore"));
                this.totalRowsToRemove = UtilMisc.toLong(entitySyncHistory.getLong("totalRowsToRemove"));

                this.perSplitMinMillis = UtilMisc.toLong(entitySyncHistory.getLong("perSplitMinMillis"));
                this.perSplitMaxMillis = UtilMisc.toLong(entitySyncHistory.getLong("perSplitMaxMillis"));
                this.perSplitMinItems = UtilMisc.toLong(entitySyncHistory.getLong("perSplitMinItems"));
                this.perSplitMaxItems = UtilMisc.toLong(entitySyncHistory.getLong("perSplitMaxItems"));

                this.splitStartTime = UtilMisc.toLong(entitySyncHistory.getLong("lastSplitStartTime"));
            } catch (GenericEntityException e) {
                throw new SyncDataErrorException("Error getting existing EntitySyncHistory values", e);
            }

            // got the previous values, now add to them with the values from the context...
            this.toCreateInserted += UtilMisc.toLong(this.context.get("toCreateInserted"));
            this.toCreateUpdated += UtilMisc.toLong(this.context.get("toCreateUpdated"));
            this.toCreateNotUpdated += UtilMisc.toLong(this.context.get("toCreateNotUpdated"));
            this.toStoreInserted += UtilMisc.toLong(this.context.get("toStoreInserted"));
            this.toStoreUpdated += UtilMisc.toLong(this.context.get("toStoreUpdated"));
            this.toStoreNotUpdated += UtilMisc.toLong(this.context.get("toStoreNotUpdated"));
            this.toRemoveDeleted += UtilMisc.toLong(this.context.get("toRemoveDeleted"));
            this.toRemoveAlreadyDeleted += UtilMisc.toLong(this.context.get("toRemoveAlreadyDeleted"));

            this.totalStoreCalls++;

            this.saveResultsReportedFromDataStore();
        }
    }

    // ======================== OFFLINE Methods ========================
    public void runOfflineStartRunning() throws SyncDataErrorException, SyncServiceErrorException, SyncAbortException {
        // check to see if this sync is already running, if so return error
        if (this.isEntitySyncRunning()) {
            throw new SyncAbortException("Not running EntitySync [" + entitySyncId + "], an instance is already running.");
        }

        // flag this context as offline
        this.isOfflineSync = true;

        String markErrorMsg = "Could not start Entity Sync service, could not mark as running";
        try {
            // not running, get started NOW
            // set running status on entity sync, run in its own tx
            Map<String, Object> startEntitySyncRes = dispatcher.runSync("updateEntitySyncRunning", UtilMisc.toMap("entitySyncId", entitySyncId, "runStatusId", "ESR_RUNNING", "preOfflineSynchTime", this.lastSuccessfulSynchTime, "userLogin", userLogin));
            if (ModelService.RESPOND_ERROR.equals(startEntitySyncRes.get(ModelService.RESPONSE_MESSAGE))) {
                throw new SyncDataErrorException(markErrorMsg, null, null, startEntitySyncRes, null);
            }
        } catch (GenericServiceException e) {
            throw new SyncServiceErrorException(markErrorMsg, e);
        }

        // finally create the initial history record
        this.createInitialHistory();
    }

    public void runSaveOfflineSyncInfo(long rowsInSplit) throws SyncDataErrorException, SyncServiceErrorException, SyncAbortException {
        this.totalRowsExported += rowsInSplit;
        this.saveResultsReportedFromDataStore();
    }

    /**
     * Static method to obtain a list of entity names which will be synchronized
     */
    public static Set<String> getEntitySyncModelNamesToUse(LocalDispatcher dispatcher, String entitySyncId) throws SyncDataErrorException, SyncAbortException {
        DispatchContext dctx = dispatcher.getDispatchContext();
        EntitySyncContext ctx = new EntitySyncContext(dctx, UtilMisc.toMap("entitySyncId", entitySyncId));
        return ctx.makeEntityNameToUseSet();
    }

    /** This class signifies an abort condition, so the state and such of the EntitySync value in the datasource should not be changed */
    @SuppressWarnings("serial")
    public static class SyncAbortException extends GeneralServiceException {
        public SyncAbortException() {
            super();
        }

        public SyncAbortException(String str) {
            super(str);
        }

        public SyncAbortException(String str, Throwable nested) {
            super(str, nested);
        }

        public SyncAbortException(Throwable nested) {
            super(nested);
        }

        public SyncAbortException(String str, List<Object> errorMsgList, Map<String, Object> errorMsgMap, Map<String, Object> nestedServiceResult, Throwable nested) {
            super(str, errorMsgList, errorMsgMap, nestedServiceResult, nested);
        }
    }

    @SuppressWarnings("serial")
    public static abstract class SyncErrorException extends GeneralServiceException {
        public SyncErrorException() { super(); }
        public SyncErrorException(String str) { super(str); }
        public SyncErrorException(String str, Throwable nested) { super(str, nested); }
        public SyncErrorException(Throwable nested) { super(nested); }
        public SyncErrorException(String str, List<Object> errorMsgList, Map<String, Object> errorMsgMap, Map<String, Object> nestedServiceResult, Throwable nested) { super(str, errorMsgList, errorMsgMap, nestedServiceResult, nested); }
        public abstract void saveSyncErrorInfo(EntitySyncContext esc);
    }

    /** This class signifies an error condition, so the state of the EntitySync value and the EntitySyncHistory value in the datasource should be changed to reflect the error */
    @SuppressWarnings("serial")
    public static class SyncOtherErrorException extends SyncErrorException {
        public SyncOtherErrorException() { super(); }
        public SyncOtherErrorException(String str) { super(str); }
        public SyncOtherErrorException(String str, Throwable nested) { super(str, nested); }
        public SyncOtherErrorException(Throwable nested) { super(nested); }
        public SyncOtherErrorException(String str, List<Object> errorMsgList, Map<String, Object> errorMsgMap, Map<String, Object> nestedServiceResult, Throwable nested) { super(str, errorMsgList, errorMsgMap, nestedServiceResult, nested); }
        @Override
        public void saveSyncErrorInfo(EntitySyncContext esc) {
            if (esc != null) {
                List<Object> errorList = new LinkedList<Object>();
                esc.saveSyncErrorInfo("ESR_OTHER_ERROR", errorList);
                this.addErrorMessages(errorList);
            }
        }
    }

    /** This class signifies an error condition, so the state of the EntitySync value and the EntitySyncHistory value in the datasource should be changed to reflect the error */
    @SuppressWarnings("serial")
    public static class SyncDataErrorException extends SyncErrorException {
        public SyncDataErrorException() { super(); }
        public SyncDataErrorException(String str) { super(str); }
        public SyncDataErrorException(String str, Throwable nested) { super(str, nested); }
        public SyncDataErrorException(Throwable nested) { super(nested); }
        public SyncDataErrorException(String str, List<Object> errorMsgList, Map<String, Object> errorMsgMap, Map<String, Object> nestedServiceResult, Throwable nested) { super(str, errorMsgList, errorMsgMap, nestedServiceResult, nested); }
        @Override
        public void saveSyncErrorInfo(EntitySyncContext esc) {
            if (esc != null) {
                List<Object> errorList = new LinkedList<Object>();
                esc.saveSyncErrorInfo("ESR_DATA_ERROR", errorList);
                this.addErrorMessages(errorList);
            }
        }
    }

    /** This class signifies an error condition, so the state of the EntitySync value and the EntitySyncHistory value in the datasource should be changed to reflect the error */
    @SuppressWarnings("serial")
    public static class SyncServiceErrorException extends SyncErrorException {
        public SyncServiceErrorException() { super(); }
        public SyncServiceErrorException(String str) { super(str); }
        public SyncServiceErrorException(String str, Throwable nested) { super(str, nested); }
        public SyncServiceErrorException(Throwable nested) { super(nested); }
        public SyncServiceErrorException(String str, List<Object> errorMsgList, Map<String, Object> errorMsgMap, Map<String, Object> nestedServiceResult, Throwable nested) { super(str, errorMsgList, errorMsgMap, nestedServiceResult, nested); }
        @Override
        public void saveSyncErrorInfo(EntitySyncContext esc) {
            if (esc != null) {
                List<Object> errorList = new LinkedList<Object>();
                esc.saveSyncErrorInfo("ESR_SERVICE_ERROR", errorList);
                this.addErrorMessages(errorList);
            }
        }
    }
}