1918310 - reinstitute the one to one mapping between reload data_even…

…ts and reload batches.
JumpMind · Mar 20, 2008 · b310945 · b310945
1 parent 133db37
commit b310945
Show file tree

Hide file tree

Showing 5 changed files with 37 additions and 18 deletions.
diff --git a/symmetric/src/main/java/org/jumpmind/symmetric/service/impl/ConfigurationService.java b/symmetric/src/main/java/org/jumpmind/symmetric/service/impl/ConfigurationService.java
@@ -106,7 +106,7 @@ public void initSystemChannels() {
             logger.debug("Channel " + Constants.CHANNEL_CONFIG + " already created.");
         }
         try {
-            jdbcTemplate.update(insertChannelSql, new Object[] { Constants.CHANNEL_RELOAD, 1, 100000, 1000 });
+            jdbcTemplate.update(insertChannelSql, new Object[] { Constants.CHANNEL_RELOAD, 1, 1, 10 });
         } catch (DataIntegrityViolationException ex) {
             logger.debug("Channel " + Constants.CHANNEL_RELOAD + " already created.");
         }

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
@@ -38,12 +38,11 @@
 import org.jumpmind.symmetric.common.csv.CsvUtil;
 import org.jumpmind.symmetric.db.IDbDialect;
 import org.jumpmind.symmetric.load.IReloadListener;
-import org.jumpmind.symmetric.model.BatchType;
 import org.jumpmind.symmetric.model.Data;
 import org.jumpmind.symmetric.model.DataEvent;
 import org.jumpmind.symmetric.model.DataEventType;
 import org.jumpmind.symmetric.model.Node;
-import org.jumpmind.symmetric.model.OutgoingBatch;
+import org.jumpmind.symmetric.model.NodeChannel;
 import org.jumpmind.symmetric.model.Trigger;
 import org.jumpmind.symmetric.model.TriggerHistory;
 import org.jumpmind.symmetric.service.IConfigurationService;
