SYMMETRICDS-208 - Record count during manual insert of data/event/batch.

JumpMind · Mar 26, 2010 · 4b273d4 · 4b273d4
1 parent 8fa53ce
commit 4b273d4
Show file tree

Hide file tree

Showing 6 changed files with 36 additions and 10 deletions.
diff --git a/symmetric/src/main/java/org/jumpmind/symmetric/service/IDataService.java b/symmetric/src/main/java/org/jumpmind/symmetric/service/IDataService.java
@@ -9,6 +9,7 @@
 import org.jumpmind.symmetric.ext.IHeartbeatListener;
 import org.jumpmind.symmetric.load.IReloadListener;
 import org.jumpmind.symmetric.model.Data;
+import org.jumpmind.symmetric.model.DataEventType;
 import org.jumpmind.symmetric.model.DataRef;
 import org.jumpmind.symmetric.model.Node;
 import org.jumpmind.symmetric.model.Trigger;
@@ -56,7 +57,7 @@ public interface IDataService {
 
     public void insertDataEvent(JdbcTemplate template, long dataId, long batchId, String routerId);
 
-    public void insertDataEventAndOutgoingBatch(long dataId, String channelId, String nodeId, String routerId);
+    public void insertDataEventAndOutgoingBatch(long dataId, String channelId, String nodeId, DataEventType eventType, String routerId);
 
     public void insertDataAndDataEventAndOutgoingBatch(Data data, String channelId, List<Node> nodes, String routerId);
 

diff --git a/symmetric/src/main/java/org/jumpmind/symmetric/service/impl/DataService.java b/symmetric/src/main/java/org/jumpmind/symmetric/service/impl/DataService.java
@@ -203,18 +203,19 @@ public void insertDataAndDataEventAndOutgoingBatch(Data data, String channelId,
             List<Node> nodes, String routerId) {
         long dataId = insertData(data);
         for (Node node : nodes) {
-            insertDataEventAndOutgoingBatch(dataId, channelId, node.getNodeId(), routerId);
+            insertDataEventAndOutgoingBatch(dataId, channelId, node.getNodeId(), data.getEventType(), routerId);
         }
     }
 
     public void insertDataAndDataEventAndOutgoingBatch(Data data, String nodeId, String routerId) {
         long dataId = insertData(data);
-        insertDataEventAndOutgoingBatch(dataId, data.getChannelId(), nodeId, routerId);
+        insertDataEventAndOutgoingBatch(dataId, data.getChannelId(), nodeId, data.getEventType(), routerId);
     }
 
-    public void insertDataEventAndOutgoingBatch(long dataId, String channelId, String nodeId,
+    public void insertDataEventAndOutgoingBatch(long dataId, String channelId, String nodeId, DataEventType eventType,
             String routerId) {
         OutgoingBatch outgoingBatch = new OutgoingBatch(nodeId, channelId);
+        outgoingBatch.incrementEventCount(eventType);
         outgoingBatchService.insertOutgoingBatch(outgoingBatch);
         insertDataEvent(new DataEvent(dataId, outgoingBatch.getBatchId(), routerId));
     }

diff --git a/symmetric/src/main/java/org/jumpmind/symmetric/service/impl/OutgoingBatchService.java b/symmetric/src/main/java/org/jumpmind/symmetric/service/impl/OutgoingBatchService.java
@@ -110,8 +110,25 @@ public Object doInPreparedStatement(PreparedStatement ps) throws SQLException, D
                         ps.setString(1, outgoingBatch.getNodeId());
                         ps.setString(2, outgoingBatch.getChannelId());
                         ps.setString(3, outgoingBatch.getStatus().name());
-                        ps.setString(4, outgoingBatch.getLastUpdatedHostName());
-                        ps.setTimestamp(5, new Timestamp(outgoingBatch.getLastUpdatedTime().getTime()));
+                        ps.setLong(4, outgoingBatch.getByteCount());
+                        ps.setLong(5, outgoingBatch.getSentCount());
+                        ps.setLong(6, outgoingBatch.getDataEventCount());
+                        ps.setLong(7, outgoingBatch.getReloadEventCount()); 
+                        ps.setLong(8, outgoingBatch.getInsertEventCount());
+                        ps.setLong(9, outgoingBatch.getUpdateEventCount());
+                        ps.setLong(10, outgoingBatch.getDeleteEventCount());
+                        ps.setLong(11, outgoingBatch.getOtherEventCount());
+                        ps.setLong(12, outgoingBatch.getRouterMillis());
+                        ps.setLong(13, outgoingBatch.getNetworkMillis()); 
+                        ps.setLong(14, outgoingBatch.getFilterMillis());
+                        ps.setLong(15, outgoingBatch.getLoadMillis());
+                        ps.setLong(16, outgoingBatch.getExtractMillis()); 
+                        ps.setString(17, outgoingBatch.getSqlState());
+                        ps.setLong(18, outgoingBatch.getSqlCode());
+                        ps.setString(19, StringUtils.abbreviate(outgoingBatch.getSqlMessage(), 1000));
+                        ps.setLong(20, outgoingBatch.getFailedDataId());
+                        ps.setString(21, outgoingBatch.getLastUpdatedHostName());
+                        ps.setTimestamp(22, new Timestamp(outgoingBatch.getLastUpdatedTime().getTime()));
                         return null;
                     }
                 });

