BUG-6497: fixup augmentation lookup
[bgpcep.git] / bgp / rib-impl / src / main / java / org / opendaylight / protocol / bgp / rib / impl / EffectiveRibInWriter.java
index 45a32878649c981d8da3fe7ebb87598ddad565d8..b62e0ec879c6b74f134de4ef278c193e2d1b1135 100644 (file)
@@ -9,40 +9,49 @@ package org.opendaylight.protocol.bgp.rib.impl;
 
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
-import com.google.common.base.Predicate;
-import com.google.common.collect.Iterables;
+import com.google.common.base.Verify;
 import java.util.Collection;
-import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import javax.annotation.Nonnull;
 import javax.annotation.concurrent.NotThreadSafe;
 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
-import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener;
+import org.opendaylight.controller.md.sal.dom.api.ClusteredDOMDataTreeChangeListener;
 import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeService;
 import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeIdentifier;
 import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
 import org.opendaylight.controller.md.sal.dom.api.DOMTransactionChain;
-import org.opendaylight.protocol.bgp.rib.spi.RIBExtensionConsumerContext;
+import org.opendaylight.protocol.bgp.rib.impl.spi.AbstractImportPolicy;
+import org.opendaylight.protocol.bgp.rib.impl.spi.ImportPolicyPeerTracker;
+import org.opendaylight.protocol.bgp.rib.impl.spi.RIBSupportContext;
+import org.opendaylight.protocol.bgp.rib.impl.spi.RIBSupportContextRegistry;
+import org.opendaylight.protocol.bgp.rib.impl.stats.peer.route.PerTableTypeRouteCounter;
+import org.opendaylight.protocol.bgp.rib.spi.IdentifierUtils;
 import org.opendaylight.protocol.bgp.rib.spi.RIBSupport;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev130925.PeerId;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev130925.PeerRole;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev130925.bgp.rib.rib.Peer;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev130925.bgp.rib.rib.peer.AdjRibIn;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev130925.bgp.rib.rib.peer.EffectiveRibIn;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev130925.rib.Tables;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev130925.rib.TablesKey;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev130925.rib.tables.Routes;
 import org.opendaylight.yangtools.concepts.ListenerRegistration;
-import org.opendaylight.yangtools.yang.common.QName;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifierWithPredicates;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
-import org.opendaylight.yangtools.yang.data.api.schema.LeafNode;
 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNodes;
 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidateNode;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.ModificationType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
- * Implementation of the BGP import policy. Listens on all Adj-RIB-In, inspects all inbound
+ * Implementation of the BGP import policy. Listens on peer's Adj-RIB-In, inspects all inbound
  * routes in the context of the advertising peer's role and applies the inbound policy.
  *
  * Inbound policy is applied as follows:
@@ -52,200 +61,314 @@ import org.slf4j.LoggerFactory;
  *    advertising peer's role
  * 3) output admitting routes with edited attributes into /bgp-rib/rib/peer/effective-rib-in/tables/routes
  *
- * Note that we maintain the peer roles using a DCL, even if we could look up our internal
- * structures. This is done so we maintain causality and loose coupling.
  */
 @NotThreadSafe
