import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import org.opendaylight.controller.cluster.access.concepts.MemberName;
import org.opendaylight.controller.cluster.datastore.ClusterWrapper;
import org.opendaylight.controller.cluster.datastore.DataStoreVersions;
import org.opendaylight.controller.cluster.datastore.DatastoreContext;
return actualFailure;
}
};
- public static final String MAILBOX = "bounded-mailbox";
+ public static final String BOUNDED_MAILBOX = "bounded-mailbox";
public static final String COMMIT = "commit";
private final ActorSystem actorSystem;
return future.transform(new Mapper<Object, PrimaryShardInfo>() {
@Override
- public PrimaryShardInfo checkedApply(Object response) throws Exception {
+ public PrimaryShardInfo checkedApply(Object response) throws UnknownMessageException {
if(response instanceof RemotePrimaryShardFound) {
LOG.debug("findPrimaryShardAsync received: {}", response);
RemotePrimaryShardFound found = (RemotePrimaryShardFound)response;
private PrimaryShardInfo onPrimaryShardFound(String shardName, String primaryActorPath,
short primaryVersion, DataTree localShardDataTree) {
ActorSelection actorSelection = actorSystem.actorSelection(primaryActorPath);
- PrimaryShardInfo info = new PrimaryShardInfo(actorSelection, primaryVersion,
- Optional.fromNullable(localShardDataTree));
+ PrimaryShardInfo info = localShardDataTree == null ? new PrimaryShardInfo(actorSelection, primaryVersion) :
+ new PrimaryShardInfo(actorSelection, primaryVersion, localShardDataTree);
primaryShardInfoCache.putSuccessful(shardName, info);
return info;
}
public void shutdown() {
FiniteDuration duration = datastoreContext.getShardRaftConfig().getElectionTimeOutInterval().$times(3);
try {
- Await.ready(Patterns.gracefulStop(shardManager, duration, new Shutdown()), duration);
+ Await.ready(Patterns.gracefulStop(shardManager, duration, Shutdown.INSTANCE), duration);
} catch(Exception e) {
LOG.warn("ShardManager for {} data store did not shutdown gracefully", getDataStoreName(), e);
}
return clusterWrapper;
}
- public String getCurrentMemberName(){
+ public MemberName getCurrentMemberName(){
return clusterWrapper.getCurrentMemberName();
}
/**
* Send the message to each and every shard
- *
- * @param message
*/
- public void broadcast(final Object message){
+ public void broadcast(final Function<Short, Object> messageSupplier, Class<?> messageClass){
for(final String shardName : configuration.getAllShardNames()){
Future<PrimaryShardInfo> primaryFuture = findPrimaryShardAsync(shardName);
public void onComplete(Throwable failure, PrimaryShardInfo primaryShardInfo) {
if(failure != null) {
LOG.warn("broadcast failed to send message {} to shard {}: {}",
- message.getClass().getSimpleName(), shardName, failure);
+ messageClass.getSimpleName(), shardName, failure);
} else {
+ Object message = messageSupplier.apply(primaryShardInfo.getPrimaryShardVersion());
primaryShardInfo.getPrimaryShardActor().tell(message, ActorRef.noSender());
}
}
}
}
- /**
- * @deprecated This method is present only to support backward compatibility with Helium and should not be
- * used any further
- *
- *
- * @param primaryPath
- * @param localPathOfRemoteActor
- * @return
- */
- @Deprecated
- public String resolvePath(final String primaryPath,
- final String localPathOfRemoteActor) {
- StringBuilder builder = new StringBuilder();
- String[] primaryPathElements = primaryPath.split("/");
- builder.append(primaryPathElements[0]).append("//")
- .append(primaryPathElements[1]).append(primaryPathElements[2]);
- String[] remotePathElements = localPathOfRemoteActor.split("/");
- for (int i = 3; i < remotePathElements.length; i++) {
- builder.append("/").append(remotePathElements[i]);
- }
-
- return builder.toString();
- }
-
/**
* This is a utility method that lets us get a Timer object for any operation. This is a little open-ended to allow
* us to create a timer for pretty much anything.