diff --git a/...tric/src/main/resources/org/jumpmind/symmetric/service/impl/outgoingbatch-service-sql.xml b/...tric/src/main/resources/org/jumpmind/symmetric/service/impl/outgoingbatch-service-sql.xml
@@ -20,8 +20,12 @@
     </entry>         
     <entry key="insertOutgoingBatchSql">
       <value>
-        insert into $[sym.sync.table.prefix]_outgoing_batch (batch_id, 
-        node_id, channel_id, status, last_update_hostname, last_update_time, create_time) values (null, ?, ?, ?, ?, ?, current_timestamp)
+        insert into $[sym.sync.table.prefix]_outgoing_batch (
+        node_id, channel_id, status, byte_count, sent_count, data_event_count, 
+        reload_event_count, insert_event_count, update_event_count, delete_event_count, other_event_count, 
+        router_millis, network_millis, filter_millis, load_millis, extract_millis, sql_state, sql_code, sql_message, 
+        failed_data_id, last_update_hostname, last_update_time, batch_id, create_time) 
+        values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, null, current_timestamp)
       </value>
     </entry>
     <entry key="updateOutgoingBatchSql">

diff --git a/symmetric/src/test/java/org/jumpmind/symmetric/service/impl/PurgeServiceTest.java b/symmetric/src/test/java/org/jumpmind/symmetric/service/impl/PurgeServiceTest.java
@@ -250,9 +250,10 @@ private void setupPartiallySentData() {
         getDataService().insertDataAndDataEventAndOutgoingBatch(data,
                 TestConstants.TEST_CLIENT_EXTERNAL_ID, router.getRouter().getRouterId());
         int dataId = getJdbcTemplate().queryForInt("select max(data_id) from sym_data");
-        getDataService().insertDataEventAndOutgoingBatch(dataId, data.getChannelId(), "00003",
+        data.setDataId(dataId);
+        getDataService().insertDataEventAndOutgoingBatch(dataId, data.getChannelId(), "00003", DataEventType.INSERT,
                 router.getRouter().getRouterId());
-        getDataService().insertDataEventAndOutgoingBatch(dataId, data.getChannelId(), "00010",
+        getDataService().insertDataEventAndOutgoingBatch(dataId, data.getChannelId(), "00010", DataEventType.INSERT,
                 router.getRouter().getRouterId());
         getOutgoingBatchService().markAllAsSentForNode(TestConstants.TEST_CLIENT_NODE);
     }

diff --git a/symmetric/src/test/java/org/jumpmind/symmetric/service/mock/MockDataService.java b/symmetric/src/test/java/org/jumpmind/symmetric/service/mock/MockDataService.java
@@ -29,6 +29,7 @@
 import org.jumpmind.symmetric.ext.IHeartbeatListener;
 import org.jumpmind.symmetric.load.IReloadListener;
 import org.jumpmind.symmetric.model.Data;
+import org.jumpmind.symmetric.model.DataEventType;
 import org.jumpmind.symmetric.model.DataRef;
 import org.jumpmind.symmetric.model.Node;
 import org.jumpmind.symmetric.model.Trigger;
@@ -108,6 +109,7 @@ public void insertDataEvent(JdbcTemplate template, long dataId, long batchId, St
     }
 
     public void insertDataEventAndOutgoingBatch(long dataId, String channelId, String nodeId,
+            DataEventType eventType,
             String routerId) {
 
     }