Updates to monitor process information about loads that are running.

symmetric-core/src/main/java/org/jumpmind/symmetric/model/LoadSummary.java

@@ -0,0 +1,230 @@
+package org.jumpmind.symmetric.model;
+
+import java.io.Serializable;
+import java.util.Date;
+
+public class LoadSummary  implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private long loadId;
+    private String nodeId;
+    private boolean inError;
+    private int finishedBatchCount;
+    private int pendingBatchCount;
+    private long currentBatchId;
+    private long currentDataEventCount;
+    private String createBy;
+    private Date createTime;
+    private Date lastUpdateTime;
+    private String channelQueue;
+    private int tableCount;
+    private boolean isFullLoad;
+    private boolean isCreateFirst;
+    private boolean isDeleteFirst;
+    private boolean isRequestProcessed;
+    private boolean isConditional;
+    private boolean isCustomSql;
+    private long batchCount;
+    private String currentTableName;
+    private long dataCount;
+    private String processStatus;
+    private String processName;
+
+    public boolean isActive() {
+        return pendingBatchCount > 0;
+    }
+
+    public void setInError(boolean inError) {
+        this.inError = inError;
+    }
+
+    public boolean isInError() {
+        return inError;
+    }
+
+    public long getLoadId() {
+        return loadId;
+    }
+
+    public void setLoadId(long loadId) {
+        this.loadId = loadId;
+    }
+
+    public String getNodeId() {
+        return nodeId;
+    }
+
+    public void setNodeId(String nodeId) {
+        this.nodeId = nodeId;
+    }
+
+    public int getFinishedBatchCount() {
+        return finishedBatchCount;
+    }
+
+    public void setFinishedBatchCount(int finishedBatchCount) {
+        this.finishedBatchCount = finishedBatchCount;
+    }
+
+    public int getPendingBatchCount() {
+        return pendingBatchCount;
+    }
+
+    public void setPendingBatchCount(int pendingBatchCount) {
+        this.pendingBatchCount = pendingBatchCount;
+    }
+
+    public long getCurrentBatchId() {
+        return currentBatchId;
+    }
+
+    public void setCurrentBatchId(long currentBatchId) {
+        this.currentBatchId = currentBatchId;
+    }
+
+    public long getCurrentDataEventCount() {
+        return currentDataEventCount;
+    }
+
+    public void setCurrentDataEventCount(long currentDataEventCount) {
+        this.currentDataEventCount = currentDataEventCount;
+    }
+
+    public Date getCreateTime() {
+        return createTime;
+    }
+
+    public void setCreateTime(Date createTime) {
+        this.createTime = createTime;
+    }
+
+    public Date getLastUpdateTime() {
+        return lastUpdateTime;
+    }
+
+    public void setLastUpdateTime(Date lastUpdateTime) {
+        this.lastUpdateTime = lastUpdateTime;
+    }
+
+    public void setCreateBy(String createBy) {
+        this.createBy = createBy;
+    }
+
+    public String getCreateBy() {
+        return createBy;
+    }
+
+    public String getLoadNodeId() {
+        return String.format("%010d-%s", loadId, nodeId);
+    }
+
+    public String getChannelQueue() {
+        return channelQueue;
+    }
+
+    public void setChannelQueue(String channelQueue) {
+        this.channelQueue = channelQueue;
+    }
+
+    public int getTableCount() {
+        return tableCount;
+    }
+
+    public void setTableCount(int tableCount) {
+        this.tableCount = tableCount;
+    }
+
+    public boolean isFullLoad() {
+        return isFullLoad;
+    }
+
+    public void setFullLoad(boolean isFullLoad) {
+        this.isFullLoad = isFullLoad;
+    }
+
+    public boolean isCreateFirst() {
+        return isCreateFirst;
+    }
+
+    public void setCreateFirst(boolean isCreateFirst) {
+        this.isCreateFirst = isCreateFirst;
+    }
+
+    public boolean isDeleteFirst() {
+        return isDeleteFirst;
+    }
+
+    public void setDeleteFirst(boolean isDeleteFirst) {
+        this.isDeleteFirst = isDeleteFirst;
+    }
+
+    public boolean isRequestProcessed() {
+        return isRequestProcessed;
+    }
+
+    public void setRequestProcessed(boolean isRequestProcessed) {
+        this.isRequestProcessed = isRequestProcessed;
+    }
+
+    public boolean isConditional() {
+        return isConditional;
+    }
+
+    public void setConditional(boolean isConditional) {
+        this.isConditional = isConditional;
+    }
+
+    public boolean isCustomSql() {
+        return isCustomSql;
+    }
+
+    public void setCustomSql(boolean isCustomSql) {
+        this.isCustomSql = isCustomSql;
+    }
+
+    public long getBatchCount() {
+        return batchCount;
+    }
+
+    public void setBatchCount(long batchCount) {
+        this.batchCount = batchCount;
+    }
+
+    public String getCurrentTableName() {
+        return currentTableName;
+    }
+
+    public void setCurrentTableName(String currentTableName) {
+        this.currentTableName = currentTableName;
+    }
+
+    public long getDataCount() {
+        return dataCount;
+    }
+
+    public void setDataCount(long dataCount) {
+        this.dataCount = dataCount;
+    }
+
+    public String getProcessStatus() {
+        return processStatus;
+    }
+
+    public void setProcessStatus(String processStatus) {
+        this.processStatus = processStatus;
+    }
+
+    public String getProcessName() {
+        return processName;
+    }
+
+    public void setProcessName(String processName) {
+        this.processName = processName;
+    }
+
+
+
+
+}
+

