DatabaseService related fixes.
- Note: This patch does not fix the issue running as single node.
Change-Id: Iabfa548ca4e40e4ec5c9e76ae936300437e53d22
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/ClusterMessagingProtocolClient.java b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/ClusterMessagingProtocolClient.java
index 4f5e372..54d8e2b 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/ClusterMessagingProtocolClient.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/ClusterMessagingProtocolClient.java
@@ -112,7 +112,7 @@
clusterService.addListener(listener);
// wait for specified controller node to come up
- return null;
+ return appeared;
}
@Override
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/ClusterMessagingProtocolServer.java b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/ClusterMessagingProtocolServer.java
index a5f207e..0ecc0a5 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/ClusterMessagingProtocolServer.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/ClusterMessagingProtocolServer.java
@@ -81,7 +81,7 @@
}
}
if (handler == null) {
- log.error("There was no handler for registered!");
+ log.error("There was no handler registered!");
return;
}
}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/DatabaseClient.java b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/DatabaseClient.java
index 876f6cc..27e04c4 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/DatabaseClient.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/DatabaseClient.java
@@ -1,5 +1,6 @@
package org.onlab.onos.store.service.impl;
+import static com.google.common.base.Preconditions.checkNotNull;
import static org.slf4j.LoggerFactory.getLogger;
import java.util.Arrays;
@@ -12,7 +13,6 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
-import net.kuujo.copycat.cluster.Member;
import net.kuujo.copycat.cluster.TcpMember;
import net.kuujo.copycat.event.EventHandler;
import net.kuujo.copycat.event.LeaderElectEvent;
@@ -40,23 +40,28 @@
private final Logger log = getLogger(getClass());
private final DatabaseProtocolService protocol;
- private volatile ProtocolClient copycat = null;
- private volatile Member currentLeader = null;
+ private volatile ProtocolClient client = null;
+ private volatile TcpMember currentLeader = null;
+
public DatabaseClient(DatabaseProtocolService protocol) {
- this.protocol = protocol;
+ this.protocol = checkNotNull(protocol);
}
+ // FIXME This handler relies on a fact that local node is part of Raft cluster
@Override
public void handle(LeaderElectEvent event) {
- Member newLeader = event.leader();
+ final TcpMember newLeader = event.leader();
if (newLeader != null && !newLeader.equals(currentLeader)) {
+ log.info("{} became the new leader", newLeader);
+ ProtocolClient prevClient = client;
+ ProtocolClient newclient = protocol.createClient(newLeader);
+ newclient.connect();
+ client = newclient;
currentLeader = newLeader;
- if (copycat != null) {
- copycat.close();
+ if (prevClient != null) {
+ prevClient.close();
}
- copycat = protocol.createClient((TcpMember) currentLeader);
- copycat.connect();
}
}
@@ -92,7 +97,7 @@
SubmitRequest request =
new SubmitRequest(nextRequestId(), operationName, Arrays.asList(args));
- CompletableFuture<SubmitResponse> submitResponse = copycat.submit(request);
+ CompletableFuture<SubmitResponse> submitResponse = client.submit(request);
log.debug("Sent {} to {}", request, currentLeader);
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/DatabaseManager.java b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/DatabaseManager.java
index 306e3dc..6fcd6b9 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/DatabaseManager.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/DatabaseManager.java
@@ -173,9 +173,10 @@
Log consensusLog = new MapDBLog(LOG_FILE_PREFIX + localNode.id(),
ClusterMessagingProtocol.SERIALIZER);
+ copycat = new Copycat(stateMachine, consensusLog, cluster, copycatMessagingProtocol);
+
client = new DatabaseClient(copycatMessagingProtocol);
- copycat = new Copycat(stateMachine, consensusLog, cluster, copycatMessagingProtocol);
copycat.event(LeaderElectEvent.class).registerHandler(client);
copycat.event(LeaderElectEvent.class).registerHandler(expirationTracker);