Handle partition changes for replications correctly.

git-svn-id: https://svn.erp5.org/repos/neo/branches/prototype3@213 71dcc9de-d417-0410-9af5-da40c76e7ee4

neo/storage/operation.py

@@ -4,7 +4,8 @@ from neo.storage.handler import StorageEventHandler
 from neo.protocol import INVALID_UUID, INVALID_SERIAL, INVALID_TID, \
         INVALID_PARTITION, \
         RUNNING_STATE, BROKEN_STATE, TEMPORARILY_DOWN_STATE, \
-        MASTER_NODE_TYPE, STORAGE_NODE_TYPE, CLIENT_NODE_TYPE
+        MASTER_NODE_TYPE, STORAGE_NODE_TYPE, CLIENT_NODE_TYPE, \
+        DISCARDED_STATE, OUT_OF_DATE_STATE
 from neo.util import dump
 from neo.node import MasterNode, StorageNode, ClientNode
 from neo.connection import ClientConnection
@@ -194,6 +195,7 @@ class OperationEventHandler(StorageEventHandler):
             pt = app.pt
             if app.ptid >= ptid:
                 # Ignore this packet.
+                logging.info('ignoring older partition changes')
                 return
 
             # First, change the table on memory.
@@ -208,6 +210,13 @@ class OperationEventHandler(StorageEventHandler):
 
                 pt.setCell(offset, node, state)
 
+                if uuid == app.uuid:
+                    # If this is for myself, this can affect replications.
+                    if state == DISCARDED_STATE:
+                        app.replicator.removePartition(offset)
+                    elif state == OUT_OF_DATE_STATE:
+                        app.replicator.addPartition(offset)
+
             # Then, the database.
             app.dm.changePartitionTable(ptid, cell_list)
         else:

neo/storage/replicator.py

@@ -390,3 +390,21 @@ class Replicator(object):
             if self.replication_done:
                 logging.info('replication is done')
                 self._finishReplication()
+
+    def removePartition(self, rid):
+        """This is a callback from OperationEventHandler."""
+        try:
+            self.partition_list.remove(rid)
+        except ValueError:
+            pass
+
+        try:
+            self.new_partition_list.remove(rid)
+        except ValueError:
+            pass
+
+    def addPartition(self, rid):
+        """This is a callback from OperationEventHandler."""
+        if rid not in self.partition_list \
+                and rid not in self.new_partition_list:
+            self.new_partition_list.append(rid)