@Test public void onDataTreeChanged() throws Exception { final List<RoutingDefinition> oldDefs = Lists.newArrayList(RouteTestFactory.createDummyRoutingDef(1, 1)); final SxpClusterRoute oldRouting = new SxpClusterRouteBuilder().setRoutingDefinition(oldDefs).build(); final List<RoutingDefinition> newDefs = Lists.newArrayList(RouteTestFactory.createDummyRoutingDef(1, 2), RouteTestFactory.createDummyRoutingDef(3, 3)); final SxpClusterRoute newRouting = new SxpClusterRouteBuilder().setRoutingDefinition(newDefs).build(); Mockito.when(dataObjectModification.getDataBefore()).thenReturn(oldRouting); Mockito.when(dataObjectModification.getDataAfter()).thenReturn(newRouting); Mockito.when(modification.getRootNode()).thenReturn(dataObjectModification); manager.init(); Assert.assertEquals(RouteListenerState.STOPPED, manager.getState()); manager.instantiateServiceInstance(); Assert.assertEquals(RouteListenerState.BEFORE_FIRST, manager.getState()); manager.onDataTreeChanged(Collections.singletonList(modification)); Mockito.verify(dataBroker) .registerDataTreeChangeListener(SxpClusterRouteManager.ROUTING_DEFINITION_DT_IDENTIFIER, manager); Mockito.verify(cssProvider).registerClusterSingletonService(manager); Mockito.verify(routeReactor).updateRouting(oldRouting, newRouting); Assert.assertEquals(RouteListenerState.WORKING, manager.getState()); }
@Override public void onDataTreeChanged(@Nonnull Collection<DataTreeModification<SxpClusterRoute>> changes) { if (!state.isProcessing()) { return; } synchronized (stateLock) { if (!state.isProcessing()) { LOG.info("Surprise: data changed notification obtained outside listening time frame"); return; } changes.forEach(change -> { if (RouteListenerState.BEFORE_FIRST.equals(state) && change.getRootNode().getDataBefore() != null) { LOG.warn("first notification after lister registered contains BEFORE"); } final DataObjectModification<SxpClusterRoute> rootNode = change.getRootNode(); final SxpClusterRoute dataBefore = rootNode.getDataBefore(); final SxpClusterRoute dataAfter = rootNode.getDataAfter(); final FluentFuture<? extends CommitInfo> routingOutcome = routeReactor.updateRouting(dataBefore, dataAfter); Futures.addCallback(routingOutcome, LOGGING_CALLBACK); }); state = RouteListenerState.WORKING; } }
SxpClusterRouteManager implements ClusteredDataTreeChangeListener<SxpClusterRoute>, ClusterSingletonService, AutoCloseable { @Override public void onDataTreeChanged(@Nonnull Collection<DataTreeModification<SxpClusterRoute>> changes) { if (!state.isProcessing()) { return; } synchronized (stateLock) { if (!state.isProcessing()) { LOG.info("Surprise: data changed notification obtained outside listening time frame"); return; } changes.forEach(change -> { if (RouteListenerState.BEFORE_FIRST.equals(state) && change.getRootNode().getDataBefore() != null) { LOG.warn("first notification after lister registered contains BEFORE"); } final DataObjectModification<SxpClusterRoute> rootNode = change.getRootNode(); final SxpClusterRoute dataBefore = rootNode.getDataBefore(); final SxpClusterRoute dataAfter = rootNode.getDataAfter(); final FluentFuture<? extends CommitInfo> routingOutcome = routeReactor.updateRouting(dataBefore, dataAfter); Futures.addCallback(routingOutcome, LOGGING_CALLBACK); }); state = RouteListenerState.WORKING; } } }
SxpClusterRouteManager implements ClusteredDataTreeChangeListener<SxpClusterRoute>, ClusterSingletonService, AutoCloseable { @Override public void onDataTreeChanged(@Nonnull Collection<DataTreeModification<SxpClusterRoute>> changes) { if (!state.isProcessing()) { return; } synchronized (stateLock) { if (!state.isProcessing()) { LOG.info("Surprise: data changed notification obtained outside listening time frame"); return; } changes.forEach(change -> { if (RouteListenerState.BEFORE_FIRST.equals(state) && change.getRootNode().getDataBefore() != null) { LOG.warn("first notification after lister registered contains BEFORE"); } final DataObjectModification<SxpClusterRoute> rootNode = change.getRootNode(); final SxpClusterRoute dataBefore = rootNode.getDataBefore(); final SxpClusterRoute dataAfter = rootNode.getDataAfter(); final FluentFuture<? extends CommitInfo> routingOutcome = routeReactor.updateRouting(dataBefore, dataAfter); Futures.addCallback(routingOutcome, LOGGING_CALLBACK); }); state = RouteListenerState.WORKING; } } SxpClusterRouteManager(final DataBroker dataBroker, final ClusterSingletonServiceProvider cssProvider,
            final RouteReactor routeReactor); }