@@ -171,22 +170,20 @@ public void insertReloadEvent(Node targetNode) {
             for (Trigger trigger : triggers) {
                 String xml = dbDialect.getCreateTableXML(trigger);
                 insertCreateEvent(targetNode, trigger, xml);
-                OutgoingBatch newBatch = new OutgoingBatch();
-                newBatch.setBatchType(BatchType.INITIAL_LOAD);
-                newBatch.setChannelId(trigger.getChannelId());
-                newBatch.setNodeId(targetNode.getNodeId());
-                outgoingBatchService.insertOutgoingBatch(newBatch);
+                buildReloadBatches(targetNode.getNodeId());
             }            
         }
         if (deleteFirstForReload) {
             for (ListIterator<Trigger> iterator = triggers.listIterator(triggers.size()); iterator.hasPrevious();) {
                 Trigger trigger = iterator.previous();
                 insertPurgeEvent(targetNode, trigger);
+                buildReloadBatches(targetNode.getNodeId());
             }
         }
 
         for (Trigger trigger : triggers) {
             insertReloadEvent(targetNode, trigger);
+            buildReloadBatches(targetNode.getNodeId());
         }
 
         if (listeners != null) {
@@ -200,6 +197,20 @@ public void insertReloadEvent(Node targetNode) {
         // remove all incoming events from the node are starting a reload for.
         purgeService.purgeAllIncomingEventsForNode(targetNode.getNodeId());
     }
+
+    /**
+     * This should be called after a reload event is inserted so there is a one to one between 
+     * data events and reload batches.
+     */
+    private void buildReloadBatches(String nodeId) {
+        List<NodeChannel> channels = new ArrayList<NodeChannel>(1);
+        NodeChannel channel = new NodeChannel();
+        channel.setId(Constants.CHANNEL_RELOAD);
+        channel.setEnabled(true);
+        channels.add(channel);
+        outgoingBatchService.buildOutgoingBatches(nodeId, channels);
+
+    }
 
     private void insertNodeSecurityUpdate(Node node) {
         Data data = createData(tablePrefix + "_node_security", " t.node_id = '" + node.getNodeId() + "'");

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
@@ -249,7 +249,7 @@ public void setBatchStatus(String batchId, Status status) {
     public boolean isInitialLoadComplete(String nodeId) {
 
         NodeSecurity security = nodeService.findNodeSecurity(nodeId);
-        if (security != null && security.isInitialLoadEnabled()) {
+        if (security == null || security.isInitialLoadEnabled()) {
             return false;
         }
 

diff --git a/symmetric/src/test/java/org/jumpmind/symmetric/IntegrationTest.java b/symmetric/src/test/java/org/jumpmind/symmetric/IntegrationTest.java
@@ -38,7 +38,7 @@ public class IntegrationTest extends AbstractIntegrationTest implements ITest {
 
     static final String insertCustomerSql = "insert into test_customer (customer_id, name, is_active, address, city, state, zip, entry_time, notes, icon) values(?,?,?,?,?,?,?,?,?,?)";
 
-    static final String insertTestTriggerTableSql = "insert into test_triggers_table (string_one_value, string_two_value) values(?,?)";
+    static final String insertTestTriggerTableSql = "insert into test_triggers_table (id, string_one_value, string_two_value) values(?, ?,?)";
 
     static final String updateTestTriggerTableSql = "update test_triggers_table set string_one_value=?";
 
@@ -101,14 +101,22 @@ protected void register() {
 
     protected void initialLoad() {
         rootJdbcTemplate.update(insertCustomerSql, new Object[] { 301, "Linus", "1", "42 Blanket Street",
-                "Santa Claus", "IN", 90009, new Date(), "This is a test", BINARY_DATA });        
+                "Santa Claus", "IN", 90009, new Date(), "This is a test", BINARY_DATA });    
+        rootJdbcTemplate.update(insertTestTriggerTableSql, new Object[] {1, "wow", "mom"});
+        rootJdbcTemplate.update(insertTestTriggerTableSql, new Object[] {2, "mom", "wow"});
         INodeService nodeService = (INodeService) getRootEngine().getApplicationContext().getBean(
                 Constants.NODE_SERVICE);
-        getRootEngine().reloadNode(
-                nodeService.findNodeByExternalId(TestConstants.TEST_CLIENT_NODE_GROUP,
-                        TestConstants.TEST_CLIENT_EXTERNAL_ID).getNodeId());
+        String nodeId = nodeService.findNodeByExternalId(TestConstants.TEST_CLIENT_NODE_GROUP,
+                TestConstants.TEST_CLIENT_EXTERNAL_ID).getNodeId();
+        getRootEngine().reloadNode(nodeId);
+        IOutgoingBatchService outgoingBatchService = (IOutgoingBatchService)getRootEngine().getApplicationContext().getBean(Constants.OUTGOING_BATCH_SERVICE);
+        Assert.assertFalse(outgoingBatchService.isInitialLoadComplete(nodeId));
         getClientEngine().pull();
-        Assert.assertEquals(clientJdbcTemplate.queryForInt("select count(*) from sym_node_security where initial_load_enabled=1"), 0, "Initial load was not successful");
+        Assert.assertTrue(outgoingBatchService.isInitialLoadComplete(nodeId));
+        Assert.assertEquals(clientJdbcTemplate.queryForInt("select count(*) from test_triggers_table"), 2, "test_triggers_table on the client did not contain the expected number of rows");
+        Assert.assertEquals(clientJdbcTemplate.queryForInt("select count(*) from test_customer"), 2, "test_customer on the client did not contain the expected number of rows");
+        Assert.assertEquals(clientJdbcTemplate.queryForInt("select count(*) from sym_node_security where initial_load_enabled=1"), 0, "Initial load was not successful according to the client");
+        Assert.assertEquals(rootJdbcTemplate.queryForInt("select count(*) from sym_node_security where initial_load_enabled=1"), 0, "Initial load was not successful accordign to the root");
     }
 
     protected void testSyncToClient() {
@@ -140,12 +148,12 @@ protected void testSyncToClient() {
 
     protected void testSyncToRootAutoGeneratedPrimaryKey() {
         final String NEW_VALUE = "unique new value one value";
-        clientJdbcTemplate.update(insertTestTriggerTableSql, new Object[] { "value one", "value \" two" });
+        clientJdbcTemplate.update(insertTestTriggerTableSql, new Object[] { 3, "value one", "value \" two" });
         getClientEngine().push();
         clientJdbcTemplate.update(updateTestTriggerTableSql, new Object[] { NEW_VALUE });
         getClientEngine().push();
         Assert.assertEquals(rootJdbcTemplate.queryForInt(
-                "select count(*) from test_triggers_table where string_one_value=?", new Object[] { NEW_VALUE }), 1,
+                "select count(*) from test_triggers_table where string_one_value=?", new Object[] { NEW_VALUE }), 3,
                 "The update on test_triggers_table did not work.");
     }
 

diff --git a/symmetric/src/test/resources/test-continuous-setup.sql b/symmetric/src/test/resources/test-continuous-setup.sql
@@ -1,5 +1,5 @@
 insert into sym_channel values('testchannel', 1, 50, 50, 1, null);
-insert into sym_channel values('config', 0, 50, 50, 1, null);
+insert into sym_channel values('config', 0, 1, 50, 1, null);
 insert into sym_node_group values ('symmetric','a group representing symmetric configuration');
 insert into sym_node_group values ('test-root-group','a test config');
 insert into sym_node_group values ('test-node-group','a test config');