netconf/netconf-topology/src/test/java/org/opendaylight/netconf/topology/ActorTest.java

   1 /*
   2  * Copyright (c) 2015 Cisco 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
   9 package org.opendaylight.netconf.topology;
  10
  11 import static com.jayway.awaitility.Awaitility.await;
  12 import static org.mockito.Matchers.any;
  13 import static org.mockito.Mockito.when;
  14
  15 import akka.actor.ActorRef;
  16 import akka.actor.ActorSystem;
  17 import akka.actor.TypedActor;
  18 import akka.actor.TypedActorExtension;
  19 import akka.actor.TypedProps;
  20 import akka.japi.Creator;
  21 import com.google.common.base.Function;
  22 import com.google.common.base.Optional;
  23 import com.google.common.base.Preconditions;
  24 import com.google.common.collect.Lists;
  25 import com.google.common.util.concurrent.CheckedFuture;
  26 import com.google.common.util.concurrent.FutureCallback;
  27 import com.google.common.util.concurrent.Futures;
  28 import com.google.common.util.concurrent.ListenableFuture;
  29 import com.google.common.util.concurrent.SettableFuture;
  30 import com.typesafe.config.ConfigFactory;
  31 import java.util.ArrayList;
  32 import java.util.Collections;
  33 import java.util.List;
  34 import java.util.concurrent.Callable;
  35 import java.util.concurrent.ExecutorService;
  36 import java.util.concurrent.Executors;
  37 import java.util.concurrent.TimeUnit;
  38 import javassist.ClassPool;
  39 import javax.annotation.Nonnull;
  40 import javax.annotation.Nullable;
  41 import org.junit.Before;
  42 import org.junit.Ignore;
  43 import org.junit.Test;
  44 import org.mockito.Mock;
  45 import org.mockito.MockitoAnnotations;
  46 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  47 import org.opendaylight.controller.md.sal.binding.api.DataChangeListener;
  48 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
  49 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipService;
  50 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
  51 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  52 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  53 import org.opendaylight.controller.md.sal.dom.api.DOMNotification;
  54 import org.opendaylight.controller.md.sal.dom.api.DOMRpcService;
  55 import org.opendaylight.netconf.sal.connect.netconf.listener.NetconfSessionPreferences;
  56 import org.opendaylight.netconf.topology.NodeManagerCallback.NodeManagerCallbackFactory;
  57 import org.opendaylight.netconf.topology.TopologyManagerCallback.TopologyManagerCallbackFactory;
  58 import org.opendaylight.netconf.topology.example.ExampleNodeManagerCallback;
  59 import org.opendaylight.netconf.topology.example.ExampleTopologyManagerCallback;
  60 import org.opendaylight.netconf.topology.example.LoggingSalNodeWriter;
  61 import org.opendaylight.netconf.topology.impl.NetconfNodeOperationalDataAggregator;
  62 import org.opendaylight.netconf.topology.util.BaseTopologyManager;
  63 import org.opendaylight.netconf.topology.util.NoopRoleChangeStrategy;
  64 import org.opendaylight.netconf.topology.util.TopologyRoleChangeStrategy;
  65 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev100924.Host;
  66 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev100924.IpAddress;
  67 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev100924.Ipv4Address;
  68 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev100924.PortNumber;
  69 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.$YangModuleInfoImpl;
  70 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNode;
  71 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNodeBuilder;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNodeConnectionStatus.ConnectionStatus;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.status.AvailableCapabilitiesBuilder;
  74 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.status.ClusteredConnectionStatusBuilder;
  75 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.status.UnavailableCapabilitiesBuilder;
  76 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.status.clustered.connection.status.NodeStatus.Status;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.status.clustered.connection.status.NodeStatusBuilder;
  78 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.status.unavailable.capabilities.UnavailableCapability;
  79 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
  80 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
  81 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  82 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeBuilder;
  83 import org.opendaylight.yangtools.binding.data.codec.gen.impl.StreamWriterGenerator;
  84 import org.opendaylight.yangtools.binding.data.codec.impl.BindingNormalizedNodeCodecRegistry;
  85 import org.opendaylight.yangtools.sal.binding.generator.impl.ModuleInfoBackedContext;
  86 import org.opendaylight.yangtools.sal.binding.generator.util.BindingRuntimeContext;
  87 import org.opendaylight.yangtools.sal.binding.generator.util.JavassistUtils;
  88 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  89 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  90 import org.slf4j.Logger;
  91 import org.slf4j.LoggerFactory;
  92
  93 public class ActorTest {
  94
  95     private static final Logger LOG = LoggerFactory.getLogger(ActorTest.class);
  96
  97     private static final String TOPOLOGY_NETCONF = "TopologyNetconf";
  98
  99     @Mock
 100     private EntityOwnershipService entityOwnershipService;
 101
 102     @Mock
 103     private DataBroker dataBroker;
 104
 105     @Mock
 106     private ReadOnlyTransaction mockedReadOnlyTx;
 107
 108     private static final BindingNormalizedNodeCodecRegistry CODEC_REGISTRY;
 109
 110     static {
 111         final ModuleInfoBackedContext moduleInfoBackedContext = ModuleInfoBackedContext.create();
 112         moduleInfoBackedContext.addModuleInfos(Collections.singletonList($YangModuleInfoImpl.getInstance()));
 113         final Optional<SchemaContext> schemaContextOptional = moduleInfoBackedContext.tryToCreateSchemaContext();
 114         Preconditions.checkState(schemaContextOptional.isPresent());
 115         final SchemaContext topologySchemaCtx = schemaContextOptional.get();
 116
 117         final JavassistUtils javassist = JavassistUtils.forClassPool(ClassPool.getDefault());
 118         CODEC_REGISTRY = new BindingNormalizedNodeCodecRegistry(StreamWriterGenerator.create(javassist));
 119         CODEC_REGISTRY.onBindingRuntimeContextUpdated(BindingRuntimeContext.create(moduleInfoBackedContext, topologySchemaCtx));
 120     }
 121
 122     private static final String PATH_MASTER = "akka.tcp://NetconfNode@127.0.0.1:2552/user/TopologyNetconf";
 123     private static final String PATH_SLAVE1 = "akka.tcp://NetconfNode@127.0.0.1:2553/user/TopologyNetconf";
 124     private static final String PATH_SLAVE2 = "akka.tcp://NetconfNode@127.0.0.1:2554/user/TopologyNetconf";
 125
 126     private static final List<String> PATHS_MASTER = Lists.newArrayList(PATH_SLAVE1, PATH_SLAVE2);
 127     private static final List<String> PATHS_SLAVE1 = Lists.newArrayList(PATH_MASTER, PATH_SLAVE2);
 128     private static final List<String> PATHS_SLAVE2 = Lists.newArrayList(PATH_MASTER, PATH_SLAVE1);
 129
 130     private static final ActorSystem ACTOR_SYSTEM = ActorSystem.create("NetconfNode", ConfigFactory.load("netconf-node1"));
 131     private static final ActorSystem ACTOR_SYSTEM_SLAVE1 = ActorSystem.create("NetconfNode", ConfigFactory.load("netconf-node2"));
 132     private static final ActorSystem ACTOR_SYSTEM_SLAVE2 = ActorSystem.create("NetconfNode", ConfigFactory.load("netconf-node3"));
 133
 134     private static final ExecutorService callbackExecutor = Executors.newFixedThreadPool(8);
 135
 136     private TopologyManager master = null;
 137
 138     @Before
 139     public void setup() {
 140         MockitoAnnotations.initMocks(this);
 141         final SettableFuture<Optional<Topology>> settableFuture = SettableFuture.create();
 142         final CheckedFuture<Optional<Topology>, ReadFailedException> checkedFuture = Futures.makeChecked(settableFuture, new Function<Exception, ReadFailedException>() {
 143             @Nullable
 144             @Override
 145             public ReadFailedException apply(Exception input) {
 146                 return new ReadFailedException("Dummy future should never return this");
 147             }
 148         });
 149         settableFuture.set(Optional.<Topology>absent());
 150         when(mockedReadOnlyTx.read(any(LogicalDatastoreType.class), any(InstanceIdentifier.class))).thenReturn(checkedFuture);
 151         when(dataBroker.registerDataChangeListener(
 152                 any(LogicalDatastoreType.class),
 153                 any(InstanceIdentifier.class),
 154                 any(DataChangeListener.class),
 155                 any(DataChangeScope.class))).thenReturn(null);
 156         when(dataBroker.newReadOnlyTransaction()).thenReturn(mockedReadOnlyTx);
 157     }
 158
 159     private void setMaster(final TopologyManager manager) {
 160
 161     }
 162
 163     @Test
 164     public void testRealActors() throws Exception {
 165
 166         EntityOwnershipService topoOwnership = new TestingEntityOwnershipService();
 167         // load from config
 168         final TopologyManager master = createManagerWithOwnership(ACTOR_SYSTEM, TOPOLOGY_NETCONF, true, createRealTopoTestingNodeCallbackFactory(), new TopologyRoleChangeStrategy(dataBroker, topoOwnership, TOPOLOGY_NETCONF, "topology-manager"));
 169         Thread.sleep(1000);
 170         final TopologyManager slave1 = createManagerWithOwnership(ACTOR_SYSTEM_SLAVE1, TOPOLOGY_NETCONF, false, createRealTopoTestingNodeCallbackFactory(), new TopologyRoleChangeStrategy(dataBroker, topoOwnership, TOPOLOGY_NETCONF, "topology-manager"));
 171         final TopologyManager slave2 = createManagerWithOwnership(ACTOR_SYSTEM_SLAVE2, TOPOLOGY_NETCONF, false, createRealTopoTestingNodeCallbackFactory(), new TopologyRoleChangeStrategy(dataBroker, topoOwnership, TOPOLOGY_NETCONF, "topology-manager"));
 172
 173         await().atMost(30L, TimeUnit.SECONDS).until(new Callable<Boolean>() {
 174             @Override
 175             public Boolean call() throws Exception {
 176                 return master.hasAllPeersUp();
 177             }
 178         });
 179
 180         final List<ListenableFuture<Node>> futures = new ArrayList<>();
 181         for (int i = 0; i <= 1; i++) {
 182             final String nodeid = "testing-node" + i;
 183             final Node testingNode = new NodeBuilder()
 184                     .setNodeId(new NodeId(nodeid))
 185                     .addAugmentation(NetconfNode.class,
 186                             new NetconfNodeBuilder()
 187                                     .setHost(new Host(new IpAddress(new Ipv4Address("127.0.0.1"))))
 188                                     .setPort(new PortNumber(10000 + i))
 189                                     .build())
 190                     .build();
 191             final ListenableFuture<Node> nodeListenableFuture = master.onNodeCreated(new NodeId(nodeid), testingNode);
 192             futures.add(nodeListenableFuture);
 193             Futures.addCallback(nodeListenableFuture, new FutureCallback<Node>() {
 194                 @Override
 195                 public void onSuccess(Node result) {
 196                     LOG.warn("Node {} created succesfully on all nodes", result.getNodeId().getValue());
 197                 }
 198
 199                 @Override
 200                 public void onFailure(Throwable t) {
 201                     LOG.warn("Node creation failed. ", t);
 202                 }
 203             });
 204         }
 205
 206         for (int i = 0; i <= 1; i++) {
 207             final String nodeid = "testing-node" + i;
 208             final Node testingNode = new NodeBuilder()
 209                     .setNodeId(new NodeId(nodeid))
 210                     .addAugmentation(NetconfNode.class,
 211                             new NetconfNodeBuilder()
 212                                     .setHost(new Host(new IpAddress(new Ipv4Address("127.0.0.1"))))
 213                                     .setPort(new PortNumber(10000 + i))
 214                                     .build())
 215                     .build();
 216             final ListenableFuture<Node> nodeListenableFuture = master.onNodeUpdated(new NodeId(nodeid), testingNode);
 217             futures.add(nodeListenableFuture);
 218             Futures.addCallback(nodeListenableFuture, new FutureCallback<Node>() {
 219                 @Override
 220                 public void onSuccess(Node result) {
 221                     LOG.warn("Node {} updated succesfully on all nodes", result.getNodeId().getValue());
 222                 }
 223
 224                 @Override
 225                 public void onFailure(Throwable t) {
 226                     LOG.warn("Node update failed. ", t);
 227                 }
 228             });
 229         }
 230         LOG.debug("Waiting for updates to finish");
 231         Futures.allAsList(futures).get();
 232
 233
 234         final List<ListenableFuture<Void>> deleteFutures = new ArrayList<>();
 235         for (int i = 0; i <= 1; i++) {
 236             final String nodeid = "testing-node" + i;
 237             final ListenableFuture<Void> nodeListenableFuture = master.onNodeDeleted(new NodeId(nodeid));
 238             deleteFutures.add(nodeListenableFuture);
 239             Futures.addCallback(nodeListenableFuture, new FutureCallback<Void>() {
 240                 @Override
 241                 public void onSuccess(Void result) {
 242                     LOG.warn("Node {} succesfully deleted on all nodes", nodeid);
 243                 }
 244
 245                 @Override
 246                 public void onFailure(Throwable t) {
 247                     LOG.warn("Node delete failed. ", t);
 248                 }
 249             });
 250
 251         }
 252         LOG.warn("All tasks submitted");
 253         Futures.allAsList(futures).get();
 254         Futures.allAsList(deleteFutures).get();
 255
 256         TypedActor.get(ACTOR_SYSTEM).stop(master);
 257         TypedActor.get(ACTOR_SYSTEM_SLAVE1).stop(slave1);
 258         TypedActor.get(ACTOR_SYSTEM_SLAVE2).stop(slave2);
 259
 260     }
 261
 262     // TODO seems like stopping actors is not enough to create an actor with same name, split this into multiple classes?
 263     @Ignore
 264     @Test
 265     public void testWithDummyOwnershipService() throws Exception {
 266
 267         final TestingEntityOwnershipService ownershipService = new TestingEntityOwnershipService();
 268         // load from config
 269         final TopologyManager master = createNoopRoleChangeNode(ACTOR_SYSTEM, TOPOLOGY_NETCONF, true, createRealTopoCallbackFactory(ownershipService));
 270         final TopologyManager slave1 = createNoopRoleChangeNode(ACTOR_SYSTEM_SLAVE1, TOPOLOGY_NETCONF, false, createRealTopoCallbackFactory(ownershipService));
 271         final TopologyManager slave2 = createNoopRoleChangeNode(ACTOR_SYSTEM_SLAVE2, TOPOLOGY_NETCONF, false, createRealTopoCallbackFactory(ownershipService));
 272
 273         await().atMost(10L, TimeUnit.SECONDS).until(new Callable<Boolean>() {
 274             @Override
 275             public Boolean call() throws Exception {
 276                 return master.hasAllPeersUp();
 277             }
 278         });
 279
 280         final List<ListenableFuture<Node>> futures = new ArrayList<>();
 281         for (int i = 0; i <= 0; i++) {
 282             final String nodeid = "testing-node" + i;
 283             final Node testingNode = new NodeBuilder()
 284                     .setNodeId(new NodeId(nodeid))
 285                     .addAugmentation(NetconfNode.class,
 286                             new NetconfNodeBuilder()
 287                                     .setHost(new Host(new IpAddress(new Ipv4Address("127.0.0.1"))))
 288                                     .setPort(new PortNumber(10000 + i))
 289                                     .build())
 290                     .build();
 291             final ListenableFuture<Node> nodeListenableFuture = master.onNodeCreated(new NodeId(nodeid), testingNode);
 292             futures.add(nodeListenableFuture);
 293             Futures.addCallback(nodeListenableFuture, new FutureCallback<Node>() {
 294                 @Override
 295                 public void onSuccess(Node result) {
 296                     LOG.warn("Node {} created succesfully on all nodes", result.getNodeId().getValue());
 297                 }
 298
 299                 @Override
 300                 public void onFailure(Throwable t) {
 301                     LOG.warn("Node creation failed. ", t);
 302                 }
 303             });
 304         }
 305
 306         Futures.allAsList(futures).get();
 307         ownershipService.distributeOwnership();
 308
 309         Thread.sleep(30000);
 310         TypedActor.get(ACTOR_SYSTEM).stop(master);
 311         TypedActor.get(ACTOR_SYSTEM_SLAVE1).stop(slave1);
 312         TypedActor.get(ACTOR_SYSTEM_SLAVE2).stop(slave2);
 313     }
 314
 315     private TopologyManager createNoopRoleChangeNode(final ActorSystem actorSystem, final String topologyId, final boolean isMaster,
 316                                                      final TopologyManagerCallbackFactory topologyManagerCallbackFactory) {
 317
 318         final TypedActorExtension typedActorExtension = TypedActor.get(actorSystem);
 319         return typedActorExtension.typedActorOf(new TypedProps<>(TopologyManager.class, new Creator<BaseTopologyManager>() {
 320             @Override
 321             public BaseTopologyManager create() throws Exception {
 322                 return new BaseTopologyManager(actorSystem,
 323                         CODEC_REGISTRY,
 324                         dataBroker,
 325                         topologyId,
 326                         topologyManagerCallbackFactory,
 327                         new TestingSuccesfulStateAggregator(),
 328                         new LoggingSalNodeWriter(),
 329                         new NoopRoleChangeStrategy(),
 330                         isMaster);
 331             }
 332         }), TOPOLOGY_NETCONF);
 333     }
 334
 335     private TopologyManager createManagerWithOwnership(final ActorSystem actorSystem, final String topologyId, final boolean isMaster,
 336                                                        final TopologyManagerCallbackFactory topologyManagerCallbackFactory, final RoleChangeStrategy roleChangeStrategy) {
 337         final TypedActorExtension typedActorExtension = TypedActor.get(actorSystem);
 338         return typedActorExtension.typedActorOf(new TypedProps<>(TopologyManager.class, new Creator<BaseTopologyManager>() {
 339             @Override
 340             public BaseTopologyManager create() throws Exception {
 341                 return new BaseTopologyManager(actorSystem,
 342                         CODEC_REGISTRY,
 343                         dataBroker,
 344                         topologyId,
 345                         topologyManagerCallbackFactory,
 346                         new NetconfNodeOperationalDataAggregator(),
 347                         new LoggingSalNodeWriter(),
 348                         roleChangeStrategy,
 349                         isMaster);
 350             }
 351         }), TOPOLOGY_NETCONF);
 352     }
 353
 354     private TopologyManagerCallbackFactory createRealTopoTestingNodeCallbackFactory() {
 355         final NodeManagerCallbackFactory nodeManagerCallbackFactory = new NodeManagerCallbackFactory() {
 356             @Override
 357             public NodeManagerCallback create(String nodeId, String topologyId, ActorSystem actorSystem) {
 358                 return new LoggingNodeManagerCallback();
 359             }
 360         };
 361
 362         return new TopologyManagerCallbackFactory() {
 363             @Override
 364             public TopologyManagerCallback create(ActorSystem actorSystem, String topologyId) {
 365                 return new ExampleTopologyManagerCallback(actorSystem, dataBroker, topologyId, nodeManagerCallbackFactory, new LoggingSalNodeWriter());
 366             }
 367         };
 368     }
 369
 370     private TopologyManagerCallbackFactory createRealTopoCallbackFactory(final EntityOwnershipService entityOwnershipService) {
 371         final NodeManagerCallbackFactory nodeManagerCallbackFactory = new NodeManagerCallbackFactory() {
 372             @Override
 373             public NodeManagerCallback create(String nodeId, String topologyId, ActorSystem actorSystem) {
 374                 return new ExampleNodeManagerCallback();
 375             }
 376         };
 377
 378         return new TopologyManagerCallbackFactory() {
 379             @Override
 380             public TopologyManagerCallback create(ActorSystem actorSystem, String topologyId) {
 381                 return new ExampleTopologyManagerCallback(actorSystem, dataBroker, topologyId, nodeManagerCallbackFactory);
 382             }
 383         };
 384     }
 385
 386     private TopologyManagerCallbackFactory createTestingTopoCallbackFactory() {
 387         return new TopologyManagerCallbackFactory() {
 388             @Override
 389             public TopologyManagerCallback create(ActorSystem actorSystem, String topologyId) {
 390                 return new TestingTopologyManagerCallback();
 391             }
 392         };
 393     }
 394
 395     public static class LoggingNodeManagerCallback implements NodeManagerCallback {
 396
 397         @Nonnull
 398         @Override
 399         public Node getInitialState(@Nonnull NodeId nodeId, @Nonnull Node configNode) {
 400             final NetconfNode netconfNode = configNode.getAugmentation(NetconfNode.class);
 401             return new NodeBuilder()
 402                     .setNodeId(nodeId)
 403                     .addAugmentation(NetconfNode.class,
 404                             new NetconfNodeBuilder()
 405                                     .setHost(netconfNode.getHost())
 406                                     .setPort(netconfNode.getPort())
 407                                     .setConnectionStatus(ConnectionStatus.Connecting)
 408                                     .setAvailableCapabilities(new AvailableCapabilitiesBuilder().setAvailableCapability(new ArrayList<String>()).build())
 409                                     .setUnavailableCapabilities(new UnavailableCapabilitiesBuilder().setUnavailableCapability(new ArrayList<UnavailableCapability>()).build())
 410                                     .setClusteredConnectionStatus(
 411                                             new ClusteredConnectionStatusBuilder()
 412                                                     .setNodeStatus(
 413                                                             Lists.newArrayList(
 414                                                                     new NodeStatusBuilder()
 415                                                                             .setNode("testing-node")
 416                                                                             .setStatus(Status.Unavailable)
 417                                                                             .build()))
 418                                                     .build())
 419                                     .build())
 420                     .build();
 421         }
 422
 423         @Nonnull
 424         @Override
 425         public Node getFailedState(@Nonnull NodeId nodeId, @Nonnull Node configNode) {
 426             final NetconfNode netconfNode = configNode.getAugmentation(NetconfNode.class);
 427             return new NodeBuilder()
 428                     .setNodeId(nodeId)
 429                     .addAugmentation(NetconfNode.class,
 430                             new NetconfNodeBuilder()
 431                                     .setHost(netconfNode.getHost())
 432                                     .setPort(netconfNode.getPort())
 433                                     .setConnectionStatus(ConnectionStatus.UnableToConnect)
 434                                     .setAvailableCapabilities(new AvailableCapabilitiesBuilder().setAvailableCapability(new ArrayList<String>()).build())
 435                                     .setUnavailableCapabilities(new UnavailableCapabilitiesBuilder().setUnavailableCapability(new ArrayList<UnavailableCapability>()).build())
 436                                     .setClusteredConnectionStatus(
 437                                             new ClusteredConnectionStatusBuilder()
 438                                                     .setNodeStatus(
 439                                                             Collections.singletonList(
 440                                                                     new NodeStatusBuilder()
 441                                                                             .setNode("testing-node")
 442                                                                             .setStatus(Status.Failed)
 443                                                                             .build()))
 444                                                     .build())
 445                                     .build())
 446                     .build();
 447         }
 448
 449         @Nonnull
 450         @Override
 451         public ListenableFuture<Node> onNodeCreated(@Nonnull NodeId nodeId, @Nonnull Node configNode) {
 452             LOG.debug("Creating node {} with config {}", nodeId, configNode);
 453             final NetconfNode augmentation = configNode.getAugmentation(NetconfNode.class);
 454             return Futures.immediateFuture(new NodeBuilder()
 455                     .setNodeId(nodeId)
 456                     .addAugmentation(NetconfNode.class,
 457                             new NetconfNodeBuilder()
 458                                     .setConnectionStatus(ConnectionStatus.Connected)
 459                                     .setHost(augmentation.getHost())
 460                                     .setPort(augmentation.getPort())
 461                                     .setAvailableCapabilities(new AvailableCapabilitiesBuilder().setAvailableCapability(new ArrayList<String>()).build())
 462                                     .setUnavailableCapabilities(new UnavailableCapabilitiesBuilder().setUnavailableCapability(new ArrayList<UnavailableCapability>()).build())
 463                                     .setClusteredConnectionStatus(
 464                                             new ClusteredConnectionStatusBuilder()
 465                                                     .setNodeStatus(
 466                                                             Collections.singletonList(
 467                                                                     new NodeStatusBuilder()
 468                                                                             .setNode("testing-node")
 469                                                                             .setStatus(Status.Connected)
 470                                                                             .build()))
 471                                                     .build())
 472                                     .build())
 473                     .build());
 474         }
 475
 476         @Nonnull
 477         @Override
 478         public ListenableFuture<Node> onNodeUpdated(@Nonnull NodeId nodeId, @Nonnull Node configNode) {
 479             LOG.debug("Updating node {} with config {}", nodeId, configNode);
 480             final NetconfNode augmentation = configNode.getAugmentation(NetconfNode.class);
 481             return Futures.immediateFuture(new NodeBuilder()
 482                     .setNodeId(nodeId)
 483                     .addAugmentation(NetconfNode.class,
 484                             new NetconfNodeBuilder()
 485                                     .setConnectionStatus(ConnectionStatus.Connected)
 486                                     .setHost(augmentation.getHost())
 487                                     .setPort(augmentation.getPort())
 488                                     .setAvailableCapabilities(new AvailableCapabilitiesBuilder().setAvailableCapability(new ArrayList<String>()).build())
 489                                     .setUnavailableCapabilities(new UnavailableCapabilitiesBuilder().setUnavailableCapability(new ArrayList<UnavailableCapability>()).build())
 490                                     .setClusteredConnectionStatus(
 491                                             new ClusteredConnectionStatusBuilder()
 492                                                     .setNodeStatus(
 493                                                             Collections.singletonList(
 494                                                                     new NodeStatusBuilder()
 495                                                                             .setNode("testing-node")
 496                                                                             .setStatus(Status.Connected)
 497                                                                             .build()))
 498                                                     .build())
 499                                     .build())
 500                     .build());
 501         }
 502
 503         @Nonnull
 504         @Override
 505         public ListenableFuture<Void> onNodeDeleted(@Nonnull NodeId nodeId) {
 506             LOG.debug("Deleting node {}", nodeId);
 507             return Futures.immediateFuture(null);
 508         }
 509
 510         @Nonnull
 511         @Override
 512         public ListenableFuture<Node> getCurrentStatusForNode(@Nonnull NodeId nodeId) {
 513             return null;
 514         }
 515
 516         @Override
 517         public void onRoleChanged(RoleChangeDTO roleChangeDTO) {
 518
 519         }
 520
 521         @Override
 522         public void onReceive(Object o, ActorRef actorRef) {
 523
 524         }
 525
 526         @Override
 527         public void onDeviceConnected(SchemaContext remoteSchemaContext, NetconfSessionPreferences netconfSessionPreferences, DOMRpcService deviceRpc) {
 528
 529         }
 530
 531         @Override
 532         public void onDeviceDisconnected() {
 533
 534         }
 535
 536         @Override
 537         public void onDeviceFailed(Throwable throwable) {
 538
 539         }
 540
 541         @Override
 542         public void onNotification(DOMNotification domNotification) {
 543
 544         }
 545
 546         @Override
 547         public void close() {
 548
 549         }
 550     }
 551
 552     public static class TestingTopologyManagerCallback implements TopologyManagerCallback {
 553
 554         public TestingTopologyManagerCallback() {
 555
 556         }
 557
 558         @Override
 559         public ListenableFuture<Node> onNodeCreated(NodeId nodeId, Node node) {
 560             LOG.warn("Actor system that called this: {}", TypedActor.context().system().settings().toString());
 561             return Futures.immediateFuture(new NodeBuilder()
 562                     .setNodeId(nodeId)
 563                     .addAugmentation(NetconfNode.class,
 564                             new NetconfNodeBuilder()
 565                                     .setConnectionStatus(ConnectionStatus.Connected)
 566                                     .setHost(new Host(new IpAddress(new Ipv4Address("127.0.0.1"))))
 567                                     .setPort(new PortNumber(2555))
 568                                     .setAvailableCapabilities(new AvailableCapabilitiesBuilder().setAvailableCapability(new ArrayList<String>()).build())
 569                                     .setUnavailableCapabilities(new UnavailableCapabilitiesBuilder().setUnavailableCapability(new ArrayList<UnavailableCapability>()).build())
 570                                     .build())
 571                     .build());
 572         }
 573
 574         @Override
 575         public ListenableFuture<Node> onNodeUpdated(NodeId nodeId, Node node) {
 576             LOG.warn("Actor system that called this: {}", TypedActor.context().system().settings().toString());
 577             LOG.debug("Update called on node {}, with config {}", nodeId.getValue(), node);
 578             return Futures.immediateFuture(new NodeBuilder()
 579                     .setNodeId(nodeId)
 580                     .addAugmentation(NetconfNode.class,
 581                             new NetconfNodeBuilder()
 582                                     .setConnectionStatus(ConnectionStatus.Connected)
 583                                     .setHost(new Host(new IpAddress(new Ipv4Address("127.0.0.1"))))
 584                                     .setPort(new PortNumber(65535))
 585                                     .setAvailableCapabilities(new AvailableCapabilitiesBuilder().setAvailableCapability(new ArrayList<String>()).build())
 586                                     .setUnavailableCapabilities(new UnavailableCapabilitiesBuilder().setUnavailableCapability(new ArrayList<UnavailableCapability>()).build())
 587                                     .build())
 588                     .build());
 589         }
 590
 591         @Override
 592         public ListenableFuture<Void> onNodeDeleted(NodeId nodeId) {
 593             LOG.debug("Delete called on node {}", nodeId.getValue());
 594             return Futures.immediateFuture(null);
 595         }
 596
 597         @Nonnull
 598         @Override
 599         public ListenableFuture<Node> getCurrentStatusForNode(@Nonnull NodeId nodeId) {
 600             return null;
 601         }
 602
 603         @Override
 604         public void onRoleChanged(RoleChangeDTO roleChangeDTO) {
 605
 606         }
 607
 608         @Override
 609         public void onReceive(Object o, ActorRef actorRef) {
 610
 611         }
 612
 613         @Nonnull
 614         @Override
 615         public Node getInitialState(@Nonnull NodeId nodeId, @Nonnull Node configNode) {
 616             return new NodeBuilder()
 617                     .setNodeId(nodeId)
 618                     .addAugmentation(NetconfNode.class,
 619                             new NetconfNodeBuilder()
 620                                     .setConnectionStatus(ConnectionStatus.Connecting)
 621                                     .setHost(new Host(new IpAddress(new Ipv4Address("127.0.0.1"))))
 622                                     .setPort(new PortNumber(65535))
 623                                     .setAvailableCapabilities(new AvailableCapabilitiesBuilder().setAvailableCapability(new ArrayList<String>()).build())
 624                                     .setUnavailableCapabilities(new UnavailableCapabilitiesBuilder().setUnavailableCapability(new ArrayList<UnavailableCapability>()).build())
 625                                     .build())
 626                     .build();
 627         }
 628
 629         @Nonnull
 630         @Override
 631         public Node getFailedState(@Nonnull NodeId nodeId, @Nonnull Node configNode) {
 632             return new NodeBuilder()
 633                     .setNodeId(nodeId)
 634                     .addAugmentation(NetconfNode.class,
 635                             new NetconfNodeBuilder()
 636                                     .setConnectionStatus(ConnectionStatus.UnableToConnect)
 637                                     .setHost(new Host(new IpAddress(new Ipv4Address("127.0.0.1"))))
 638                                     .setPort(new PortNumber(65535))
 639                                     .setAvailableCapabilities(new AvailableCapabilitiesBuilder().setAvailableCapability(new ArrayList<String>()).build())
 640                                     .setUnavailableCapabilities(new UnavailableCapabilitiesBuilder().setUnavailableCapability(new ArrayList<UnavailableCapability>()).build())
 641                                     .build())
 642                     .build();
 643         }
 644     }
 645
 646     public class TestingSuccesfulStateAggregator implements StateAggregator {
 647
 648         @Override
 649         public ListenableFuture<Node> combineCreateAttempts(List<ListenableFuture<Node>> stateFutures) {
 650             final SettableFuture<Node> future = SettableFuture.create();
 651             final ListenableFuture<List<Node>> allAsList = Futures.allAsList(stateFutures);
 652             Futures.addCallback(allAsList, new FutureCallback<List<Node>>() {
 653                 @Override
 654                 public void onSuccess(List<Node> result) {
 655                     for (int i = 0; i < result.size() - 1; i++) {
 656                         if (!result.get(i).equals(result.get(i + 1))) {
 657                             LOG.warn("Node 1 {}: {}", result.get(i).getClass(), result.get(i));
 658                             LOG.warn("Node 2 {}: {}", result.get(i + 1).getClass(), result.get(i + 1));
 659                             future.setException(new IllegalStateException("Create futures have different result"));
 660                             LOG.warn("Future1 : {}  Future2 : {}", result.get(i), result.get(i+1));
 661                         }
 662                     }
 663                     future.set(result.get(0));
 664                 }
 665
 666                 @Override
 667                 public void onFailure(Throwable t) {
 668                     LOG.error("One of the combined create attempts failed {}", t);
 669                     future.setException(t);
 670                 }
 671             }, TypedActor.context().dispatcher());
 672
 673             return future;
 674         }
 675
 676         @Override
 677         public ListenableFuture<Node> combineUpdateAttempts(List<ListenableFuture<Node>> stateFutures) {
 678             final SettableFuture<Node> future = SettableFuture.create();
 679             final ListenableFuture<List<Node>> allAsList = Futures.allAsList(stateFutures);
 680             Futures.addCallback(allAsList, new FutureCallback<List<Node>>() {
 681                 @Override
 682                 public void onSuccess(List<Node> result) {
 683                     for (int i = 0; i < result.size() - 1; i++) {
 684                         if (!result.get(i).equals(result.get(i + 1))) {
 685                             future.setException(new IllegalStateException("Update futures have different result"));
 686                         }
 687                     }
 688                     future.set(result.get(0));
 689                 }
 690
 691                 @Override
 692                 public void onFailure(Throwable t) {
 693                     LOG.error("One of the combined update attempts failed {}", t);
 694                     future.setException(t);
 695                 }
 696             });
 697             return future;
 698         }
 699
 700         @Override
 701         public ListenableFuture<Void> combineDeleteAttempts(List<ListenableFuture<Void>> stateFutures) {
 702             final SettableFuture<Void> future = SettableFuture.create();
 703             final ListenableFuture<List<Void>> allAsList = Futures.allAsList(stateFutures);
 704             Futures.addCallback(allAsList, new FutureCallback<List<Void>>() {
 705                 @Override
 706                 public void onSuccess(List<Void> result) {
 707                     future.set(null);
 708                 }
 709
 710                 @Override
 711                 public void onFailure(Throwable t) {
 712                     LOG.error("One of the combined delete attempts failed {}", t);
 713                     future.setException(t);
 714                 }
 715             });
 716             return future;
 717         }
 718     }
 719 }