symmetric-core/src/main/java/org/jumpmind/symmetric/model/ProcessInfo.java

@@ -74,6 +74,8 @@ public String toString() {
 
     private long currentBatchId;
 
+    private long currentBatchCount;
+
     private String currentChannelId;
 
     private boolean threadPerChannel;
@@ -156,8 +158,20 @@ public void incrementCurrentDataCount() {
     public void incrementBatchCount() {
         this.batchCount++;
     }
+
+    public void incrementCurrentBatchCount() {
+        this.currentBatchCount++;
+    }
+
+    public long getCurrentBatchCount() {
+		return currentBatchCount;
+	}
+
+	public void setCurrentBatchCount(long currentBatchCount) {
+		this.currentBatchCount = currentBatchCount;
+	}
 
-    public long getCurrentBatchId() {
+	public long getCurrentBatchId() {
         return currentBatchId;
     }
 

symmetric-core/src/main/java/org/jumpmind/symmetric/service/IDataService.java

@@ -33,6 +33,7 @@
 import org.jumpmind.symmetric.model.DataGap;
 import org.jumpmind.symmetric.model.Node;
 import org.jumpmind.symmetric.model.OutgoingBatch.Status;
+import org.jumpmind.symmetric.model.ProcessInfo;
 import org.jumpmind.symmetric.model.TableReloadRequest;
 import org.jumpmind.symmetric.model.TableReloadRequestKey;
 import org.jumpmind.symmetric.model.TriggerHistory;
@@ -67,9 +68,9 @@ public interface IDataService {
      */
     public String sendSQL(String nodeId, String catalogName, String schemaName, String tableName, String sql);
 
-    public void insertReloadEvents(Node targetNode, boolean reverse);
+    public void insertReloadEvents(Node targetNode, boolean reverse, ProcessInfo processInfo);
 
-    public void insertReloadEvents(Node targetNode, boolean reverse, List<TableReloadRequest> reloadRequests);
+    public void insertReloadEvents(Node targetNode, boolean reverse, List<TableReloadRequest> reloadRequests, ProcessInfo processInfo);
 
     public boolean insertReloadEvent(TableReloadRequest request, boolean deleteAtClient);
 

symmetric-core/src/main/java/org/jumpmind/symmetric/service/IOutgoingBatchService.java

@@ -24,8 +24,10 @@
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 import org.jumpmind.db.sql.ISqlTransaction;
+import org.jumpmind.symmetric.model.LoadSummary;
 import org.jumpmind.symmetric.model.OutgoingBatch;
 import org.jumpmind.symmetric.model.OutgoingBatchSummary;
 import org.jumpmind.symmetric.model.OutgoingBatches;
@@ -99,6 +101,12 @@ public List<OutgoingBatch> listOutgoingBatches(List<String> nodeIds, List<String
             List<OutgoingBatch.Status> statuses, List<String> loads, long startAtBatchId, int rowsExpected, boolean ascending);
 
     public List<OutgoingLoadSummary> getLoadSummaries(boolean activeOnly);
+
+    public Set<Long> getActiveLoads(String sourceNodeId);
+
+    public LoadSummary getLoadSummary(long loadId);
+
+    public Map<String, Map<String, Integer>> getLoadStatusSummarySql(long loadId);
 
     public void copyOutgoingBatches(String channelId, long startBatchId, String fromNodeId, String toNodeId);