SxpClusterRouteManager implements ClusteredDataTreeChangeListener<SxpClusterRoute>, ClusterSingletonService, AutoCloseable { @Override public void onDataTreeChanged(@Nonnull Collection<DataTreeModification<SxpClusterRoute>> changes) { if (!state.isProcessing()) { return; } synchronized (stateLock) { if (!state.isProcessing()) { LOG.info("Surprise: data changed notification obtained outside listening time frame"); return; } changes.forEach(change -> { if (RouteListenerState.BEFORE_FIRST.equals(state) && change.getRootNode().getDataBefore() != null) { LOG.warn("first notification after lister registered contains BEFORE"); } final DataObjectModification<SxpClusterRoute> rootNode = change.getRootNode(); final SxpClusterRoute dataBefore = rootNode.getDataBefore(); final SxpClusterRoute dataAfter = rootNode.getDataAfter(); final FluentFuture<? extends CommitInfo> routingOutcome = routeReactor.updateRouting(dataBefore, dataAfter); Futures.addCallback(routingOutcome, LOGGING_CALLBACK); }); state = RouteListenerState.WORKING; } } SxpClusterRouteManager(final DataBroker dataBroker, final ClusterSingletonServiceProvider cssProvider,
            final RouteReactor routeReactor); void init(); @Override void onDataTreeChanged(@Nonnull Collection<DataTreeModification<SxpClusterRoute>> changes); @Override ServiceGroupIdentifier getIdentifier(); @Override void instantiateServiceInstance(); @Override ListenableFuture<Void> closeServiceInstance(); @Override void close(); }
SxpClusterRouteManager implements ClusteredDataTreeChangeListener<SxpClusterRoute>, ClusterSingletonService, AutoCloseable { @Override public void onDataTreeChanged(@Nonnull Collection<DataTreeModification<SxpClusterRoute>> changes) { if (!state.isProcessing()) { return; } synchronized (stateLock) { if (!state.isProcessing()) { LOG.info("Surprise: data changed notification obtained outside listening time frame"); return; } changes.forEach(change -> { if (RouteListenerState.BEFORE_FIRST.equals(state) && change.getRootNode().getDataBefore() != null) { LOG.warn("first notification after lister registered contains BEFORE"); } final DataObjectModification<SxpClusterRoute> rootNode = change.getRootNode(); final SxpClusterRoute dataBefore = rootNode.getDataBefore(); final SxpClusterRoute dataAfter = rootNode.getDataAfter(); final FluentFuture<? extends CommitInfo> routingOutcome = routeReactor.updateRouting(dataBefore, dataAfter); Futures.addCallback(routingOutcome, LOGGING_CALLBACK); }); state = RouteListenerState.WORKING; } } SxpClusterRouteManager(final DataBroker dataBroker, final ClusterSingletonServiceProvider cssProvider,
            final RouteReactor routeReactor); void init(); @Override void onDataTreeChanged(@Nonnull Collection<DataTreeModification<SxpClusterRoute>> changes); @Override ServiceGroupIdentifier getIdentifier(); @Override void instantiateServiceInstance(); @Override ListenableFuture<Void> closeServiceInstance(); @Override void close();  }