-final class EffectiveRibInWriter {
-    private static final Predicate<PathArgument> IS_PEER = new Predicate<PathArgument>() {
-        @Override
-        public boolean apply(final PathArgument input) {
-            return input.getNodeType().equals(Peer.QNAME);
+final class EffectiveRibInWriter implements AutoCloseable {
+    private static final Logger LOG = LoggerFactory.getLogger(EffectiveRibInWriter.class);
+    protected static final NodeIdentifier TABLE_ROUTES = new NodeIdentifier(Routes.QNAME);
+
+    private final class AdjInTracker implements AutoCloseable, ClusteredDOMDataTreeChangeListener {
+        private final RIBSupportContextRegistry registry;
+        private final YangInstanceIdentifier peerIId;
+        private final YangInstanceIdentifier effRibTables;
+        private final ListenerRegistration<?> reg;
+        private final DOMTransactionChain chain;
+        private final PerTableTypeRouteCounter effectiveRibInRouteCounters;
+        private final PerTableTypeRouteCounter adjRibInRouteCounters;
+        private final Map<TablesKey, Set<YangInstanceIdentifier>> effectiveRibInRouteMap = new ConcurrentHashMap<>();
+        private final Map<TablesKey, Set<YangInstanceIdentifier>> adjRibInRouteMap = new ConcurrentHashMap<>();
+
+        AdjInTracker(final DOMDataTreeChangeService service, final RIBSupportContextRegistry registry, final DOMTransactionChain chain, final YangInstanceIdentifier peerIId,
+                @Nonnull final PerTableTypeRouteCounter effectiveRibInRouteCounters, @Nonnull final PerTableTypeRouteCounter adjRibInRouteCounters) {
+            this.registry = Preconditions.checkNotNull(registry);
+            this.chain = Preconditions.checkNotNull(chain);
+            this.peerIId = Preconditions.checkNotNull(peerIId);
+            this.effRibTables = this.peerIId.node(EffectiveRibIn.QNAME).node(Tables.QNAME);
+            this.effectiveRibInRouteCounters = Preconditions.checkNotNull(effectiveRibInRouteCounters);
+            this.adjRibInRouteCounters = Preconditions.checkNotNull(adjRibInRouteCounters);
+
+            final DOMDataTreeIdentifier treeId = new DOMDataTreeIdentifier(LogicalDatastoreType.OPERATIONAL, this.peerIId.node(AdjRibIn.QNAME).node(Tables.QNAME));
+            LOG.debug("Registered Effective RIB on {}", this.peerIId);
+            this.reg = service.registerDataTreeChangeListener(treeId, this);
         }
-    };
-    private static final Predicate<PathArgument> IS_TABLES = new Predicate<PathArgument>() {
-        @Override
-        public boolean apply(final PathArgument input) {
-            return input.getNodeType().equals(Tables.QNAME);
+
+        /**
+         * @deprecated Should always pass in route counter
+         * @param service
+         * @param registry
+         * @param chain
+         * @param peerIId
+         */
+        @Deprecated
+        AdjInTracker(final DOMDataTreeChangeService service, final RIBSupportContextRegistry registry, final DOMTransactionChain chain, final YangInstanceIdentifier peerIId) {
+            this(service, registry, chain, peerIId, new PerTableTypeRouteCounter("effective-rib-in"), new PerTableTypeRouteCounter("adj-rib-in"));
         }
-    };
-    private static final Logger LOG = LoggerFactory.getLogger(EffectiveRibInWriter.class);
-    private static final QName PEER_ID = QName.create(Peer.QNAME, "peer-id");
 
-    // FIXME: implement as id.firstIdentifierOf(IS_PEER), null indicating not found
-    private static final NodeIdentifierWithPredicates firstKeyOf(final YangInstanceIdentifier id, final Predicate<PathArgument> match) {
-        final PathArgument ret = Iterables.find(id.getPathArguments(), IS_PEER);
-        Preconditions.checkArgument(ret instanceof NodeIdentifierWithPredicates, "Non-key peer identifier %s", ret);
-        return (NodeIdentifierWithPredicates) ret;
-    }
+        private void updateRoute(@Nonnull final PerTableTypeRouteCounter counter, @Nonnull final Map<TablesKey, Set<YangInstanceIdentifier>> routeMap,
+                @Nonnull final TablesKey tablesKey, @Nonnull final YangInstanceIdentifier routeId) {
+            routeMap.putIfAbsent(tablesKey, new HashSet<>());
+            routeMap.get(tablesKey).add(routeId);
 
-    static final NodeIdentifierWithPredicates peerKey(final YangInstanceIdentifier id) {
-        return firstKeyOf(id, IS_PEER);
-    }
+            updateRouteCounter(counter, routeMap,tablesKey);
+        }
 
-    static final PeerId peerId(final NodeIdentifierWithPredicates peerKey) {
-        return (PeerId) peerKey.getKeyValues().get(PEER_ID);
-    }
+        private void deleteRoute(@Nonnull final PerTableTypeRouteCounter counter, @Nonnull final Map<TablesKey, Set<YangInstanceIdentifier>> routeMap,
+                @Nonnull final TablesKey tablesKey, @Nonnull final YangInstanceIdentifier routeId) {
+            if (routeMap.containsKey(tablesKey)) {
+                routeMap.get(tablesKey).remove(routeId);
+            }
 
-    private static final NodeIdentifierWithPredicates tableKey(final YangInstanceIdentifier id) {
-        return firstKeyOf(id, IS_TABLES);
-    }
+            updateRouteCounter(counter, routeMap,tablesKey);
+        }
 
-    /**
-     * Maintains the mapping of PeerId -> Role inside. We are subscribed to our target leaf,
-     * but that is a wildcard:
-     *     /bgp-rib/rib/peer/peer-role
-     *
-     * MD-SAL assumption: we are getting one {@link DataTreeCandidate} for each expanded
-     *                    wildcard path, so are searching for a particular key.
-     */
-    private final class PeerRoleListener implements DOMDataTreeChangeListener {
-        @Override
-        public void onDataTreeChanged(final Collection<DataTreeCandidate> changes) {
-            synchronized (policies) {
-                for (DataTreeCandidate tc : changes) {
-                    // Obtain the peer's key
-                    final NodeIdentifierWithPredicates peerKey = peerKey(tc.getRootPath());
-
-                    // Check for removal
-                    final Optional<NormalizedNode<?, ?>> maybePeerRole = tc.getRootNode().getDataAfter();
-                    if (maybePeerRole.isPresent()) {
-                        final LeafNode<?> peerRoleLeaf = (LeafNode<?>) maybePeerRole.get();
-                        // FIXME: need codec here
-                        final PeerRole peerRole = (PeerRole) peerRoleLeaf.getValue();
-
-                        // Lookup policy based on role
-                        final AbstractImportPolicy policy = AbstractImportPolicy.forRole(peerRole);
-
-                        // Update lookup map
-                        policies.put(peerId(peerKey), policy);
-                    } else {
-                        policies.remove(peerId(peerKey));
-                    }
-                }
-            }
+        private void deleteRoute(@Nonnull final PerTableTypeRouteCounter counter, @Nonnull final Map<TablesKey, Set<YangInstanceIdentifier>> routeMap,
+                @Nonnull final TablesKey tablesKey) {
+            routeMap.remove(tablesKey);
+
+            updateRouteCounter(counter, routeMap,tablesKey);
         }
-    }
 
-    /**
-     * Maintains the individual routes for a particular table's routes under:
-     *     /bgp-rib/rib/peer/adj-rib-in/tables/routes
-     */
-    private final class TableRouteListener implements DOMDataTreeChangeListener {
-        private final NodeIdentifierWithPredicates tableKey;
-        private final YangInstanceIdentifier target;
-        private final RIBSupport ribSupport;
-        private final PeerId peerId;
+        private void updateRouteCounter(@Nonnull final PerTableTypeRouteCounter counter, @Nonnull final Map<TablesKey, Set<YangInstanceIdentifier>> routeMap,
+                @Nonnull final TablesKey tablesKey) {
+            counter.getCounterOrSetDefault(tablesKey)
+            .setCount(routeMap.getOrDefault(tablesKey, new HashSet<>()).size());
+        }
 
-        TableRouteListener(final RIBSupport ribSupport, final NodeIdentifierWithPredicates peerKey, final NodeIdentifierWithPredicates tableKey) {
-            this.ribSupport = Preconditions.checkNotNull(ribSupport);
-            this.tableKey = Preconditions.checkNotNull(tableKey);
+        private void processRoute(final DOMDataWriteTransaction tx, final RIBSupport ribSupport, final AbstractImportPolicy policy, final YangInstanceIdentifier routesPath, final DataTreeCandidateNode route) {
+            LOG.debug("Process route {}", route.getIdentifier());
+            final YangInstanceIdentifier routeId = ribSupport.routePath(routesPath, route.getIdentifier());
+            final TablesKey tablesKey = new TablesKey(ribSupport.getAfi(), ribSupport.getSafi());
+            switch (route.getModificationType()) {
+            case DELETE:
+            case DISAPPEARED:
+                tx.delete(LogicalDatastoreType.OPERATIONAL, routeId);
+                LOG.debug("Route deleted. routeId={}", routeId);
+
+                deleteRoute(this.adjRibInRouteCounters, this.adjRibInRouteMap, tablesKey, routeId);
+                deleteRoute(this.effectiveRibInRouteCounters, this.effectiveRibInRouteMap, tablesKey, routeId);
+                break;
+            case UNMODIFIED:
+                // No-op
+                break;
+            case APPEARED:
+            case SUBTREE_MODIFIED:
+            case WRITE:
+                tx.put(LogicalDatastoreType.OPERATIONAL, routeId, route.getDataAfter().get());
+                // count adj-rib-in route first
+                updateRoute(this.adjRibInRouteCounters, this.adjRibInRouteMap, tablesKey, routeId);
+                updateRoute(this.effectiveRibInRouteCounters, this.effectiveRibInRouteMap, tablesKey, routeId);
+                // Lookup per-table attributes from RIBSupport
+                final ContainerNode advertisedAttrs = (ContainerNode) NormalizedNodes.findNode(route.getDataAfter(), ribSupport.routeAttributesIdentifier()).orNull();
+                final ContainerNode effectiveAttrs;
 
-            // Lookup peer ID
-            this.peerId = (PeerId) Preconditions.checkNotNull(peerKey.getKeyValues().get(PEER_ID));
+                if (advertisedAttrs != null) {
+                    effectiveAttrs = policy.effectiveAttributes(advertisedAttrs);
+                } else {
+                    effectiveAttrs = null;
+                }
 
-            // FIXME: need target table ID
-            target = null;
-        }
+                LOG.debug("Route {} effective attributes {} towards {}", route.getIdentifier(), effectiveAttrs, routeId);
 
-        private void updateRoutes(final DOMDataWriteTransaction tx, final DataTreeCandidateNode routes, final ContainerNode effectiveAttrs) {
-            final YangInstanceIdentifier routeId = target.node(routes.getIdentifier());
+                if (effectiveAttrs != null) {
+                    tx.put(LogicalDatastoreType.OPERATIONAL, routeId.node(ribSupport.routeAttributesIdentifier()), effectiveAttrs);
 
-            if (effectiveAttrs != null) {
-                tx.put(LogicalDatastoreType.OPERATIONAL, routeId, routes.getDataAfter().get());
-                tx.put(LogicalDatastoreType.OPERATIONAL, routeId.node(ribSupport.routeAttributes()), effectiveAttrs);
-            } else if (routes.getDataBefore().isPresent()) {
-                tx.delete(LogicalDatastoreType.OPERATIONAL, routeId);
-            }
+                    updateRoute(this.effectiveRibInRouteCounters, this.effectiveRibInRouteMap, tablesKey, routeId);
+                } else {
+                    LOG.warn("Route {} advertised empty attributes", routeId);
+                    tx.delete(LogicalDatastoreType.OPERATIONAL,  routeId);
 
+                    deleteRoute(this.effectiveRibInRouteCounters, this.effectiveRibInRouteMap, tablesKey, routeId);
+                }
+                break;
+            default:
+                LOG.warn("Ignoring unhandled route {}", route);
+                break;
+            }
         }
 
-        @Override
-        public void onDataTreeChanged(final Collection<DataTreeCandidate> changes) {
-            // FIXME: note that we need to detect table clears efficiently and propagate them
-
-            final DOMDataWriteTransaction tx = chain.newWriteOnlyTransaction();
+        private void processTableChildren(final DOMDataWriteTransaction tx, final RIBSupport ribSupport, final YangInstanceIdentifier tablePath, final Collection<DataTreeCandidateNode> children) {
+            for (final DataTreeCandidateNode child : children) {
+                final PathArgument childIdentifier = child.getIdentifier();
+                final Optional<NormalizedNode<?, ?>> childDataAfter = child.getDataAfter();
+                final TablesKey tablesKey = new TablesKey(ribSupport.getAfi(), ribSupport.getSafi());
+                LOG.debug("Process table {} type {}, dataAfter {}, dataBefore {}", childIdentifier, child
+                        .getModificationType(), childDataAfter, child.getDataBefore());
+                final YangInstanceIdentifier childPath = tablePath.node(childIdentifier);
+                switch (child.getModificationType()) {
+                case DELETE:
+                case DISAPPEARED:
+                    tx.delete(LogicalDatastoreType.OPERATIONAL, childPath);
+                    LOG.debug("Route deleted. routeId={}", childPath);
 
-            for (DataTreeCandidate tc : changes) {
-                // Lookup per-table attributes from RIBSupport
-                final ContainerNode adverisedAttrs = (ContainerNode) NormalizedNodes.findNode(tc.getRootNode().getDataAfter(), ribSupport.routeAttributes()).orNull();
-                final ContainerNode effectiveAttrs;
+                    deleteRoute(this.adjRibInRouteCounters, this.adjRibInRouteMap, tablesKey, childPath);
+                    deleteRoute(this.effectiveRibInRouteCounters, this.effectiveRibInRouteMap, tablesKey, childPath);
+                    break;
+                case UNMODIFIED:
+                    // No-op
+                    break;
+                case SUBTREE_MODIFIED:
+                    processModifiedRouteTables(child, childIdentifier,tx, ribSupport, EffectiveRibInWriter.this.importPolicy, childPath, childDataAfter);
+                    break;
+                case APPEARED:
+                case WRITE:
+                    writeRouteTables(child, childIdentifier,tx, ribSupport, EffectiveRibInWriter.this.importPolicy, childPath, childDataAfter);
 
-                if (adverisedAttrs != null && tc.getRootNode().getDataAfter().isPresent()) {
-                    synchronized (policies) {
-                        final AbstractImportPolicy policy = policies.get(peerId);
-                        effectiveAttrs = policy.effectiveAttributes(adverisedAttrs);
-                    }
-                } else {
-                    effectiveAttrs = null;
+                    break;
+                default:
+                    LOG.warn("Ignoring unhandled child {}", child);
+                    break;
                 }
+            }
+        }
 
-                LOG.debug("Route change {} effective attributes {}", tc.getRootPath(), effectiveAttrs);
+        private void processModifiedRouteTables(final DataTreeCandidateNode child, final PathArgument childIdentifier, final DOMDataWriteTransaction tx,
+                final RIBSupport ribSupport, final AbstractImportPolicy policy, final YangInstanceIdentifier childPath, final Optional<NormalizedNode<?, ?>> childDataAfter) {
+            if (TABLE_ROUTES.equals(childIdentifier)) {
+                for (final DataTreeCandidateNode route : ribSupport.changedRoutes(child)) {
+                    processRoute(tx, ribSupport, policy, childPath, route);
+                }
+            } else {
+                tx.put(LogicalDatastoreType.OPERATIONAL, childPath, childDataAfter.get());
+            }
+        }
 
-                updateRoutes(tx, tc.getRootNode(), effectiveAttrs);
+        private void writeRouteTables(final DataTreeCandidateNode child, final PathArgument childIdentifier, final DOMDataWriteTransaction tx, final RIBSupport ribSupport, final AbstractImportPolicy policy, final YangInstanceIdentifier childPath, final Optional<NormalizedNode<?, ?>> childDataAfter) {
+            if (TABLE_ROUTES.equals(childIdentifier)) {
+                final Collection<DataTreeCandidateNode> changedRoutes = ribSupport.changedRoutes(child);
+                if (!changedRoutes.isEmpty()) {
+                    tx.put(LogicalDatastoreType.OPERATIONAL, childPath, childDataAfter.get());
+                    // Routes are special, as they may end up being filtered. The previous put conveniently
+                    // ensured that we have them in at target, so a subsequent delete will not fail :)
+                    for (final DataTreeCandidateNode route : changedRoutes) {
+                        processRoute(tx, ribSupport, policy, childPath, route);
+                    }
+                }
             }
+        }
 
-            tx.submit();
+        private RIBSupportContext getRibSupport(final NodeIdentifierWithPredicates tableKey) {
+            return this.registry.getRIBSupportContext(tableKey);
         }
-    }
 
-    /**
-     * Maintains {@link TableRouteListener} instances.
-     */
-    private final class TableListener implements DOMDataTreeChangeListener {
-        private final Map<YangInstanceIdentifier, ListenerRegistration<?>> routeListeners = new HashMap<>();
-        private final RIBExtensionConsumerContext registry;
-        private final DOMDataTreeChangeService service;
+        private YangInstanceIdentifier effectiveTablePath(final NodeIdentifierWithPredicates tableKey) {
+            return this.effRibTables.node(tableKey);
+        }
 
-        TableListener(final DOMDataTreeChangeService service, final RIBExtensionConsumerContext registry) {
-            this.registry = Preconditions.checkNotNull(registry);
-            this.service = Preconditions.checkNotNull(service);
+        private void modifyTable(final DOMDataWriteTransaction tx, final NodeIdentifierWithPredicates tableKey, final DataTreeCandidateNode table) {
+            final RIBSupportContext ribSupport = getRibSupport(tableKey);
+            final YangInstanceIdentifier tablePath = effectiveTablePath(tableKey);
+
+            processTableChildren(tx, ribSupport.getRibSupport(), tablePath, table.getChildNodes());
         }
 
-        @Override
-        public void onDataTreeChanged(final Collection<DataTreeCandidate> changes) {
+        private void writeTable(final DOMDataWriteTransaction tx, final NodeIdentifierWithPredicates tableKey, final DataTreeCandidateNode table) {
+            final RIBSupportContext ribSupport = getRibSupport(tableKey);
+            final YangInstanceIdentifier tablePath = effectiveTablePath(tableKey);
 
-            for (DataTreeCandidate tc : changes) {
-                // Obtain the peer's key
-                final NodeIdentifierWithPredicates peerKey = peerKey(tc.getRootPath());
+            // Create an empty table
+            LOG.trace("Create Empty table", tablePath);
+            ribSupport.createEmptyTableStructure(tx, tablePath);
 
-                // Lookup
-                final NodeIdentifierWithPredicates tableKey = tableKey(tc.getRootPath());
+            processTableChildren(tx, ribSupport.getRibSupport(), tablePath, table.getChildNodes());
+        }
 
-                switch (tc.getRootNode().getModificationType()) {
-                case DELETE:
-                    final ListenerRegistration<?> reg = routeListeners.remove(tc.getRootPath());
-                    if (reg != null) {
-                        reg.close();
-                    }
-                    break;
-                case WRITE:
-                    // FIXME: use codec to translate
-                    final RIBSupport ribSupport = registry.getRIBSupport(null);
-                    if (ribSupport != null) {
-                        final TableRouteListener routeListener = new TableRouteListener(ribSupport, peerKey, tableKey);
-                        final ListenerRegistration<?> r = service.registerDataTreeChangeListener(
-                            new DOMDataTreeIdentifier(LogicalDatastoreType.OPERATIONAL,  tc.getRootPath()), routeListener);
-
-                        routeListeners.put(tc.getRootPath(), r);
-                    } else {
-                        LOG.warn("No RIB support for table {}, ignoring advertisements from peer %s", tableKey, peerKey);
+        @Override
+        public void onDataTreeChanged(@Nonnull final Collection<DataTreeCandidate> changes) {
+            LOG.trace("Data changed called to effective RIB. Change : {}", changes);
+
+            // we have a lot of transactions created for 'nothing' because a lot of changes
+            // are skipped, so ensure we only create one transaction when we really need it
+            DOMDataWriteTransaction tx = null;
+            for (final DataTreeCandidate tc : changes) {
+                final YangInstanceIdentifier rootPath = tc.getRootPath();
+
+                final DataTreeCandidateNode root = tc.getRootNode();
+                for (final DataTreeCandidateNode table : root.getChildNodes()) {
+                    if (tx == null) {
+                        tx = this.chain.newWriteOnlyTransaction();
                     }
-                    break;
-                default:
-                    // No-op
-                    break;
+                    changeDataTree(tx, rootPath, root, table);
                 }
             }
+            if (tx != null) {
+                tx.submit();
+            }
+        }
+
+        private void changeDataTree(final DOMDataWriteTransaction tx, final YangInstanceIdentifier rootPath,
+                final DataTreeCandidateNode root, final DataTreeCandidateNode table) {
+            final PathArgument lastArg = table.getIdentifier();
+            Verify.verify(lastArg instanceof NodeIdentifierWithPredicates, "Unexpected type %s in path %s", lastArg.getClass(), rootPath);
+            final NodeIdentifierWithPredicates tableKey = (NodeIdentifierWithPredicates) lastArg;
+            final RIBSupport ribSupport = getRibSupport(tableKey).getRibSupport();
+            final ModificationType modificationType = root.getModificationType();
+            switch (modificationType) {
+            case DELETE:
+            case DISAPPEARED:
+                final YangInstanceIdentifier effectiveTablePath = effectiveTablePath(tableKey);
+                LOG.debug("Delete Effective Table {} modification type {}, ", effectiveTablePath, modificationType);
+
+                // delete the corresponding effective table
+                tx.delete(LogicalDatastoreType.OPERATIONAL, effectiveTablePath);
+
+                deleteRoute(this.adjRibInRouteCounters, this.adjRibInRouteMap, new TablesKey(ribSupport.getAfi(), ribSupport.getSafi()));
+                deleteRoute(this.effectiveRibInRouteCounters, this.effectiveRibInRouteMap, new TablesKey(ribSupport.getAfi(), ribSupport.getSafi()));
+                break;
+            case SUBTREE_MODIFIED:
+                modifyTable(tx, tableKey, table);
+                break;
+            case UNMODIFIED:
+                LOG.info("Ignoring spurious notification on {} data {}", rootPath, table);
+                break;
+            case APPEARED:
+            case WRITE:
+                writeTable(tx, tableKey, table);
+                break;
+            default:
+                LOG.warn("Ignoring unhandled root {}", root);
+                break;
+            }
         }
+
+        @Override
+        public void close() {
+            this.reg.close();
+        }
+    }
+
+    private final AdjInTracker adjInTracker;
+    private final AbstractImportPolicy importPolicy;
+
+    @Deprecated
+    static EffectiveRibInWriter create(@Nonnull final DOMDataTreeChangeService service, @Nonnull final DOMTransactionChain chain,
+            @Nonnull final YangInstanceIdentifier peerIId, @Nonnull final ImportPolicyPeerTracker importPolicyPeerTracker, @Nonnull final RIBSupportContextRegistry registry, final PeerRole peerRole) {
+        return new EffectiveRibInWriter(service, chain, peerIId, importPolicyPeerTracker, registry, peerRole);
     }
 
-    private final Map<PeerId, AbstractImportPolicy> policies = new HashMap<>();
-    private final DOMTransactionChain chain;
+    static EffectiveRibInWriter create(@Nonnull final DOMDataTreeChangeService service, @Nonnull final DOMTransactionChain chain,
+            @Nonnull final YangInstanceIdentifier peerIId, @Nonnull final ImportPolicyPeerTracker importPolicyPeerTracker, @Nonnull final RIBSupportContextRegistry registry, final PeerRole peerRole,
+            @Nonnull final PerTableTypeRouteCounter effectiveRouteCounters, @Nonnull final PerTableTypeRouteCounter adjRibInRouteCounters) {
+        return new EffectiveRibInWriter(service, chain, peerIId, importPolicyPeerTracker, registry, peerRole, effectiveRouteCounters, adjRibInRouteCounters);
+    }
 
-    private EffectiveRibInWriter(final DOMTransactionChain chain) {
-        this.chain = Preconditions.checkNotNull(chain);
+    @Deprecated
+    private EffectiveRibInWriter(final DOMDataTreeChangeService service, final DOMTransactionChain chain, final YangInstanceIdentifier peerIId,
+            final ImportPolicyPeerTracker importPolicyPeerTracker, final RIBSupportContextRegistry registry, final PeerRole peerRole) {
+        importPolicyPeerTracker.peerRoleChanged(peerIId, peerRole);
+        this.importPolicy = importPolicyPeerTracker.policyFor(IdentifierUtils.peerId((NodeIdentifierWithPredicates) peerIId.getLastPathArgument()));
+        this.adjInTracker = new AdjInTracker(service, registry, chain, peerIId);
+    }
+
+    private EffectiveRibInWriter(final DOMDataTreeChangeService service, final DOMTransactionChain chain, final YangInstanceIdentifier peerIId,
+            final ImportPolicyPeerTracker importPolicyPeerTracker, final RIBSupportContextRegistry registry, final PeerRole peerRole,
+            @Nonnull final PerTableTypeRouteCounter effectiveRouteCounters, @Nonnull final PerTableTypeRouteCounter adjRibInRouteCounters) {
+        importPolicyPeerTracker.peerRoleChanged(peerIId, peerRole);
+        this.importPolicy = importPolicyPeerTracker.policyFor(IdentifierUtils.peerId((NodeIdentifierWithPredicates) peerIId.getLastPathArgument()));
+        this.adjInTracker = new AdjInTracker(service, registry, chain, peerIId, effectiveRouteCounters, adjRibInRouteCounters);
+    }
 
-        // FIXME: subscribe peerRoleListener, tableListener
+    @Override
+    public void close() {
+        this.adjInTracker.close();
     }
 }