+ final FluentFuture<? extends CommitInfo> future = tx.commit();
+ this.submitted = future;
+ future.addCallback(new FutureCallback<CommitInfo>() {
+ @Override
+ public void onSuccess(final CommitInfo result) {
+ LOG.trace("Removing routes {}, succeed", nlri);
+ }
+
+ @Override
+ public void onFailure(final Throwable throwable) {
+ LOG.error("Removing routes failed", throwable);
+ }
+ }, MoreExecutors.directExecutor());
+ }
+
+ void releaseChain() {
+ if (this.submitted != null) {
+ try {
+ this.submitted.get();
+ } catch (final InterruptedException | ExecutionException throwable) {
+ LOG.error("Write routes failed", throwable);
+ }
+ }
+ }
+
+ void storeStaleRoutes(final Set<TablesKey> gracefulTables) {
+ final CountDownLatch latch = new CountDownLatch(gracefulTables.size());
+
+ try (DOMDataTreeReadTransaction tx = this.chain.getDomChain().newReadOnlyTransaction()) {
+ for (TablesKey tablesKey : gracefulTables) {
+ final TableContext ctx = this.tables.get(tablesKey);
+ if (ctx == null) {
+ LOG.warn("Missing table for address family {}", tablesKey);
+ latch.countDown();
+ continue;
+ }
+
+ tx.read(LogicalDatastoreType.OPERATIONAL, ctx.routesPath()).addCallback(
+ new FutureCallback<Optional<NormalizedNode<?, ?>>>() {
+ @Override
+ public void onSuccess(final Optional<NormalizedNode<?, ?>> routesOptional) {
+ try {
+ if (routesOptional.isPresent()) {
+ synchronized (AdjRibInWriter.this.staleRoutesRegistry) {
+ final MapNode routesNode = (MapNode) routesOptional.get();
+ final List<NodeIdentifierWithPredicates> routes = routesNode.getValue().stream()
+ .map(MapEntryNode::getIdentifier)
+ .collect(Collectors.toList());
+ if (!routes.isEmpty()) {
+ AdjRibInWriter.this.staleRoutesRegistry.put(tablesKey, routes);
+ }
+ }
+ }
+ } finally {
+ latch.countDown();
+ }
+ }
+
+ @Override
+ public void onFailure(final Throwable throwable) {
+ LOG.warn("Failed to store stale routes for table {}", tablesKey, throwable);
+ latch.countDown();
+ }
+ }, MoreExecutors.directExecutor());
+ }
+ }
+
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ LOG.warn("Interrupted while waiting to store stale routes with {} tasks of {} to finish", latch.getCount(),
+ gracefulTables, e);
+ }
+ }
+
+ void removeStaleRoutes(final TablesKey tableKey) {
+ final TableContext ctx = this.tables.get(tableKey);
+ if (ctx == null) {
+ LOG.debug("No table for {}, not removing any stale routes", tableKey);
+ return;
+ }
+ final Collection<NodeIdentifierWithPredicates> routeKeys = this.staleRoutesRegistry.get(tableKey);
+ if (routeKeys == null || routeKeys.isEmpty()) {
+ LOG.debug("No stale routes present in table {}", tableKey);
+ return;
+ }
+ LOG.trace("Removing routes {}", routeKeys);
+ final DOMDataTreeWriteTransaction tx = this.chain.getDomChain().newWriteOnlyTransaction();
+ routeKeys.forEach(routeKey -> {
+ tx.delete(LogicalDatastoreType.OPERATIONAL, ctx.routePath(routeKey));
+ });
+ final FluentFuture<? extends CommitInfo> future = tx.commit();
+ this.submitted = future;
+ future.addCallback(new FutureCallback<CommitInfo>() {
+ @Override
+ public void onSuccess(final CommitInfo result) {
+ LOG.trace("Removing routes {}, succeed", routeKeys);
+ synchronized (AdjRibInWriter.this.staleRoutesRegistry) {
+ staleRoutesRegistry.remove(tableKey);
+ }
+ }
+
+ @Override
+ public void onFailure(final Throwable throwable) {
+ LOG.warn("Removing routes {}, failed", routeKeys, throwable);
+ }
+ }, MoreExecutors.directExecutor());