Master has to advertise its ID when taking over.

7 files changed, 32 insertions, 15 deletions

diff --git a/same/src/main/java/com/orbekk/same/Client.java b/same/src/main/java/com/orbekk/same/Client.java
index 4f42c4e..18fecd2 100644
--- a/same/src/main/java/com/orbekk/same/Client.java
+++ b/same/src/main/java/com/orbekk/same/Client.java
@@ -26,7 +26,7 @@ public class Client implements DiscoveryListener {
     State state;
     private String myUrl;
     String masterUrl;
-    private int masterId = -1;
+    private int masterId = 0;
     private MasterController masterController = null;
     private Broadcaster broadcaster;
     private Future<Integer> currentMasterProposal = null;
@@ -125,6 +125,12 @@ public class Client implements DiscoveryListener {
         @Override
         public void masterTakeover(String masterUrl, String networkName, 
                 int masterId) throws Exception {
+            if (masterId <= Client.this.masterId) {
+                logger.warn("{}:{} tried to take over, but current master is " +
+                		"{}:{}. Ignoring", new Object[]{masterUrl, masterId,
+                                state.getDataOf(".masterUrl"),
+                                Client.this.masterId}); 
+            }
             logger.info("MasterTakeover({}, {}, {})",
                     new Object[]{masterUrl, networkName, masterId});
             abortMasterElection();
@@ -133,8 +139,13 @@ public class Client implements DiscoveryListener {
         }
 
         @Override
-        public void masterDown() throws Exception {
-            logger.info("Master is down.");
+        public void masterDown(int masterId) throws Exception {
+            if (masterId < Client.this.masterId) {
+                logger.info("Master {} is down, but current master is {}. Ignoring.",
+                        masterId, Client.this.masterId);
+                return;
+            }
+            logger.warn("Master down.");
             connectionState = ConnectionState.UNSTABLE;
             tryBecomeMaster();
         }
@@ -282,7 +293,7 @@ public class Client implements DiscoveryListener {
         } catch (CancellationException e) {
         }
         if (!currentMasterProposal.isCancelled() && result != null) {
-            masterController.enableMaster(state);
+            masterController.enableMaster(state, result);
         }
     }
     
@@ -293,13 +304,17 @@ public class Client implements DiscoveryListener {
         }
     }
     
+    private int getMasterIdEstimate() {
+        return masterId;
+    }
+    
     public void startMasterElection() {
         List<String> participants = state.getList(".participants");
         broadcaster.broadcast(participants, new ServiceOperation() {
             @Override public void run(String url) {
                 ClientService client = connections.getClient(url);
                 try {
-                    client.masterDown();
+                    client.masterDown(getMasterIdEstimate());
                 } catch (Exception e) {
                     logger.info("{}.masterDown() did not respond (ignored): " +
                     		url, e);
diff --git a/same/src/main/java/com/orbekk/same/ClientService.java b/same/src/main/java/com/orbekk/same/ClientService.java
index dfca2cc..f1247a5 100644
--- a/same/src/main/java/com/orbekk/same/ClientService.java
+++ b/same/src/main/java/com/orbekk/same/ClientService.java
@@ -18,5 +18,5 @@ public interface ClientService {
             int masterId) throws Exception;
     
     /** The master is down, so start a new master election. */
-    void masterDown() throws Exception;
+    void masterDown(int masterId) throws Exception;
 }
diff --git a/same/src/main/java/com/orbekk/same/Master.java b/same/src/main/java/com/orbekk/same/Master.java
index 4402487..32d6ec2 100644
--- a/same/src/main/java/com/orbekk/same/Master.java
+++ b/same/src/main/java/com/orbekk/same/Master.java
@@ -136,7 +136,7 @@ public class Master {
     }
 
     /** This master should take over from an earlier master. */
-    public void resumeFrom(State lastKnownState) {
+    public void resumeFrom(State lastKnownState, final int masterId) {
         state = lastKnownState;
         broadcaster.broadcast(state.getList(".participants"),
                 new ServiceOperation() {
@@ -145,7 +145,7 @@ public class Master {
                 ClientService client = connections.getClient(url);
                 try {
                     client.masterTakeover(myUrl,
-                            state.getDataOf(".networkName"), 0);
+                            state.getDataOf(".networkName"), masterId);
                 } catch (Exception e) {
                     logger.info("Client {} failed to acknowledge new master. " +
                     		"Removing {}", url);
diff --git a/same/src/main/java/com/orbekk/same/MasterController.java b/same/src/main/java/com/orbekk/same/MasterController.java
index 05ff91b..155fc4c 100644
--- a/same/src/main/java/com/orbekk/same/MasterController.java
+++ b/same/src/main/java/com/orbekk/same/MasterController.java
@@ -1,6 +1,6 @@
 package com.orbekk.same;
 
 public interface MasterController {
-    void enableMaster(State lastKnownState);
+    void enableMaster(State lastKnownState, int masterId);
     void disableMaster();
 }
diff --git a/same/src/main/java/com/orbekk/same/SameController.java b/same/src/main/java/com/orbekk/same/SameController.java
index ceeca73..8788be0 100644
--- a/same/src/main/java/com/orbekk/same/SameController.java
+++ b/same/src/main/java/com/orbekk/same/SameController.java
@@ -37,13 +37,13 @@ public class SameController {
 
     private MasterController masterController = new MasterController() {
         @Override
-        public void enableMaster(State lastKnownState) {
+        public void enableMaster(State lastKnownState, int masterId) {
             String masterUrl = configuration.get("baseUrl") +
                     "MasterService.json";
             master = Master.create(connections, serviceBroadcaster,
                     masterUrl, configuration.get("networkName"));
             if (lastKnownState != null) {
-                master.resumeFrom(lastKnownState);
+                master.resumeFrom(lastKnownState, masterId);
             }
             master.start();
             masterService.setService(master.getService());
@@ -173,7 +173,7 @@ public class SameController {
 
     public void createNetwork(String networkName) {
         masterController.disableMaster();
-        masterController.enableMaster(null);
+        masterController.enableMaster(null, 1);
         String masterUrl = configuration.get("baseUrl") +
                 "MasterService.json";
         joinNetwork(masterUrl);
diff --git a/same/src/test/java/com/orbekk/same/FunctionalTest.java b/same/src/test/java/com/orbekk/same/FunctionalTest.java
index fb8bd10..0da38a6 100644
--- a/same/src/test/java/com/orbekk/same/FunctionalTest.java
+++ b/same/src/test/java/com/orbekk/same/FunctionalTest.java
@@ -112,8 +112,8 @@ public class FunctionalTest {
         joinClients();
         MasterController controller = new MasterController() {
             @Override
-            public void enableMaster(State lastKnownState) {
-                newMaster.resumeFrom(lastKnownState);
+            public void enableMaster(State lastKnownState, int masterId) {
+                newMaster.resumeFrom(lastKnownState, masterId);
             }
             @Override
             public void disableMaster() {
diff --git a/same/src/test/java/com/orbekk/same/MasterTest.java b/same/src/test/java/com/orbekk/same/MasterTest.java
index 927b504..e2cbadd 100644
--- a/same/src/test/java/com/orbekk/same/MasterTest.java
+++ b/same/src/test/java/com/orbekk/same/MasterTest.java
@@ -37,10 +37,12 @@ public class MasterTest {
         @Override
         public void masterTakeover(String masterUrl, String networkName, int masterId)
                 throws Exception {
+            throw new Exception("Unreachable client");
         }
 
         @Override
-        public void masterDown() throws Exception {
+        public void masterDown(int masterId) throws Exception {
+            throw new Exception("Unreachable client");
         }
     }