import java.util.Iterator;
import java.util.LinkedList;
import java.util.Queue;
+import org.opendaylight.controller.cluster.access.concepts.MemberName;
import org.opendaylight.controller.cluster.datastore.DataStoreVersions;
import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier;
* @author Thomas Pantelis
*/
class EntityOwnershipShardCommitCoordinator {
- private static final Object COMMIT_RETRY_MESSAGE = "entityCommitRetry";
+ private static final Object COMMIT_RETRY_MESSAGE = new Object() {
+ @Override
+ public String toString() {
+ return "entityCommitRetry";
+ }
+ };
private final Logger log;
private int transactionIDCounter = 0;
- private final String localMemberName;
+ private final MemberName localMemberName;
private final Queue<Modification> pendingModifications = new LinkedList<>();
private BatchedModifications inflightCommit;
private Cancellable retryCommitSchedule;
- EntityOwnershipShardCommitCoordinator(String localMemberName, Logger log) {
+ EntityOwnershipShardCommitCoordinator(MemberName localMemberName, Logger log) {
this.localMemberName = localMemberName;
this.log = log;
}
boolean handleMessage(Object message, EntityOwnershipShard shard) {
boolean handled = true;
- if(CommitTransactionReply.SERIALIZABLE_CLASS.isInstance(message)) {
+ if(CommitTransactionReply.isSerializedType(message)) {
// Successful reply from a local commit.
inflightCommitSucceeded(shard);
} else if(message instanceof akka.actor.Status.Failure) {
// Failure reply from a local commit.
inflightCommitFailure(((Failure)message).cause(), shard);
- } else if(message.equals(COMMIT_RETRY_MESSAGE)) {
+ } else if(COMMIT_RETRY_MESSAGE.equals(message)) {
retryInflightCommit(shard);
} else {
handled = false;