import akka.actor.typed.ActorRef;
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.actor.typed.ActorRef;
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.cluster.ddata.typed.javadsl.DistributedData;
import akka.cluster.ddata.typed.javadsl.Replicator;
import akka.cluster.ddata.typed.javadsl.ReplicatorMessageAdapter;
import akka.cluster.ddata.typed.javadsl.DistributedData;
import akka.cluster.ddata.typed.javadsl.Replicator;
import akka.cluster.ddata.typed.javadsl.ReplicatorMessageAdapter;
import java.time.Duration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.time.Duration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
+import org.eclipse.jdt.annotation.Nullable;
+import org.opendaylight.controller.eos.akka.owner.supervisor.command.DataCenterActivated;
+import org.opendaylight.controller.eos.akka.owner.supervisor.command.GetEntitiesBackendRequest;
+import org.opendaylight.controller.eos.akka.owner.supervisor.command.GetEntityBackendRequest;
+import org.opendaylight.controller.eos.akka.owner.supervisor.command.GetEntityOwnerBackendRequest;
import org.opendaylight.controller.eos.akka.owner.supervisor.command.InitialCandidateSync;
import org.opendaylight.controller.eos.akka.owner.supervisor.command.InitialOwnerSync;
import org.opendaylight.controller.eos.akka.owner.supervisor.command.OwnerSupervisorCommand;
import org.opendaylight.controller.eos.akka.owner.supervisor.command.InitialCandidateSync;
import org.opendaylight.controller.eos.akka.owner.supervisor.command.InitialOwnerSync;
import org.opendaylight.controller.eos.akka.owner.supervisor.command.OwnerSupervisorCommand;
// String representation of Entity to DOMEntity
private final Map<String, DOMEntity> entityLookup = new HashMap<>();
// String representation of Entity to DOMEntity
private final Map<String, DOMEntity> entityLookup = new HashMap<>();
- private OwnerSyncer(final ActorContext<OwnerSupervisorCommand> context) {
+ private OwnerSyncer(final ActorContext<OwnerSupervisorCommand> context,
+ final @Nullable ActorRef<OwnerSupervisorReply> notifyDatacenterStarted,
+ final BindingInstanceIdentifierCodec iidCodec) {
LOG.debug("Starting candidate and owner sync");
final ActorRef<Replicator.Command> replicator = DistributedData.get(context.getSystem()).replicator();
LOG.debug("Starting candidate and owner sync");
final ActorRef<Replicator.Command> replicator = DistributedData.get(context.getSystem()).replicator();
new ReplicatorMessageAdapter<OwnerSupervisorCommand, ORMap<DOMEntity, ORSet<String>>>(context, replicator,
Duration.ofSeconds(5)).askGet(
askReplyTo -> new Replicator.Get<>(CandidateRegistry.KEY, Replicator.readLocal(), askReplyTo),
InitialCandidateSync::new);
new ReplicatorMessageAdapter<OwnerSupervisorCommand, ORMap<DOMEntity, ORSet<String>>>(context, replicator,
Duration.ofSeconds(5)).askGet(
askReplyTo -> new Replicator.Get<>(CandidateRegistry.KEY, Replicator.readLocal(), askReplyTo),
InitialCandidateSync::new);
- public static Behavior<OwnerSupervisorCommand> create() {
- return Behaviors.setup(OwnerSyncer::new);
+ public static Behavior<OwnerSupervisorCommand> create(final ActorRef<OwnerSupervisorReply> notifyDatacenterStarted,
+ final BindingInstanceIdentifierCodec iidCodec) {
+ return Behaviors.setup(ctx -> new OwnerSyncer(ctx, notifyDatacenterStarted, iidCodec));
return newReceiveBuilder()
.onMessage(InitialCandidateSync.class, this::onInitialCandidateSync)
.onMessage(InitialOwnerSync.class, this::onInitialOwnerSync)
return newReceiveBuilder()
.onMessage(InitialCandidateSync.class, this::onInitialCandidateSync)
.onMessage(InitialOwnerSync.class, this::onInitialOwnerSync)
+ .onMessage(GetEntitiesBackendRequest.class, this::onFailEntityRpc)
+ .onMessage(GetEntityBackendRequest.class, this::onFailEntityRpc)
+ .onMessage(GetEntityOwnerBackendRequest.class, this::onFailEntityRpc)
+ private Behavior<OwnerSupervisorCommand> onFailEntityRpc(final OwnerSupervisorRequest message) {
+ LOG.debug("Failing rpc request. {}", message);
+ message.getReplyTo().tell(StatusReply.error(
+ "OwnerSupervisor is inactive so it cannot handle entity rpc requests."));
+ return this;
+ }
+
private Behavior<OwnerSupervisorCommand> onInitialCandidateSync(final InitialCandidateSync rsp) {
final Replicator.GetResponse<ORMap<DOMEntity, ORSet<String>>> response = rsp.getResponse();
if (response instanceof Replicator.GetSuccess) {
private Behavior<OwnerSupervisorCommand> onInitialCandidateSync(final InitialCandidateSync rsp) {
final Replicator.GetResponse<ORMap<DOMEntity, ORSet<String>>> response = rsp.getResponse();
if (response instanceof Replicator.GetSuccess) {
private Behavior<OwnerSupervisorCommand> switchToSupervisor() {
LOG.debug("Initial sync done, switching to supervisor. candidates: {}, owners: {}",
currentCandidates, currentOwners);
private Behavior<OwnerSupervisorCommand> switchToSupervisor() {
LOG.debug("Initial sync done, switching to supervisor. candidates: {}, owners: {}",
currentCandidates, currentOwners);
- return Behaviors.setup(ctx ->
- OwnerSupervisor.create(currentCandidates, currentOwners));
+ return Behaviors.setup(ctx -> OwnerSupervisor.create(currentCandidates, currentOwners, iidCodec));