+ if(cohortEntry.isExpired(cacheExpiryTimeoutInMillis) || cohortEntry.isAborted()) {
+ continue;
+ }
+
+ final LinkedList<BatchedModifications> newModifications = new LinkedList<>();
+ cohortEntry.getDataTreeModification().applyToCursor(new AbstractBatchedModificationsCursor() {
+ @Override
+ protected BatchedModifications getModifications() {
+ if(newModifications.isEmpty() ||
+ newModifications.getLast().getModifications().size() >= maxModificationsPerBatch) {
+ newModifications.add(new BatchedModifications(cohortEntry.getTransactionID(),
+ cohortEntry.getClientVersion(), ""));
+ }
+
+ return newModifications.getLast();
+ }
+ });
+
+ if(!newModifications.isEmpty()) {
+ BatchedModifications last = newModifications.getLast();
+ last.setDoCommitOnReady(cohortEntry.isDoImmediateCommit());
+ last.setReady(true);
+ last.setTotalMessagesSent(newModifications.size());
+ messages.addAll(newModifications);
+
+ if(!cohortEntry.isDoImmediateCommit() && cohortEntry.getState() == State.CAN_COMMITTED) {
+ messages.add(new CanCommitTransaction(cohortEntry.getTransactionID(),
+ cohortEntry.getClientVersion()));
+ }
+
+ if(!cohortEntry.isDoImmediateCommit() && cohortEntry.getState() == State.PRE_COMMITTED) {
+ messages.add(new CommitTransaction(cohortEntry.getTransactionID(),
+ cohortEntry.getClientVersion()));
+ }