Java > Open Source Codes > org > ofbiz > entityext > synchronization > EntitySyncContext


1 /*
2  * $Id: EntitySyncContext.java 5462 2005-08-05 18:35:48Z jonesde $
3  *
4  * Copyright (c) 2001-2005 The Open For Business Project - www.ofbiz.org
5  *
6  * Permission is hereby granted, free of charge, to any person obtaining a
7  * copy of this software and associated documentation files (the "Software"),
8  * to deal in the Software without restriction, including without limitation
9  * the rights to use, copy, modify, merge, publish, distribute, sublicense,
10  * and/or sell copies of the Software, and to permit persons to whom the
11  * Software is furnished to do so, subject to the following conditions:
12  *
13  * The above copyright notice and this permission notice shall be included
14  * in all copies or substantial portions of the Software.
15  *
16  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
17  * OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
18  * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
19  * IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
20  * CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT
21  * OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR
22  * THE USE OR OTHER DEALINGS IN THE SOFTWARE.
23  *
24  */
25 package org.ofbiz.entityext.synchronization;
26
27 import java.io.IOException ;
28 import java.sql.Timestamp ;
29 import java.util.ArrayList ;
30 import java.util.HashMap ;
31 import java.util.HashSet ;
32 import java.util.Iterator ;
33 import java.util.LinkedList ;
34 import java.util.List ;
35 import java.util.Map ;
36 import java.util.Set ;
37 import javax.xml.parsers.ParserConfigurationException ;
38
39 import org.ofbiz.base.util.Debug;
40 import org.ofbiz.base.util.UtilDateTime;
41 import org.ofbiz.base.util.UtilMisc;
42 import org.ofbiz.base.util.UtilValidate;
43 import org.ofbiz.entity.GenericDelegator;
44 import org.ofbiz.entity.GenericEntity;
45 import org.ofbiz.entity.GenericEntityException;
46 import org.ofbiz.entity.GenericValue;
47 import org.ofbiz.entity.condition.EntityCondition;
48 import org.ofbiz.entity.condition.EntityConditionList;
49 import org.ofbiz.entity.condition.EntityExpr;
50 import org.ofbiz.entity.condition.EntityOperator;
51 import org.ofbiz.entity.model.ModelEntity;
52 import org.ofbiz.entity.model.ModelViewEntity;
53 import org.ofbiz.entity.serialize.SerializeException;
54 import org.ofbiz.entity.serialize.XmlSerializer;
55 import org.ofbiz.entity.transaction.GenericTransactionException;
56 import org.ofbiz.entity.transaction.TransactionUtil;
57 import org.ofbiz.entity.util.EntityListIterator;
58 import org.ofbiz.service.DispatchContext;
59 import org.ofbiz.service.GeneralServiceException;
60 import org.ofbiz.service.GenericServiceException;
61 import org.ofbiz.service.LocalDispatcher;
62 import org.ofbiz.service.ModelService;
63 import org.ofbiz.service.ServiceUtil;
64
65 import org.xml.sax.SAXException ;
66
67 /**
68  * Entity Engine Sync Services
69  *
70  * @author <a HREF="mailto:jonesde@ofbiz.org">David E. Jones</a>
71  * @author <a HREF="mailto:jaz@ofbiz.org">Andy Zeneski</a>
72  * @version $Rev: 5462 $
73  * @since 3.0
74  */
75 public class EntitySyncContext {
76     
77     public static final String module = EntitySyncContext.class.getName();
78     
79     // set default split to 10 seconds, ie try not to get too much data moving over at once
80 public static final long defaultSyncSplitMillis = 10000;
81
82     // default offline split is 30 minutes
83 public static final long defaultOfflineSyncSplitMillis = 1800000;
84
85     // default to 5 minutes
86 public static final long defaultSyncEndBufferMillis = 300000;
87
88     // default to 2 hours, 120m, 7200s
89 public static final long defaultMaxRunningNoUpdateMillis = 7200000;
90
91     public GenericDelegator delegator;
92     public LocalDispatcher dispatcher;
93     public Map context;
94
95     public GenericValue userLogin;
96     public boolean isOfflineSync = false;
97
98     public String entitySyncId;
99     public GenericValue entitySync;
100
101     public String targetServiceName;
102     public String targetDelegatorName;
103     
104     public Timestamp syncEndStamp;
105     public long offlineSyncSplitMillis = defaultOfflineSyncSplitMillis;
106     public long syncSplitMillis = defaultSyncSplitMillis;
107     public long syncEndBufferMillis = defaultSyncEndBufferMillis;
108     public long maxRunningNoUpdateMillis = defaultMaxRunningNoUpdateMillis;
109     
110     public Timestamp lastSuccessfulSynchTime;
111     public List entityModelToUseList;
112     public Set entityNameToUseSet;
113     public Timestamp currentRunStartTime;
114     public Timestamp currentRunEndTime;
115
116     // these values are used to make this more efficient; if we run into an entity that has 0
117 //results for a given time block, we will do a query to find the next create/update/remove
118 //time for that entity, and also keep track of a global next with the lowest future next value;
119 //using these we can skip a lot of queries and speed this up significantly
120 public Map nextEntityCreateTxTime = new HashMap ();
121     public Map nextEntityUpdateTxTime = new HashMap ();
122     public Timestamp nextCreateTxTime = null;
123     public Timestamp nextUpdateTxTime = null;
124     public Timestamp nextRemoveTxTime = null;
125     
126     // this is the other part of the history PK, leave null until we create the history object
127 public Timestamp startDate = null;
128
129     long toCreateInserted = 0;
130     long toCreateUpdated = 0;
131     long toCreateNotUpdated = 0;
132     long toStoreInserted = 0;
133     long toStoreUpdated = 0;
134     long toStoreNotUpdated = 0;
135     long toRemoveDeleted = 0;
136     long toRemoveAlreadyDeleted = 0;
137
138     long totalRowsExported = 0;
139     long totalRowsToCreate = 0;
140     long totalRowsToStore = 0;
141     long totalRowsToRemove = 0;
142
143     long totalRowsPerSplit = 0;
144     long totalStoreCalls = 0;
145     long totalSplits = 0;
146     long perSplitMinMillis = Long.MAX_VALUE;
147     long perSplitMaxMillis = 0;
148     long perSplitMinItems = Long.MAX_VALUE;
149     long perSplitMaxItems = 0;
150     long splitStartTime = 0;
151
152     public EntitySyncContext(DispatchContext dctx, Map context) throws SyncDataErrorException, SyncAbortException {
153         this.context = context;
154         this.dispatcher = dctx.getDispatcher();
155
156         this.delegator = dctx.getDelegator();
157         // what to do with the delegatorName? this is the delegatorName to use in this service...
158 String delegatorName = (String ) context.get("delegatorName");
159         if (UtilValidate.isNotEmpty(delegatorName)) {
160             this.delegator = GenericDelegator.getGenericDelegator(delegatorName);
161         }
162
163
164         this.userLogin = (GenericValue) context.get("userLogin");
165         
166         this.entitySyncId = (String ) context.get("entitySyncId");
167         Debug.logInfo("Creating EntitySyncContext with entitySyncId=" + entitySyncId, module);
168
169         boolean beganTransaction = false;
170         try {
171             beganTransaction = TransactionUtil.begin(7200);
172         } catch (GenericTransactionException e) {
173             throw new SyncDataErrorException("Unable to begin JTA transaction", e);
174         }
175
176         try {
177             this.entitySync = delegator.findByPrimaryKey("EntitySync", UtilMisc.toMap("entitySyncId", this.entitySyncId));
178             if (this.entitySync == null) {
179                 throw new SyncAbortException("Not running EntitySync [" + entitySyncId + "], no record found with that ID.");
180             }
181
182             targetServiceName = entitySync.getString("targetServiceName");
183             targetDelegatorName = entitySync.getString("targetDelegatorName");
184             
185             // 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
186 //ahead of another machine writing to the DB it will still work fine and not lose any data
187 syncEndStamp = new Timestamp (System.currentTimeMillis() - syncEndBufferMillis);
188
189             this.offlineSyncSplitMillis = getOfflineSyncSplitMillis(entitySync);
190             this.syncSplitMillis = getSyncSplitMillis(entitySync);
191             this.syncEndBufferMillis = getSyncEndBufferMillis(entitySync);
192             this.maxRunningNoUpdateMillis = getMaxRunningNoUpdateMillis(entitySync);
193
194             this.lastSuccessfulSynchTime = entitySync.getTimestamp("lastSuccessfulSynchTime");
195             this.entityModelToUseList = this.makeEntityModelToUseList();
196             this.entityNameToUseSet = this.makeEntityNameToUseSet();
197             
198             // set start and end times for the first/current pass
199 this.currentRunStartTime = getCurrentRunStartTime(lastSuccessfulSynchTime, entityModelToUseList, delegator);
200             this.setCurrentRunEndTime();
201             
202             // this is mostly for the pull side... will always be null for at the beginning of a push process, to be filled in later
203 this.startDate = (Timestamp ) context.get("startDate");
204         } catch (GenericEntityException e) {
205             try {
206                 TransactionUtil.rollback(beganTransaction, "Entity Engine error while getting Entity Sync init information", e);
207             } catch (GenericTransactionException e2) {
208                 Debug.logWarning(e2, "Unable to call rollback()", module);
209             }
210             throw new SyncDataErrorException("Error initializing EntitySync Context", e);
211         }
212
213         try {
214             TransactionUtil.commit(beganTransaction);
215         } catch (GenericTransactionException e) {
216             throw new SyncDataErrorException("Unable to commit transaction", e);
217         }
218     }
219     
220     /**
221      * To see if it is running check:
222      * - in the running status
223      * - AND when the entitySync was last updated, and if it was more than maxRunningNoUpdateMillis ago, then don't consider it to be running
224      * @return boolean representing if the EntitySync should be considered running
225      */
226     public boolean isEntitySyncRunning() {
227         boolean isInRunning = ("ESR_RUNNING".equals(this.entitySync.getString("runStatusId")) ||
228                 "ESR_PENDING".equals(this.entitySync.getString("runStatusId")));
229
230         if (!isInRunning) {
231             return false;
232         }
233         
234         Timestamp esLastUpdated = this.entitySync.getTimestamp(ModelEntity.STAMP_FIELD);
235         if (esLastUpdated == null) {
236             // shouldn't ever happen, but just in case; assume is running if we don't know when it was last updated
237 return true;
238         }
239         long esLastUpdatedMillis = esLastUpdated.getTime();
240         long nowTimestampMillis = UtilDateTime.nowTimestamp().getTime();
241         long timeSinceUpdated = nowTimestampMillis - esLastUpdatedMillis;
242         if (timeSinceUpdated > this.maxRunningNoUpdateMillis) {
243             // it has been longer than the maxRunningNoUpdateMillis, so don't consider it running
244 return false;
245         }
246         
247         return true;
248     }
249     
250     public boolean hasMoreTimeToSync() {
251         return currentRunStartTime.before(syncEndStamp);
252     }
253     
254     protected void setCurrentRunEndTime() {
255         this.currentRunEndTime = getNextRunEndTime();
256     }
257     
258     protected Timestamp getNextRunEndTime() {
259         long syncSplit = this.isOfflineSync ? offlineSyncSplitMillis : syncSplitMillis;
260         Timestamp nextRunEndTime = new Timestamp (this.currentRunStartTime.getTime() + syncSplit);
261         if (nextRunEndTime.after(this.syncEndStamp)) {
262             nextRunEndTime = this.syncEndStamp;
263         }
264         return nextRunEndTime;
265     }
266     
267     public void advanceRunTimes() {
268         this.currentRunStartTime = this.currentRunEndTime;
269         this.setCurrentRunEndTime();
270     }
271
272     public void setSplitStartTime() {
273         this.splitStartTime = System.currentTimeMillis();
274     }
275     
276     protected static long getSyncSplitMillis(GenericValue entitySync) {
277         long splitMillis = defaultSyncSplitMillis;
278         Long syncSplitMillis = entitySync.getLong("syncSplitMillis");
279         if (syncSplitMillis != null) {
280             splitMillis = syncSplitMillis.longValue();
281         }
282         return splitMillis;
283     }
284
285     protected static long getOfflineSyncSplitMillis(GenericValue entitySync) {
286         long splitMillis = defaultOfflineSyncSplitMillis;
287         Long syncSplitMillis = entitySync.getLong("offlineSyncSplitMillis");
288         if (syncSplitMillis != null) {
289             splitMillis = syncSplitMillis.longValue();
290         }
291         return splitMillis;
292     }
293
294     protected static long getSyncEndBufferMillis(GenericValue entitySync) {
295         long syncEndBufferMillis = defaultSyncEndBufferMillis;
296         Long syncEndBufferMillisLong = entitySync.getLong("syncEndBufferMillis");
297         if (syncEndBufferMillisLong != null) {
298             syncEndBufferMillis = syncEndBufferMillisLong.longValue();
299         }
300         return syncEndBufferMillis;
301     }
302     
303     protected static long getMaxRunningNoUpdateMillis(GenericValue entitySync) {
304         long maxRunningNoUpdateMillis = defaultMaxRunningNoUpdateMillis;
305         Long maxRunningNoUpdateMillisLong = entitySync.getLong("maxRunningNoUpdateMillis");
306         if (maxRunningNoUpdateMillisLong != null) {
307             maxRunningNoUpdateMillis = maxRunningNoUpdateMillisLong.longValue();
308         }
309         return maxRunningNoUpdateMillis;
310     }
311     
312     /** create history record, target service should run in own tx */
313     public void createInitialHistory() throws SyncDataErrorException, SyncServiceErrorException {
314         String errorMsg = "Not running EntitySync [" + entitySyncId + "], could not create EntitySyncHistory";
315         try {
316             Map initialHistoryRes = dispatcher.runSync("createEntitySyncHistory", UtilMisc.toMap("entitySyncId", entitySyncId, "runStatusId", "ESR_RUNNING", "beginningSynchTime", this.currentRunStartTime, "lastCandidateEndTime", this.currentRunEndTime, "userLogin", userLogin));
317             if (ServiceUtil.isError(initialHistoryRes)) {
318                 throw new SyncDataErrorException(errorMsg, null, null, initialHistoryRes, null);
319             }
320             this.startDate = (Timestamp ) initialHistoryRes.get("startDate");
321         } catch (GenericServiceException e) {
322             throw new SyncServiceErrorException(errorMsg, e);
323         }
324     }
325     
326     public ArrayList assembleValuesToCreate() throws SyncDataErrorException {
327         // first grab all values inserted in the date range, then get the updates (leaving out all values inserted in the data range)
328 ArrayList valuesToCreate = new ArrayList (); // make it an ArrayList to easily merge in sorted lists
329
330         if (this.nextCreateTxTime != null && (this.nextCreateTxTime.equals(currentRunEndTime) || this.nextCreateTxTime.after(currentRunEndTime))) {
331             // 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...
332 return valuesToCreate;
333         }
334
335         //Debug.logInfo("Getting values to create; currentRunStartTime=" + currentRunStartTime + ", currentRunEndTime=" + currentRunEndTime, module);
336
337         int entitiesSkippedForKnownNext = 0;
338         
339         // iterate through entities, get all records with tx stamp in the current time range, put all in a single list
340 Iterator entityModelToUseCreateIter = entityModelToUseList.iterator();
341         while (entityModelToUseCreateIter.hasNext()) {
342             int insertBefore = 0;
343             ModelEntity modelEntity = (ModelEntity) entityModelToUseCreateIter.next();
344
345             // first test to see if we know that there are no records for this entity in this time period...
346 Timestamp knownNextCreateTime = (Timestamp ) this.nextEntityCreateTxTime.get(modelEntity.getEntityName());
347             if (knownNextCreateTime != null && (knownNextCreateTime.equals(currentRunEndTime) || knownNextCreateTime.after(currentRunEndTime))) {
348                 //Debug.logInfo("In assembleValuesToCreate found knownNextCreateTime [" + knownNextCreateTime + "] after currentRunEndTime [" + currentRunEndTime + "], so skipping time per period for entity [" + modelEntity.getEntityName() + "]", module);
349 entitiesSkippedForKnownNext++;
350                 continue;
351             }
352
353             boolean beganTransaction = false;
354             try {
355                 beganTransaction = TransactionUtil.begin(7200);
356             } catch (GenericTransactionException e) {
357                 throw new SyncDataErrorException("Unable to begin JTA transaction", e);
358             }
359
360             try {
361                 // get the values created within the current time range
362 EntityCondition findValCondition = new EntityConditionList(UtilMisc.toList(
363                         new EntityExpr(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.GREATER_THAN_EQUAL_TO, currentRunStartTime),
364                         new EntityExpr(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.LESS_THAN, currentRunEndTime)), EntityOperator.AND);
365                 EntityListIterator eli = delegator.findListIteratorByCondition(modelEntity.getEntityName(), findValCondition, null, UtilMisc.toList(ModelEntity.CREATE_STAMP_TX_FIELD, ModelEntity.CREATE_STAMP_FIELD));
366                 GenericValue nextValue = null;
367                 long valuesPerEntity = 0;
368                 while ((nextValue = (GenericValue) eli.next()) != null) {
369                     // sort by the tx stamp and then the record stamp
370 // 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
371 while (insertBefore < valuesToCreate.size() && ((GenericValue) valuesToCreate.get(insertBefore)).getTimestamp(ModelEntity.CREATE_STAMP_TX_FIELD).before(nextValue.getTimestamp(ModelEntity.CREATE_STAMP_TX_FIELD))) {
372                         insertBefore++;
373                     }
374                     while (insertBefore < valuesToCreate.size() && ((GenericValue) valuesToCreate.get(insertBefore)).getTimestamp(ModelEntity.CREATE_STAMP_FIELD).before(nextValue.getTimestamp(ModelEntity.CREATE_STAMP_FIELD))) {
375                         insertBefore++;
376                     }
377                     valuesToCreate.add(insertBefore, nextValue);
378                     valuesPerEntity++;
379                 }
380                 eli.close();
381                 
382                 // definately remove this message and related data gathering
383 //long preCount = delegator.findCountByCondition(modelEntity.getEntityName(), findValCondition, null);
384 //long entityTotalCount = delegator.findCountByCondition(modelEntity.getEntityName(), null, null);
385 //if (entityTotalCount > 0 || preCount > 0 || valuesPerEntity > 0) Debug.logInfo("Got " + valuesPerEntity + "/" + preCount + "/" + entityTotalCount + " values for entity " + modelEntity.getEntityName(), module);
386
387                 // if we didn't find anything for this entity, find the next value's Timestamp and keep track of it
388 if (valuesPerEntity == 0) {
389                     Timestamp startCheckStamp = new Timestamp (System.currentTimeMillis() - syncEndBufferMillis);
390                     
391                     EntityCondition findNextCondition = new EntityConditionList(UtilMisc.toList(
392                             new EntityExpr(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.NOT_EQUAL, null),
393                             new EntityExpr(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.GREATER_THAN_EQUAL_TO, currentRunEndTime)),
394                             EntityOperator.AND);
395                     EntityListIterator eliNext = delegator.findListIteratorByCondition(modelEntity.getEntityName(), findNextCondition, null, UtilMisc.toList(ModelEntity.CREATE_STAMP_TX_FIELD));
396                     // get the first element and it's tx time value...
397 GenericValue firstVal = (GenericValue) eliNext.next();
398                     eliNext.close();
399                     Timestamp nextTxTime;
400                     if (firstVal != null) {
401                         nextTxTime = firstVal.getTimestamp(ModelEntity.CREATE_STAMP_TX_FIELD);
402                     } else {
403                         // no results? well, then it's safe to say that up to the pre-querytime (minus the buffer, as usual) we are okay
404 nextTxTime = startCheckStamp;
405                     }
406                     if (this.nextCreateTxTime == null || nextTxTime.before(this.nextCreateTxTime)) {
407                         this.nextCreateTxTime = nextTxTime;
408                         Debug.logInfo("EntitySync: Set nextCreateTxTime to [" + nextTxTime + "]", module);
409                     }
410                     Timestamp curEntityNextTxTime = (Timestamp ) this.nextEntityCreateTxTime.get(modelEntity.getEntityName());
411                     if (curEntityNextTxTime == null || nextTxTime.before(curEntityNextTxTime)) {
412                         this.nextEntityCreateTxTime.put(modelEntity.getEntityName(), nextTxTime);
413                         Debug.logInfo("EntitySync: Set nextEntityCreateTxTime to [" + nextTxTime + "] for the entity [" + modelEntity.getEntityName() + "]", module);
414                     }
415                 }
416             } catch (GenericEntityException e) {
417                 try {
418                     TransactionUtil.rollback(beganTransaction, "Entity Engine error in assembleValuesToCreate", e);
419                     
420                 } catch (GenericTransactionException e2) {
421                     Debug.logWarning(e2, "Unable to call rollback()", module);
422                 }
423                 throw new SyncDataErrorException("Error getting values to create from the datasource", e);
424             } catch (Throwable t) {
425                 try {
426                     TransactionUtil.rollback(beganTransaction, "Throwable error in assembleValuesToCreate", t);
427                 } catch (GenericTransactionException e2) {
428                     Debug.logWarning(e2, "Unable to call rollback()", module);
429                 }
430                 throw new SyncDataErrorException("Caught runtime error while getting values to create", t);
431             }
432
433             try {
434                 TransactionUtil.commit(beganTransaction);
435             } catch (GenericTransactionException e) {
436                 throw new SyncDataErrorException("Commit transaction failed", e);
437             }
438         }
439
440         if (entitiesSkippedForKnownNext > 0) {
441             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);
442         }
443         
444         // TEST SECTION: leave false for normal use
445 boolean logValues = false;
446         if (logValues && valuesToCreate.size() > 0) {
447             StringBuffer toCreateInfo = new StringBuffer ();
448             Iterator valuesToCreateIter = valuesToCreate.iterator();
449             while (valuesToCreateIter.hasNext()) {
450                 GenericValue valueToCreate = (GenericValue) valuesToCreateIter.next();
451                 toCreateInfo.append("\n-->[");
452                 toCreateInfo.append(valueToCreate.get(ModelEntity.CREATE_STAMP_TX_FIELD));
453                 toCreateInfo.append(":");
454                 toCreateInfo.append(valueToCreate.get(ModelEntity.CREATE_STAMP_FIELD));
455                 toCreateInfo.append("] ");
456                 toCreateInfo.append(valueToCreate.getPrimaryKey());
457             }
458             Debug.logInfo(toCreateInfo.toString(), module);
459         }
460         
461         return valuesToCreate;
462     }
463
464     public ArrayList assembleValuesToStore() throws SyncDataErrorException {
465         // simulate two ordered lists and merge them on-the-fly for faster combined sorting
466 ArrayList valuesToStore = new ArrayList (); // make it an ArrayList to easily merge in sorted lists
467
468         if (this.nextUpdateTxTime != null && (this.nextUpdateTxTime.equals(currentRunEndTime) || this.nextUpdateTxTime.after(currentRunEndTime))) {
469             // 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...
470 return valuesToStore;
471         }
472
473         // Debug.logInfo("Getting values to store; currentRunStartTime=" + currentRunStartTime + ", currentRunEndTime=" + currentRunEndTime, module);
474
475         int entitiesSkippedForKnownNext = 0;
476         
477         // iterate through entities, get all records with tx stamp in the current time range, put all in a single list
478 Iterator entityModelToUseUpdateIter = entityModelToUseList.iterator();
479         while (entityModelToUseUpdateIter.hasNext()) {
480             int insertBefore = 0;
481             ModelEntity modelEntity = (ModelEntity) entityModelToUseUpdateIter.next();
482             
483             // first test to see if we know that there are no records for this entity in this time period...
484 Timestamp knownNextUpdateTime = (Timestamp ) this.nextEntityUpdateTxTime.get(modelEntity.getEntityName());
485             if (knownNextUpdateTime != null && (knownNextUpdateTime.equals(currentRunEndTime) || knownNextUpdateTime.after(currentRunEndTime))) {
486                 entitiesSkippedForKnownNext++;
487                 continue;
488             }
489
490             boolean beganTransaction = false;
491             try {
492                 beganTransaction = TransactionUtil.begin(7200);
493             } catch (GenericTransactionException e) {
494                 throw new SyncDataErrorException("Unable to begin JTA transaction", e);
495             }
496
497             try {
498                 // 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
499 EntityCondition createdBeforeStartCond = new EntityExpr(
500                         new EntityExpr(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.EQUALS, null),
501                         EntityOperator.OR,
502                         new EntityExpr(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.LESS_THAN, currentRunStartTime));
503                 EntityCondition findValCondition = new EntityConditionList(UtilMisc.toList(
504                         new EntityExpr(ModelEntity.STAMP_TX_FIELD, EntityOperator.GREATER_THAN_EQUAL_TO, currentRunStartTime),
505                         new EntityExpr(ModelEntity.STAMP_TX_FIELD, EntityOperator.LESS_THAN, currentRunEndTime),
506                         createdBeforeStartCond),
507                         EntityOperator.AND);
508                 EntityListIterator eli = delegator.findListIteratorByCondition(modelEntity.getEntityName(), findValCondition, null, UtilMisc.toList(ModelEntity.STAMP_TX_FIELD, ModelEntity.STAMP_FIELD));
509                 GenericValue nextValue = null;
510                 long valuesPerEntity = 0;
511                 while ((nextValue = (GenericValue) eli.next()) != null) {
512                     // sort by the tx stamp and then the record stamp
513 // 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
514 while (insertBefore < valuesToStore.size() && ((GenericValue) valuesToStore.get(insertBefore)).getTimestamp(ModelEntity.STAMP_TX_FIELD).before(nextValue.getTimestamp(ModelEntity.STAMP_TX_FIELD))) {
515                         insertBefore++;
516                     }
517                     while (insertBefore < valuesToStore.size() && ((GenericValue) valuesToStore.get(insertBefore)).getTimestamp(ModelEntity.STAMP_FIELD).before(nextValue.getTimestamp(ModelEntity.STAMP_FIELD))) {
518                         insertBefore++;
519                     }
520                     valuesToStore.add(insertBefore, nextValue);
521                     valuesPerEntity++;
522                 }
523                 eli.close();
524                 
525                 // definately remove this message and related data gathering
526 //long preCount = delegator.findCountByCondition(modelEntity.getEntityName(), findValCondition, null);
527 //long entityTotalCount = delegator.findCountByCondition(modelEntity.getEntityName(), null, null);
528 //if (entityTotalCount > 0 || preCount > 0 || valuesPerEntity > 0) Debug.logInfo("Got " + valuesPerEntity + "/" + preCount + "/" + entityTotalCount + " values for entity " + modelEntity.getEntityName(), module);
529
530                 // if we didn't find anything for this entity, find the next value's Timestamp and keep track of it
531 if (valuesPerEntity == 0) {
532                     Timestamp startCheckStamp = new Timestamp (System.currentTimeMillis() - syncEndBufferMillis);
533                     
534                     EntityCondition findNextCondition = new EntityConditionList(UtilMisc.toList(
535                             new EntityExpr(ModelEntity.STAMP_TX_FIELD, EntityOperator.NOT_EQUAL, null),
536                             new EntityExpr(ModelEntity.STAMP_TX_FIELD, EntityOperator.GREATER_THAN_EQUAL_TO, currentRunEndTime),
537                             new EntityExpr(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.NOT_EQUAL, null),
538                             new EntityExpr(ModelEntity.CREATE_STAMP_TX_FIELD, EntityOperator.LESS_THAN, currentRunEndTime)),
539                             EntityOperator.AND);
540                     EntityListIterator eliNext = delegator.findListIteratorByCondition(modelEntity.getEntityName(), findNextCondition, null, UtilMisc.toList(ModelEntity.STAMP_TX_FIELD));
541                     // get the first element and it's tx time value...
542 GenericValue firstVal = (GenericValue) eliNext.next();
543                     eliNext.close();
544                     Timestamp nextTxTime;
545                     if (firstVal != null) {
546                         nextTxTime = firstVal.getTimestamp(ModelEntity.CREATE_STAMP_TX_FIELD);
547                     } else {
548                         // no results? well, then it's safe to say that up to the pre-querytime (minus the buffer, as usual) we are okay
549 nextTxTime = startCheckStamp;
550                     }
551                     if (this.nextUpdateTxTime == null || nextTxTime.before(this.nextUpdateTxTime)) {
552                         this.nextUpdateTxTime = nextTxTime;
553                         Debug.logInfo("EntitySync: Set nextUpdateTxTime to [" + nextTxTime + "]", module);
554                     }
555                     Timestamp curEntityNextTxTime = (Timestamp ) this.nextEntityUpdateTxTime.get(modelEntity.getEntityName());
556                     if (curEntityNextTxTime == null || nextTxTime.before(curEntityNextTxTime)) {
557                         this.nextEntityUpdateTxTime.put(modelEntity.getEntityName(), nextTxTime);
558                         Debug.logInfo("EntitySync: Set nextEntityUpdateTxTime to [" + nextTxTime + "] for the entity [" + modelEntity.getEntityName() + "]", module);
559                     }
560                 }
561             } catch (GenericEntityException e) {
562                 try {
563                     TransactionUtil.rollback(beganTransaction, "Entity Engine error in assembleValuesToStore", e);
564                 } catch (GenericTransactionException e2) {
565                     Debug.logWarning(e2, "Unable to call rollback()", module);
566                 }
567                 throw new SyncDataErrorException("Error getting values to store from the datasource", e);
568             } catch (Throwable t) {
569                 try {
570                     TransactionUtil.rollback(beganTransaction, "General error in assembleValuesToStore", t);
571                 } catch (GenericTransactionException e2) {
572                     Debug.logWarning(e2, "Unable to call rollback()", module);
573                 }
574                 throw new SyncDataErrorException("Caught runtime error while getting values to store", t);
575             }
576
577             try {
578                 TransactionUtil.commit(beganTransaction);
579             } catch (GenericTransactionException e) {
580                 throw new SyncDataErrorException("Commit transaction failed", e);
581             }
582         }
583
584         if (entitiesSkippedForKnownNext > 0) {
585             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);
586         }
587         
588         // TEST SECTION: leave false for normal use
589 boolean logValues = false;
590         if (logValues && valuesToStore.size() > 0) {
591             StringBuffer toStoreInfo = new StringBuffer ();
592             Iterator valuesToStoreIter = valuesToStore.iterator();
593             while (valuesToStoreIter.hasNext()) {
594                 GenericValue valueToStore = (GenericValue) valuesToStoreIter.next();
595                 toStoreInfo.append("\n-->[");
596                 toStoreInfo.append(valueToStore.get(ModelEntity.STAMP_TX_FIELD));
597                 toStoreInfo.append(":");
598                 toStoreInfo.append(valueToStore.get(ModelEntity.STAMP_FIELD));
599                 toStoreInfo.append("] ");
600                 toStoreInfo.append(valueToStore.getPrimaryKey());
601             }
602             Debug.logInfo(toStoreInfo.toString(), module);
603         }
604         
605         return valuesToStore;
606     }
607
608     public LinkedList assembleKeysToRemove() throws SyncDataErrorException {
609         // get all removed items from the given time range, add to list for those
610 LinkedList keysToRemove = new LinkedList ();
611
612         if (this.nextRemoveTxTime != null && (this.nextRemoveTxTime.equals(currentRunEndTime) || this.nextRemoveTxTime.after(currentRunEndTime))) {
613             // 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...
614 return keysToRemove;
615         }
616
617         //Debug.logInfo("Getting keys to remove; currentRunStartTime=" + currentRunStartTime + ", currentRunEndTime=" + currentRunEndTime, module);
618
619         boolean beganTransaction = false;
620         try {
621             beganTransaction = TransactionUtil.begin(7200);
622         } catch (GenericTransactionException e) {
623             throw new SyncDataErrorException("Unable to begin JTA transaction", e);
624         }
625
626         try {
627             // 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
628 EntityCondition findValCondition = new EntityConditionList(UtilMisc.toList(
629                     new EntityExpr(ModelEntity.STAMP_TX_FIELD, EntityOperator.GREATER_THAN_EQUAL_TO, currentRunStartTime),
630                     new EntityExpr(ModelEntity.STAMP_TX_FIELD, EntityOperator.LESS_THAN, currentRunEndTime)), EntityOperator.AND);
631             EntityListIterator removeEli = delegator.findListIteratorByCondition("EntitySyncRemove", findValCondition, null, UtilMisc.toList(ModelEntity.STAMP_TX_FIELD, ModelEntity.STAMP_FIELD));
632             GenericValue entitySyncRemove = null;
633             while ((entitySyncRemove = (GenericValue) removeEli.next()) != null) {
634                 // pull the PK from the EntitySyncRemove in the primaryKeyRemoved field, de-XML-serialize it
635 String primaryKeyRemoved = entitySyncRemove.getString("primaryKeyRemoved");
636                 GenericEntity pkToRemove = null;
637                 try {
638                     pkToRemove = (GenericEntity) XmlSerializer.deserialize(primaryKeyRemoved, delegator);
639                 } catch (IOException e) {
640                     String errorMsg = "Error deserializing GenericPK to remove in Entity Sync Data for entitySyncId [" + entitySyncId + "] and entitySyncRemoveId [" + entitySyncRemove.getString("entitySyncRemoveId") + "]: " + e.toString();
641                     Debug.logError(e, errorMsg, module);
642                     throw new SyncDataErrorException(errorMsg, e);
643                 } catch (SAXException e) {
644                     String errorMsg = "Error deserializing GenericPK to remove in Entity Sync Data for entitySyncId [" + entitySyncId + "] and entitySyncRemoveId [" + entitySyncRemove.getString("entitySyncRemoveId") + "]: " + e.toString();
645                     Debug.logError(e, errorMsg, module);
646                     throw new SyncDataErrorException(errorMsg, e);
647                 } catch (ParserConfigurationException e) {
648                     String errorMsg = "Error deserializing GenericPK to remove in Entity Sync Data for entitySyncId [" + entitySyncId + "] and entitySyncRemoveId [" + entitySyncRemove.getString("entitySyncRemoveId") + "]: " + e.toString();
649                     Debug.logError(e, errorMsg, module);
650                     throw new SyncDataErrorException(errorMsg, e);
651                 } catch (SerializeException e) {
652                     String errorMsg = "Error deserializing GenericPK to remove in Entity Sync Data for entitySyncId [" + entitySyncId + "] and entitySyncRemoveId [" + entitySyncRemove.getString("entitySyncRemoveId") + "]: " + e.toString();
653                     Debug.logError(e, errorMsg, module);
654                     throw new SyncDataErrorException(errorMsg, e);
655                 }
656                 
657                 // set the stamp fields for future reference
658 pkToRemove.set(ModelEntity.STAMP_TX_FIELD, entitySyncRemove.get(ModelEntity.STAMP_TX_FIELD));
659                 pkToRemove.set(ModelEntity.STAMP_FIELD, entitySyncRemove.get(ModelEntity.STAMP_FIELD));
660                 pkToRemove.set(ModelEntity.CREATE_STAMP_TX_FIELD, entitySyncRemove.get(ModelEntity.CREATE_STAMP_TX_FIELD));
661                 pkToRemove.set(ModelEntity.CREATE_STAMP_FIELD, entitySyncRemove.get(ModelEntity.CREATE_STAMP_FIELD));
662
663                 if (this.entityNameToUseSet.contains(pkToRemove.getEntityName())) {
664                     keysToRemove.add(pkToRemove);
665                 }
666             }
667             removeEli.close();
668
669             // if we didn't find anything for this entity, find the next value's Timestamp and keep track of it
670 if (keysToRemove.size() == 0) {
671                 EntityCondition findNextCondition = new EntityExpr(ModelEntity.STAMP_TX_FIELD, EntityOperator.GREATER_THAN_EQUAL_TO, currentRunEndTime);
672                 EntityListIterator eliNext = delegator.findListIteratorByCondition("EntitySyncRemove", findNextCondition, null, UtilMisc.toList(ModelEntity.STAMP_TX_FIELD));
673                 // get the first element and it's tx time value...
674 GenericValue firstVal = (GenericValue) eliNext.next();
675                 eliNext.close();
676                 if (firstVal != null) {
677                     Timestamp nextTxTime = firstVal.getTimestamp(ModelEntity.STAMP_TX_FIELD);
678                     if (this.nextUpdateTxTime == null || nextTxTime.before(this.nextUpdateTxTime)) {
679                         this.nextUpdateTxTime = nextTxTime;
680                     }
681                 }
682             }
683         } catch (GenericEntityException e) {
684             try {
685                 TransactionUtil.rollback(beganTransaction, "Entity Engine error in assembleKeysToRemove", e);
686             } catch (GenericTransactionException e2) {
687                 Debug.logWarning(e2, "Unable to call rollback()", module);
688             }
689             throw new SyncDataErrorException("Error getting keys to remove from the datasource", e);
690         } catch (Throwable t) {
691             try {
692                 TransactionUtil.rollback(beganTransaction, "General error in assembleKeysToRemove", t);
693             } catch (GenericTransactionException e2) {
694                 Debug.logWarning(e2, "Unable to call rollback()", module);
695             }
696             throw new SyncDataErrorException("Caught runtime error while getting keys to remove", t);
697         }
698
699         try {
700             TransactionUtil.commit(beganTransaction);
701         } catch (GenericTransactionException e) {
702             throw new SyncDataErrorException("Commit transaction failed", e);
703         }
704
705         // TEST SECTION: leave false for normal use
706 boolean logValues = false;
707         if (logValues && keysToRemove.size() > 0) {
708             StringBuffer toRemoveInfo = new StringBuffer ();
709             Iterator keysToRemoveIter = keysToRemove.iterator();
710             while (keysToRemoveIter.hasNext()) {
711                 GenericEntity keyToRemove = (GenericEntity) keysToRemoveIter.next();
712                 toRemoveInfo.append("\n-->[");
713                 toRemoveInfo.append(keyToRemove.get(ModelEntity.STAMP_TX_FIELD));
714                 toRemoveInfo.append(":");
715                 toRemoveInfo.append(keyToRemove.get(ModelEntity.STAMP_FIELD));
716                 toRemoveInfo.append("] ");
717                 toRemoveInfo.append(keyToRemove);
718             }
719             Debug.logInfo(toRemoveInfo.toString(), module);
720         }
721         
722         return keysToRemove;
723     }
724     
725     public void saveResultsReportedFromDataStore() throws SyncDataErrorException, SyncServiceErrorException {
726         try {
727             long runningTimeMillis = System.currentTimeMillis() - startDate.getTime();
728
729             // get the total for this split
730 long splitTotalTime = System.currentTimeMillis() - this.splitStartTime;
731             if (splitTotalTime < this.perSplitMinMillis) {
732                 this.perSplitMinMillis = splitTotalTime;
733             }
734             if (splitTotalTime > this.perSplitMaxMillis) {
735                 this.perSplitMaxMillis = splitTotalTime;
736             }
737             
738             // start the timer for the next split
739 setSplitStartTime();
740
741             // total the rows saved so far, and gather some info about them before saving
742 this.totalRowsPerSplit = this.toCreateInserted + this.toCreateNotUpdated + this.toCreateUpdated +
743                     this.toStoreInserted + this.toStoreNotUpdated + this.toStoreUpdated +
744                     this.toRemoveAlreadyDeleted + this.toRemoveDeleted;
745             if (this.totalRowsPerSplit < this.perSplitMinItems) {
746                 this.perSplitMinItems = this.totalRowsPerSplit;
747             }
748             if (this.totalRowsPerSplit > this.perSplitMaxItems) {
749                 this.perSplitMaxItems = this.totalRowsPerSplit;
750             }
751             this.totalRowsToCreate += this.toCreateInserted + this.toCreateNotUpdated + this.toCreateUpdated;
752             this.totalRowsToStore += this.toStoreInserted + this.toStoreNotUpdated + this.toStoreUpdated;
753             this.totalRowsToRemove += this.toRemoveAlreadyDeleted + this.toRemoveDeleted;
754
755             // store latest result on EntitySync, ie update lastSuccessfulSynchTime, should run in own tx
756 Map updateEsRunResult = dispatcher.runSync("updateEntitySyncRunning", UtilMisc.toMap("entitySyncId", entitySyncId, "lastSuccessfulSynchTime", this.currentRunEndTime, "userLogin", userLogin));
757
758             // store result of service call on history with results so far, should run in own tx
759 Map updateHistoryMap = UtilMisc.toMap("entitySyncId", entitySyncId, "startDate", startDate,
760                     "lastSuccessfulSynchTime", this.currentRunEndTime, "lastCandidateEndTime", this.getNextRunEndTime(),
761                     "lastSplitStartTime", new Long (this.splitStartTime));
762             updateHistoryMap.put("toCreateInserted", new Long (toCreateInserted));
763             updateHistoryMap.put("toCreateUpdated", new Long (toCreateUpdated));
764             updateHistoryMap.put("toCreateNotUpdated", new Long (toCreateNotUpdated));
765             updateHistoryMap.put("toStoreInserted", new Long (toStoreInserted));
766             updateHistoryMap.put("toStoreUpdated", new Long (toStoreUpdated));
767             updateHistoryMap.put("toStoreNotUpdated", new Long (toStoreNotUpdated));
768             updateHistoryMap.put("toRemoveDeleted", new Long (toRemoveDeleted));
769             updateHistoryMap.put("toRemoveAlreadyDeleted", new Long (toRemoveAlreadyDeleted));
770             updateHistoryMap.put("runningTimeMillis", new Long (runningTimeMillis));
771             updateHistoryMap.put("totalStoreCalls", new Long (totalStoreCalls));
772             updateHistoryMap.put("totalSplits", new Long (totalSplits));
773             updateHistoryMap.put("totalRowsExported", new Long (totalRowsExported));
774             updateHistoryMap.put("totalRowsToCreate", new Long (totalRowsToCreate));
775             updateHistoryMap.put("totalRowsToStore", new Long (totalRowsToStore));
776             updateHistoryMap.put("totalRowsToRemove", new Long (totalRowsToRemove));
777             updateHistoryMap.put("perSplitMinMillis", new Long (perSplitMinMillis));
778             updateHistoryMap.put("perSplitMaxMillis", new Long (perSplitMaxMillis));
779             updateHistoryMap.put("perSplitMinItems", new Long (perSplitMinItems));
780             updateHistoryMap.put("perSplitMaxItems", new Long (perSplitMaxItems));
781             updateHistoryMap.put("userLogin", userLogin);
782             Map updateEsHistRunResult = dispatcher.runSync("updateEntitySyncHistory", updateHistoryMap);
783             
784             // now we have updated EntitySync and EntitySyncHistory, check both ops for errors...
785 if (ServiceUtil.isError(updateEsRunResult)) {
786                 String errorMsg = "Error running EntitySync [" + entitySyncId + "], update of EntitySync record with lastSuccessfulSynchTime failed.";
787                 throw new SyncDataErrorException(errorMsg, null, null, updateEsRunResult, null);
788             }
789             
790             if (ServiceUtil.isError(updateEsHistRunResult)) {
791                 String errorMsg = "Error running EntitySync [" + entitySyncId + "], update of EntitySyncHistory (startDate:[" + startDate + "]) record with lastSuccessfulSynchTime and result stats failed.";
792                 throw new SyncDataErrorException(errorMsg, null, null, updateEsHistRunResult, null);
793             }
794         } catch (GenericServiceException e) {
795             throw new SyncServiceErrorException("Error saving results reported from data store", e);
796         }
797     }
798     
799     public void saveFinalSyncResults() throws SyncDataErrorException, SyncServiceErrorException {
800         String newStatusId = "ESR_COMPLETE";
801         if (this.isOfflineSync && totalRowsExported > 0) {
802             newStatusId = "ESR_PENDING";
803         }
804
805         // the lastSuccessfulSynchTime on EntitySync will already be set, so just set status as completed
806 String esErrMsg = "Could not mark Entity Sync as complete, but all synchronization was successful";
807         try {
808             Map completeEntitySyncRes = dispatcher.runSync("updateEntitySyncRunning", UtilMisc.toMap("entitySyncId", entitySyncId, "runStatusId", newStatusId, "userLogin", userLogin));
809             if (ServiceUtil.isError(completeEntitySyncRes)) {
810                 // what to do here? try again?
811 throw new SyncDataErrorException(esErrMsg, null, null, completeEntitySyncRes, null);
812             }
813         } catch (GenericServiceException e) {
814             throw new SyncServiceErrorException(esErrMsg, e);
815         }
816         
817         // if nothing moved over, remove the history record, otherwise store status
818 long totalRows = totalRowsToCreate + totalRowsToStore + totalRowsToRemove;
819         if (totalRows == 0) {
820             String eshRemoveErrMsg = "Could not remove Entity Sync History (done becuase nothing was synced in this call), but all synchronization was successful";
821             try {
822                 Map deleteEntitySyncHistRes = dispatcher.runSync("deleteEntitySyncHistory", UtilMisc.toMap("entitySyncId", entitySyncId, "startDate", startDate, "userLogin", userLogin));
823                 if (ServiceUtil.isError(deleteEntitySyncHistRes)) {
824                     throw new SyncDataErrorException(eshRemoveErrMsg, null, null, deleteEntitySyncHistRes, null);
825                 }
826             } catch (GenericServiceException e) {
827                 throw new SyncServiceErrorException(eshRemoveErrMsg, e);
828             }
829         } else {
830             // the lastSuccessfulSynchTime on EntitySync will already be set, so just set status as completed
831 String eshCompleteErrMsg = "Could not mark Entity Sync History as complete, but all synchronization was successful";
832             try {
833                 Map completeEntitySyncHistRes = dispatcher.runSync("updateEntitySyncHistory", UtilMisc.toMap("entitySyncId", entitySyncId, "startDate", startDate, "runStatusId", "ESR_COMPLETE", "userLogin", userLogin));
834                 if (ServiceUtil.isError(completeEntitySyncHistRes)) {
835                     // what to do here? try again?
836 throw new SyncDataErrorException(eshCompleteErrMsg, null, null, completeEntitySyncHistRes, null);
837                 }
838             } catch (GenericServiceException e) {
839                 throw new SyncServiceErrorException(eshCompleteErrMsg, e);
840             }
841         }
842         
843         if (Debug.infoOn()) Debug.logInfo("Finished saveFinalSyncResults [" + entitySyncId + "]: totalRows=" + totalRows + ", totalRowsToCreate=" + totalRowsToCreate + ", totalRowsToStore=" + totalRowsToStore + ", totalRowsToRemove=" + totalRowsToRemove, module);
844     }
845
846     public Set makeEntityNameToUseSet() {
847         Set entityNameToUseSet = new HashSet ();
848         Iterator entityModelToUseUpdateIter = this.entityModelToUseList.iterator();
849         while (entityModelToUseUpdateIter.hasNext()) {
850             ModelEntity modelEntity = (ModelEntity) entityModelToUseUpdateIter.next();
851             entityNameToUseSet.add(modelEntity.getEntityName());
852         }
853         return entityNameToUseSet;
854     }
855     
856     /** 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 */
857     protected List makeEntityModelToUseList() throws GenericEntityException {
858         List entityModelToUseList = new LinkedList ();
859         List entitySyncIncludes = entitySync.getRelated("EntitySyncInclude");
860
861         // 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
862 List entitySyncGroupIncludes = entitySync.getRelated("EntitySyncInclGrpDetailView");
863         entitySyncIncludes.addAll(entitySyncGroupIncludes);
864
865         Iterator entityNameIter = delegator.getModelReader().getEntityNamesIterator();
866         while (entityNameIter.hasNext()) {
867             String entityName = (String ) entityNameIter.next();
868             ModelEntity modelEntity = delegator.getModelEntity(entityName);
869             
870             // if view-entity, throw it out
871 if (modelEntity instanceof ModelViewEntity) {
872                 continue;
873             }
874             
875             // if it doesn't have either or both of the two update stamp fields, throw it out
876 if (!modelEntity.isField(ModelEntity.STAMP_FIELD) || !modelEntity.isField(ModelEntity.STAMP_TX_FIELD)) {
877                 continue;
878             }
879             
880             // if there are no includes records, always include; otherwise check each one to make sure at least one matches
881 if (entitySyncIncludes.size() == 0) {
882                 entityModelToUseList.add(modelEntity);
883             } else {
884                 // we have different types of include applications: ESIA_INCLUDE, ESIA_EXCLUDE, ESIA_ALWAYS
885 // if we find an always we can break right there because this will always be include regardless of excludes, etc
886 // if we find an include or exclude we have to finish going through the rest of them just in case there is something that overrides it (ie an exclude for an include or an always for an exclude)
887 boolean matchesInclude = false;
888                 boolean matchesExclude = false;
889                 boolean matchesAlways = false;
890                 Iterator entitySyncIncludeIter = entitySyncIncludes.iterator();
891                 while (entitySyncIncludeIter.hasNext()) {
892                     GenericValue entitySyncInclude = (GenericValue) entitySyncIncludeIter.next();
893                     String entityOrPackage = entitySyncInclude.getString("entityOrPackage");
894                     boolean matches = false;
895                     if (entityName.equals(entityOrPackage)) {
896                         matches = true;
897                     } else if (modelEntity.getPackageName().startsWith(entityOrPackage)) {
898                         matches = true;
899                     }
900                     
901                     if (matches) {
902                         if ("ESIA_INCLUDE".equals(entitySyncInclude.getString("applEnumId"))) {
903                             matchesInclude = true;
904                         } else if ("ESIA_EXCLUDE".equals(entitySyncInclude.getString("applEnumId"))) {
905                             matchesExclude = true;
906                         } else if ("ESIA_ALWAYS".equals(entitySyncInclude.getString("applEnumId"))) {
907                             matchesAlways = true;
908                             break;
909                         }
910                     }
911                 }
912                 
913                 if (matchesAlways || (matchesInclude && !matchesExclude)) {
914                     // make sure this log message is not checked in uncommented:
915 //Debug.log("In runEntitySync adding [" + modelEntity.getEntityName() + "] to list of Entities to sync", module);
916 entityModelToUseList.add(modelEntity);
917                 }
918             }
919         }
920         
921         if (Debug.infoOn()) Debug.logInfo("In makeEntityModelToUseList for EntitySync with ID [" + entitySync.get("entitySyncId") + "] syncing " + entityModelToUseList.size() + " entities", module);
922         return entityModelToUseList;
923     }
924
925     protected static Timestamp getCurrentRunStartTime(Timestamp lastSuccessfulSynchTime, List entityModelToUseList, GenericDelegator delegator) throws GenericEntityException {
926         // if currentRunStartTime is null, what to do? I guess iterate through all entities and find earliest tx stamp
927 if (lastSuccessfulSynchTime == null) {
928             Timestamp currentRunStartTime = null;
929             Iterator entityModelToUseIter = entityModelToUseList.iterator();
930             while (entityModelToUseIter.hasNext()) {
931                 ModelEntity modelEntity = (ModelEntity) entityModelToUseIter.next();
932                 // fields to select will be PK and the STAMP_TX_FIELD, slimmed down so we don't get a ton of data back
933 List fieldsToSelect = new LinkedList (modelEntity.getPkFieldNames());
934                 // 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
935 fieldsToSelect.add(ModelEntity.STAMP_TX_FIELD);
936                 EntityListIterator eli = delegator.findListIteratorByCondition(modelEntity.getEntityName(), new EntityExpr(ModelEntity.STAMP_TX_FIELD, EntityOperator.NOT_EQUAL, null), fieldsToSelect, UtilMisc.toList(ModelEntity.STAMP_TX_FIELD));
937                 GenericValue nextValue = (GenericValue) eli.next();
938                 eli.close();
939                 if (nextValue != null) {
940                     Timestamp candidateTime = nextValue.getTimestamp(ModelEntity.STAMP_TX_FIELD);
941                     if (currentRunStartTime == null || candidateTime.before(currentRunStartTime)) {
942                         currentRunStartTime = candidateTime;
943                     }
944                 }
945             }
946             if (Debug.infoOn()) Debug.logInfo("No currentRunStartTime was stored on the EntitySync record, so searched for the earliest value and got: " + currentRunStartTime, module);
947             return currentRunStartTime;
948         } else {
949             return lastSuccessfulSynchTime;
950         }
951     }
952
953     public void saveSyncErrorInfo(String runStatusId, List errorMessages) {
954         // set error statuses on the EntitySync and EntitySyncHistory entities
955 try {
956             Map errorEntitySyncRes = dispatcher.runSync("updateEntitySyncRunning", UtilMisc.toMap("entitySyncId", entitySyncId, "runStatusId", runStatusId, "userLogin", userLogin));
957             if (ServiceUtil.isError(errorEntitySyncRes)) {
958                 errorMessages.add("Could not save error run status [" + runStatusId + "] on EntitySync with ID [" + entitySyncId + "]: " + errorEntitySyncRes.get(ModelService.ERROR_MESSAGE));
959             }
960         } catch (GenericServiceException e) {
961             errorMessages.add("Could not save error run status [" + runStatusId + "] on EntitySync with ID [" + entitySyncId + "]: " + e.toString());
962         }
963         if (startDate != null) {
964             try {
965                 Map errorEntitySyncHistoryRes = dispatcher.runSync("updateEntitySyncHistory", UtilMisc.toMap("entitySyncId", entitySyncId, "startDate", startDate, "runStatusId", runStatusId, "userLogin", userLogin));
966                 if (ServiceUtil.isError(errorEntitySyncHistoryRes)) {
967                     errorMessages.add("Could not save error run status [" + runStatusId + "] on EntitySyncHistory with ID [" + entitySyncId + "]: " + errorEntitySyncHistoryRes.get(ModelService.ERROR_MESSAGE));
968                 }
969             } catch (GenericServiceException e) {
970                 errorMessages.add("Could not save error run status [" + runStatusId + "] on EntitySyncHistory with ID [" + entitySyncId + ":" + startDate + "]: " + e.toString());
971             }
972         }
973     }
974
975     // ======================== PUSH Methods ========================
976 public void runPushStartRunning() throws SyncDataErrorException, SyncServiceErrorException, SyncAbortException {
977         if (UtilValidate.isEmpty(targetServiceName)) {
978             throw new SyncAbortException("Not running EntitySync [" + entitySyncId + "], no targetServiceName is specified, where do we send the data?");
979         }
980         
981         // check to see if this sync is already running, if so return error
982 if (this.isEntitySyncRunning()) {
983             throw new SyncAbortException("Not running EntitySync [" + entitySyncId + "], an instance is already running.");
984         }
985         
986         String markErrorMsg = "Could not start Entity Sync service, could not mark as running";
987         try {
988             // not running, get started NOW
989 // set running status on entity sync, run in its own tx
990 Map startEntitySyncRes = dispatcher.runSync("updateEntitySyncRunning", UtilMisc.toMap("entitySyncId", entitySyncId, "runStatusId", "ESR_RUNNING", "userLogin", userLogin));
991             if (ModelService.RESPOND_ERROR.equals(startEntitySyncRes.get(ModelService.RESPONSE_MESSAGE))) {
992                 throw new SyncDataErrorException(markErrorMsg, null, null, startEntitySyncRes, null);
993             }
994         } catch (GenericServiceException e) {
995             throw new SyncServiceErrorException(markErrorMsg, e);
996         }
997         
998         // finally create the initial history record
999 this.createInitialHistory();
1000    }
1001    
1002    public long setTotalRowCounts(ArrayList valuesToCreate, ArrayList valuesToStore, List keysToRemove) {
1003        this.totalRowsToCreate = valuesToCreate.size();
1004        this.totalRowsToStore = valuesToStore.size();
1005        this.totalRowsToRemove = keysToRemove.size();
1006        this.totalRowsPerSplit = this.totalRowsToCreate + this.totalRowsToStore + this.totalRowsToRemove;
1007        return this.totalRowsPerSplit;
1008    }
1009    
1010    public void runPushSendData(ArrayList valuesToCreate, ArrayList valuesToStore, List keysToRemove) throws SyncOtherErrorException, SyncServiceErrorException {
1011        // grab the totals for this data
1012 this.setTotalRowCounts(valuesToCreate, valuesToStore, keysToRemove);
1013        
1014        // call service named on EntitySync, IFF there is actually data to send over
1015 if (this.totalRowsPerSplit > 0) {
1016            Map targetServiceMap = UtilMisc.toMap("entitySyncId", entitySyncId, "valuesToCreate", valuesToCreate, "valuesToStore", valuesToStore, "keysToRemove", keysToRemove, "userLogin", userLogin);
1017            if (UtilValidate.isNotEmpty(targetDelegatorName)) {
1018                targetServiceMap.put("delegatorName", targetDelegatorName);
1019            }
1020            String serviceErrorMsg = "Error running EntitySync [" + entitySyncId + "], call to store service [" + targetServiceName + "] failed.";
1021            try {
1022                Map remoteStoreResult = dispatcher.runSync(targetServiceName, targetServiceMap);
1023                if (ServiceUtil.isError(remoteStoreResult)) {
1024                    throw new SyncOtherErrorException(serviceErrorMsg, null, null, remoteStoreResult, null);
1025                }
1026                
1027                this.totalStoreCalls++;
1028                
1029                long toCreateInsertedCur = remoteStoreResult.get("toCreateInserted") == null ? 0 : ((Long ) remoteStoreResult.get("toCreateInserted")).longValue();
1030                long toCreateUpdatedCur = remoteStoreResult.get("toCreateUpdated") == null ? 0 : ((Long ) remoteStoreResult.get("toCreateUpdated")).longValue();
1031                long toCreateNotUpdatedCur = remoteStoreResult.get("toCreateNotUpdated") == null ? 0 : ((Long ) remoteStoreResult.get("toCreateNotUpdated")).longValue();
1032                long toStoreInsertedCur = remoteStoreResult.get("toStoreInserted") == null ? 0 : ((Long ) remoteStoreResult.get("toStoreInserted")).longValue();
1033                long toStoreUpdatedCur = remoteStoreResult.get("toStoreUpdated") == null ? 0 : ((Long ) remoteStoreResult.get("toStoreUpdated")).longValue();
1034                long toStoreNotUpdatedCur = remoteStoreResult.get("toStoreNotUpdated") == null ? 0 : ((Long ) remoteStoreResult.get("toStoreNotUpdated")).longValue();
1035                long toRemoveDeletedCur = remoteStoreResult.get("toRemoveDeleted") == null ? 0 : ((Long ) remoteStoreResult.get("toRemoveDeleted")).longValue();
1036                long toRemoveAlreadyDeletedCur = remoteStoreResult.get("toRemoveAlreadyDeleted") == null ? 0 : ((Long ) remoteStoreResult.get("toRemoveAlreadyDeleted")).longValue();
1037                
1038                this.toCreateInserted += toCreateInsertedCur;
1039                this.toCreateUpdated += toCreateUpdatedCur;
1040                this.toCreateNotUpdated += toCreateNotUpdatedCur;
1041                this.toStoreInserted += toStoreInsertedCur;
1042                this.toStoreUpdated += toStoreUpdatedCur;
1043                this.toStoreNotUpdated += toStoreNotUpdatedCur;
1044                this.toRemoveDeleted += toRemoveDeletedCur;
1045                this.toRemoveAlreadyDeleted += toRemoveAlreadyDeletedCur;
1046            } catch (GenericServiceException e) {
1047                throw new SyncServiceErrorException(serviceErrorMsg, e);
1048            }
1049        }
1050    }
1051    
1052    // ======================== PULL Methods ========================
1053 public void runPullStartOrRestoreSavedResults() throws SyncDataErrorException, SyncServiceErrorException, SyncAbortException {
1054        // if EntitySync.statusId is ESR_RUNNING, make sure startDate matches EntitySync.lastHistoryStartDate; or return error
1055 if (isEntitySyncRunning() && this.startDate == null) {
1056            throw new SyncAbortException("Not running EntitySync [" + entitySyncId + "], an instance is already running and no startDate for the current run was passed.");
1057        }
1058        
1059        if (this.startDate == null) {
1060            // get it started!
1061 String markErrorMsg = "Could not start Entity Sync service, could not mark as running";
1062            try {
1063                // not running, get started NOW
1064 // set running status on entity sync, run in its own tx
1065 Map startEntitySyncRes = dispatcher.runSync("updateEntitySyncRunning", UtilMisc.toMap("entitySyncId", entitySyncId, "runStatusId", "ESR_RUNNING", "userLogin", userLogin));
1066                if (ModelService.RESPOND_ERROR.equals(startEntitySyncRes.get(ModelService.RESPONSE_MESSAGE))) {
1067                    throw new SyncDataErrorException(markErrorMsg, null, null, startEntitySyncRes, null);
1068                }
1069            } catch (GenericServiceException e) {
1070                throw new SyncServiceErrorException(markErrorMsg, e);
1071            }
1072            
1073            // finally create the initial history record
1074 this.createInitialHistory();
1075            this.setSplitStartTime();
1076        } else {
1077            try {
1078                // set the latest values from the EntitySyncHistory, based on the values on the EntitySync
1079 GenericValue entitySyncHistory = delegator.findByPrimaryKey("EntitySyncHistory", UtilMisc.toMap("entitySyncId", entitySyncId, "startDate", startDate));
1080                this.toCreateInserted = UtilMisc.toLong(entitySyncHistory.getLong("toCreateInserted"));
1081                this.toCreateUpdated = UtilMisc.toLong(entitySyncHistory.getLong("toCreateUpdated"));
1082                this.toCreateNotUpdated = UtilMisc.toLong(entitySyncHistory.getLong("toCreateNotUpdated"));
1083
1084                this.toStoreInserted = UtilMisc.toLong(entitySyncHistory.getLong("toStoreInserted"));
1085                this.toStoreUpdated = UtilMisc.toLong(entitySyncHistory.getLong("toStoreUpdated"));
1086                this.toStoreNotUpdated = UtilMisc.toLong(entitySyncHistory.getLong("toStoreNotUpdated"));
1087
1088                this.toRemoveDeleted = UtilMisc.toLong(entitySyncHistory.getLong("toRemoveDeleted"));
1089                this.toRemoveAlreadyDeleted = UtilMisc.toLong(entitySyncHistory.getLong("toRemoveAlreadyDeleted"));
1090
1091                this.totalStoreCalls = UtilMisc.toLong(entitySyncHistory.getLong("totalStoreCalls"));
1092                this.totalSplits = UtilMisc.toLong(entitySyncHistory.getLong("totalSplits"));
1093                this.totalRowsToCreate = UtilMisc.toLong(entitySyncHistory.getLong("totalRowsToCreate"));
1094                this.totalRowsToStore = UtilMisc.toLong(entitySyncHistory.getLong("totalRowsToStore"));
1095                this.totalRowsToRemove = UtilMisc.toLong(entitySyncHistory.getLong("totalRowsToRemove"));
1096
1097                this.perSplitMinMillis = UtilMisc.toLong(entitySyncHistory.getLong("perSplitMinMillis"));
1098                this.perSplitMaxMillis = UtilMisc.toLong(entitySyncHistory.getLong("perSplitMaxMillis"));
1099                this.perSplitMinItems = UtilMisc.toLong(entitySyncHistory.getLong("perSplitMinItems"));
1100                this.perSplitMaxItems = UtilMisc.toLong(entitySyncHistory.getLong("perSplitMaxItems"));
1101
1102                this.splitStartTime = UtilMisc.toLong(entitySyncHistory.getLong("lastSplitStartTime"));
1103            } catch (GenericEntityException e) {
1104                throw new SyncDataErrorException("Error getting existing EntitySyncHistory values", e);
1105            }
1106            
1107            // got the previous values, now add to them with the values from the context...
1108 this.toCreateInserted += UtilMisc.toLong(this.context.get("toCreateInserted"));
1109            this.toCreateUpdated += UtilMisc.toLong(this.context.get("toCreateUpdated"));
1110            this.toCreateNotUpdated += UtilMisc.toLong(this.context.get("toCreateNotUpdated"));
1111            this.toStoreInserted += UtilMisc.toLong(this.context.get("toStoreInserted"));
1112            this.toStoreUpdated += UtilMisc.toLong(this.context.get("toStoreUpdated"));
1113            this.toStoreNotUpdated += UtilMisc.toLong(this.context.get("toStoreNotUpdated"));
1114            this.toRemoveDeleted += UtilMisc.toLong(this.context.get("toRemoveDeleted"));
1115            this.toRemoveAlreadyDeleted += UtilMisc.toLong(this.context.get("toRemoveAlreadyDeleted"));
1116            
1117            this.totalStoreCalls++;
1118
1119            this.saveResultsReportedFromDataStore();
1120        }
1121    }
1122
1123    // ======================== OFFLINE Methods ========================
1124 public void runOfflineStartRunning() throws SyncDataErrorException, SyncServiceErrorException, SyncAbortException {
1125        // check to see if this sync is already running, if so return error
1126 if (this.isEntitySyncRunning()) {
1127            throw new SyncAbortException("Not running EntitySync [" + entitySyncId + "], an instance is already running.");
1128        }
1129
1130        // flag this context as offline
1131 this.isOfflineSync = true;
1132
1133        String markErrorMsg = "Could not start Entity Sync service, could not mark as running";
1134        try {
1135            // not running, get started NOW
1136 // set running status on entity sync, run in its own tx
1137 Map startEntitySyncRes = dispatcher.runSync("updateEntitySyncRunning", UtilMisc.toMap("entitySyncId", entitySyncId, "runStatusId", "ESR_RUNNING", "preOfflineSynchTime", this.lastSuccessfulSynchTime, "userLogin", userLogin));
1138            if (ModelService.RESPOND_ERROR.equals(startEntitySyncRes.get(ModelService.RESPONSE_MESSAGE))) {
1139                throw new SyncDataErrorException(markErrorMsg, null, null, startEntitySyncRes, null);
1140            }
1141        } catch (GenericServiceException e) {
1142            throw new SyncServiceErrorException(markErrorMsg, e);
1143        }
1144
1145        // finally create the initial history record
1146 this.createInitialHistory();
1147    }
1148
1149    public void runSaveOfflineSyncInfo(long rowsInSplit) throws SyncDataErrorException, SyncServiceErrorException, SyncAbortException {
1150        this.totalRowsExported += rowsInSplit;
1151        this.saveResultsReportedFromDataStore();
1152    }
1153
1154    /**
1155     * Static method to obtain a list of entity names which will be synchronized
1156     */
1157    public static Set getEntitySyncModelNamesToUse(LocalDispatcher dispatcher, String entitySyncId) throws SyncDataErrorException, SyncAbortException {
1158        DispatchContext dctx = dispatcher.getDispatchContext();
1159        EntitySyncContext ctx = new EntitySyncContext(dctx, UtilMisc.toMap("entitySyncId", entitySyncId));
1160        return ctx.makeEntityNameToUseSet();
1161    }
1162
1163    /** This class signifies an abort condition, so the state and such of the EntitySync value in the datasource should not be changed */
1164    public static class SyncAbortException extends GeneralServiceException {
1165        public SyncAbortException() {
1166            super();
1167        }
1168
1169        public SyncAbortException(String str) {
1170            super(str);
1171        }
1172
1173        public SyncAbortException(String str, Throwable nested) {
1174            super(str, nested);
1175        }
1176
1177        public SyncAbortException(Throwable nested) {
1178            super(nested);
1179        }
1180
1181        public SyncAbortException(String str, List errorMsgList, Map errorMsgMap, Map nestedServiceResult, Throwable nested) {
1182            super(str, errorMsgList, errorMsgMap, nestedServiceResult, nested);
1183        }
1184    }
1185
1186    public static abstract class SyncErrorException extends GeneralServiceException {
1187        public SyncErrorException() { super(); }
1188        public SyncErrorException(String str) { super(str); }
1189        public SyncErrorException(String str, Throwable nested) { super(str, nested); }
1190        public SyncErrorException(Throwable nested) { super(nested); }
1191        public SyncErrorException(String str, List errorMsgList, Map errorMsgMap, Map nestedServiceResult, Throwable nested) { super(str, errorMsgList, errorMsgMap, nestedServiceResult, nested); }
1192        public abstract void saveSyncErrorInfo(EntitySyncContext esc);
1193    }
1194    
1195    /** 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 */
1196    public static class SyncOtherErrorException extends SyncErrorException {
1197        public SyncOtherErrorException() { super(); }
1198        public SyncOtherErrorException(String str) { super(str); }
1199        public SyncOtherErrorException(String str, Throwable nested) { super(str, nested); }
1200        public SyncOtherErrorException(Throwable nested) { super(nested); }
1201        public SyncOtherErrorException(String str, List errorMsgList, Map errorMsgMap, Map nestedServiceResult, Throwable nested) { super(str, errorMsgList, errorMsgMap, nestedServiceResult, nested); }
1202        public void saveSyncErrorInfo(EntitySyncContext esc) {
1203            if (esc != null) {
1204                List errorList = new LinkedList ();
1205                esc.saveSyncErrorInfo("ESR_OTHER_ERROR", errorList);
1206                this.addErrorMessages(errorList);
1207            }
1208        }
1209    }
1210    
1211    /** 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 */
1212    public static class SyncDataErrorException extends SyncErrorException {
1213        public SyncDataErrorException() { super(); }
1214        public SyncDataErrorException(String str) { super(str); }
1215        public SyncDataErrorException(String str, Throwable nested) { super(str, nested); }
1216        public SyncDataErrorException(Throwable nested) { super(nested); }
1217        public SyncDataErrorException(String str, List errorMsgList, Map errorMsgMap, Map nestedServiceResult, Throwable nested) { super(str, errorMsgList, errorMsgMap, nestedServiceResult, nested); }
1218        public void saveSyncErrorInfo(EntitySyncContext esc) {
1219            if (esc != null) {
1220                List errorList = new LinkedList ();
1221                esc.saveSyncErrorInfo("ESR_DATA_ERROR", errorList);
1222                this.addErrorMessages(errorList);
1223            }
1224        }
1225    }
1226    
1227    /** 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 */
1228    public static class SyncServiceErrorException extends SyncErrorException {
1229        public SyncServiceErrorException() { super(); }
1230        public SyncServiceErrorException(String str) { super(str); }
1231        public SyncServiceErrorException(String str, Throwable nested) { super(str, nested); }
1232        public SyncServiceErrorException(Throwable nested) { super(nested); }
1233        public SyncServiceErrorException(String str, List errorMsgList, Map errorMsgMap, Map nestedServiceResult, Throwable nested) { super(str, errorMsgList, errorMsgMap, nestedServiceResult, nested); }
1234        public void saveSyncErrorInfo(EntitySyncContext esc) {
1235            if (esc != null) {
1236                List errorList = new LinkedList ();
1237                esc.saveSyncErrorInfo("ESR_SERVICE_ERROR", errorList);
1238                this.addErrorMessages(errorList);
1239            }
1240        }
1241    }
1242}
1243
1244

A to Z: JavaDoc & Examples

---

Daily Java News & Articles

---

Open Source Projects

---

Open Source Codes

---

Free Computer Books

---

Remove Frame

---

Popular Tags