opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/entityownership/DistributedEntityOwnershipService.java

   1 /*
   2  * Copyright (c) 2015 Brocade Communications Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.datastore.entityownership;
   9
  10 import akka.actor.ActorRef;
  11 import akka.dispatch.OnComplete;
  12 import akka.util.Timeout;
  13 import java.util.concurrent.ConcurrentHashMap;
  14 import java.util.concurrent.ConcurrentMap;
  15 import java.util.concurrent.TimeUnit;
  16 import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
  17 import org.opendaylight.controller.cluster.datastore.entityownership.messages.RegisterCandidateLocal;
  18 import org.opendaylight.controller.cluster.datastore.messages.CreateShard;
  19 import org.opendaylight.controller.md.sal.common.api.clustering.CandidateAlreadyRegisteredException;
  20 import org.opendaylight.controller.md.sal.common.api.clustering.Entity;
  21 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipCandidate;
  22 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipCandidateRegistration;
  23 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipListener;
  24 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipListenerRegistration;
  25 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipService;
  26 import org.slf4j.Logger;
  27 import org.slf4j.LoggerFactory;
  28 import scala.concurrent.Future;
  29
  30 /**
  31  * The distributed implementation of the EntityOwnershipService.
  32  *
  33  * @author Thomas Pantelis
  34  */
  35 public class DistributedEntityOwnershipService implements EntityOwnershipService, AutoCloseable {
  36     private static final Logger LOG = LoggerFactory.getLogger(DistributedEntityOwnershipService.class);
  37     static final String ENTITY_OWNERSHIP_SHARD_NAME = "entity-ownership";
  38     private static final Timeout MESSAGE_TIMEOUT = new Timeout(1, TimeUnit.MINUTES);
  39
  40     private final DistributedDataStore datastore;
  41     private final ConcurrentMap<Entity, EntityOwnershipCandidate> registeredEntities = new ConcurrentHashMap<>();
  42     private volatile ActorRef localEntityOwnershipShard;
  43
  44     public DistributedEntityOwnershipService(DistributedDataStore datastore) {
  45         this.datastore = datastore;
  46     }
  47
  48     public void start() {
  49         ActorRef shardManagerActor = datastore.getActorContext().getShardManager();
  50
  51         CreateShard createShard = new CreateShard(ENTITY_OWNERSHIP_SHARD_NAME,
  52                 datastore.getActorContext().getConfiguration().getUniqueMemberNamesForAllShards(),
  53                 newShardPropsCreator(), null);
  54
  55         Future<Object> createFuture = datastore.getActorContext().executeOperationAsync(shardManagerActor,
  56                 createShard, MESSAGE_TIMEOUT);
  57
  58         createFuture.onComplete(new OnComplete<Object>() {
  59             @Override
  60             public void onComplete(Throwable failure, Object response) {
  61                 if(failure != null) {
  62                     LOG.error("Failed to create {} shard", ENTITY_OWNERSHIP_SHARD_NAME);
  63                 } else {
  64                     LOG.info("Successfully created {} shard", ENTITY_OWNERSHIP_SHARD_NAME);
  65                 }
  66             }
  67         }, datastore.getActorContext().getClientDispatcher());
  68     }
  69
  70     private void executeEntityOwnershipShardOperation(final ActorRef shardActor, final Object message) {
  71         Future<Object> future = datastore.getActorContext().executeOperationAsync(shardActor, message, MESSAGE_TIMEOUT);
  72         future.onComplete(new OnComplete<Object>() {
  73             @Override
  74             public void onComplete(Throwable failure, Object response) {
  75                 if(failure != null) {
  76                     LOG.debug("Error sending message {} to {}", message, shardActor, failure);
  77                     // TODO - queue for retry
  78                 } else {
  79                     LOG.debug("{} message to {} succeeded", message, shardActor, failure);
  80                 }
  81             }
  82         }, datastore.getActorContext().getClientDispatcher());
  83     }
  84
  85     private void executeLocalEntityOwnershipShardOperation(final Object message) {
  86         if(localEntityOwnershipShard == null) {
  87             Future<ActorRef> future = datastore.getActorContext().findLocalShardAsync(ENTITY_OWNERSHIP_SHARD_NAME);
  88             future.onComplete(new OnComplete<ActorRef>() {
  89                 @Override
  90                 public void onComplete(Throwable failure, ActorRef shardActor) {
  91                     if(failure != null) {
  92                         LOG.error("Failed to find local {} shard", ENTITY_OWNERSHIP_SHARD_NAME, failure);
  93                     } else {
  94                         localEntityOwnershipShard = shardActor;
  95                         executeEntityOwnershipShardOperation(localEntityOwnershipShard, message);
  96                     }
  97                 }
  98             }, datastore.getActorContext().getClientDispatcher());
  99
 100         } else {
 101             executeEntityOwnershipShardOperation(localEntityOwnershipShard, message);
 102         }
 103     }
 104
 105     @Override
 106     public EntityOwnershipCandidateRegistration registerCandidate(Entity entity, EntityOwnershipCandidate candidate)
 107             throws CandidateAlreadyRegisteredException {
 108
 109         EntityOwnershipCandidate currentCandidate = registeredEntities.putIfAbsent(entity,candidate);
 110         if(currentCandidate != null) {
 111             throw new CandidateAlreadyRegisteredException(entity, currentCandidate);
 112         }
 113
 114         RegisterCandidateLocal registerCandidate = new RegisterCandidateLocal(candidate, entity);
 115
 116         LOG.debug("Registering candidate with message: " + registerCandidate);
 117
 118         executeLocalEntityOwnershipShardOperation(registerCandidate);
 119         return new DistributedEntityOwnershipCandidateRegistration(candidate, entity);
 120     }
 121
 122     @Override
 123     public EntityOwnershipListenerRegistration registerListener(Entity entity, EntityOwnershipListener listener) {
 124         // TODO Auto-generated method stub
 125         return null;
 126     }
 127
 128     @Override
 129     public void close() {
 130     }
 131
 132     protected EntityOwnershipShardPropsCreator newShardPropsCreator() {
 133         return new EntityOwnershipShardPropsCreator();
 134     }
 135 }