import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; //导入依赖的package包/类
public void init(final YangInstanceIdentifier path, final AsyncDataBroker.DataChangeScope scope) {

        dataChangeListenerActor = actorContext.getActorSystem().actorOf(
                DataChangeListener.props(listener, path).withDispatcher(actorContext.getNotificationDispatcherPath()));

        Future<ActorRef> findFuture = actorContext.findLocalShardAsync(shardName);
        findFuture.onComplete(new OnComplete<ActorRef>() {
            @Override
            public void onComplete(Throwable failure, ActorRef shard) {
                if (failure instanceof LocalShardNotFoundException) {
                    LOG.debug("No local shard found for {} - DataChangeListener {} at path {} "
                            + "cannot be registered", shardName, listener, path);
                } else if (failure != null) {
                    LOG.error("Failed to find local shard {} - DataChangeListener {} at path {} "
                            + "cannot be registered: {}", shardName, listener, path, failure);
                } else {
                    doRegistration(shard, path, scope);
                }
            }
        }, actorContext.getClientDispatcher());
    }
 

import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; //导入依赖的package包/类
private void doRegistration(ActorRef shard, final YangInstanceIdentifier path,
        DataChangeScope scope) {

    Future<Object> future = actorContext.executeOperationAsync(shard,
            new RegisterChangeListener(path, dataChangeListenerActor, scope,
                listener instanceof ClusteredDOMDataChangeListener),
            actorContext.getDatastoreContext().getShardInitializationTimeout());

    future.onComplete(new OnComplete<Object>() {
        @Override
        public void onComplete(Throwable failure, Object result) {
            if (failure != null) {
                LOG.error("Failed to register DataChangeListener {} at path {}",
                        listener, path.toString(), failure);
            } else {
                RegisterDataTreeNotificationListenerReply reply = (RegisterDataTreeNotificationListenerReply)result;
                setListenerRegistrationActor(actorContext.actorSelection(
                        reply.getListenerRegistrationPath()));
            }
        }
    }, actorContext.getClientDispatcher());
}
 

import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; //导入依赖的package包/类
@Test
public void testChangeListenerWithNoInitialData() throws Exception {

    new ShardTestKit(getSystem()) {
        {
            final TestActorRef<Shard> actor = actorFactory.createTestActor(
                    newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
                    "testChangeListenerWithNoInitialData");

            waitUntilLeader(actor);

            final Shard shard = actor.underlyingActor();
            final MockDataChangeListener listener = new MockDataChangeListener(0);
            final ActorRef dclActor = actorFactory.createActor(DataChangeListener.props(listener, TEST_PATH),
                    "testChangeListenerWithNoInitialData-DataChangeListener");
            final DataChangeListenerSupport support = new DataChangeListenerSupport(shard);
            support.onMessage(new RegisterChangeListener(TEST_PATH, dclActor, DataChangeScope.ONE, false),
                    true,true);

            listener.expectNoMoreChanges("Unexpected initial change event");
        }
    };
}
 

import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; //导入依赖的package包/类
@Test
public void testInitialChangeListenerEventWithContainerPath() throws Exception {

    new ShardTestKit(getSystem()) {
        {
            final TestActorRef<Shard> actor = actorFactory.createTestActor(
                    newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
                    "testInitialChangeListenerEventWithContainerPath");

            waitUntilLeader(actor);

            final Shard shard = actor.underlyingActor();
            writeToStore(shard.getDataStore(), TestModel.TEST_PATH,
                    ImmutableNodes.containerNode(TestModel.TEST_QNAME));
            final MockDataChangeListener listener = new MockDataChangeListener(1);
            final ActorRef dclActor = actorFactory.createActor(DataChangeListener.props(listener, TEST_PATH),
                    "testInitialChangeListenerEventWithContainerPath-DataChangeListener");
            final DataChangeListenerSupport support = new DataChangeListenerSupport(shard);
            support.onMessage(new RegisterChangeListener(TEST_PATH, dclActor, DataChangeScope.ONE, false),
                    true,true);

            listener.waitForChangeEvents(TEST_PATH);
        }
    };
}
 

import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; //导入依赖的package包/类
@Test
public void testSeparateWrites() throws InterruptedException, TimeoutException, ExecutionException {

    DataBroker dataBroker = testContext.getDataBroker();

    final SettableFuture<AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject>> eventFuture =
            SettableFuture.create();
    dataBroker.registerDataChangeListener(LogicalDatastoreType.OPERATIONAL, DEEP_WILDCARDED_PATH,
        dataChangeEvent -> eventFuture.set(dataChangeEvent), DataChangeScope.SUBTREE);

    final WriteTransaction transaction = dataBroker.newWriteOnlyTransaction();
    transaction.put(LogicalDatastoreType.OPERATIONAL, NODE_0_CWU_PATH, CWU, true);
    transaction.put(LogicalDatastoreType.OPERATIONAL, NODE_0_LVU_PATH, LVU, true);
    transaction.put(LogicalDatastoreType.OPERATIONAL, NODE_1_LVU_PATH, LVU, true);
    transaction.submit().get(5, TimeUnit.SECONDS);

    AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> event = eventFuture.get(1000, TimeUnit.MILLISECONDS);

    validateEvent(event);
}
 

import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; //导入依赖的package包/类
@Test
public void deleteChildListenParent() throws InterruptedException, ExecutionException, TimeoutException {
    DataBroker dataBroker = testContext.getDataBroker();
    final WriteTransaction initTx = dataBroker.newWriteOnlyTransaction();

    initTx.put(LogicalDatastoreType.OPERATIONAL, LIST11_PATH, createList11(), true);
    initTx.submit().get(5, TimeUnit.SECONDS);

    final SettableFuture<AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject>> event = SettableFuture.create();
    dataBroker.registerDataChangeListener(LogicalDatastoreType.OPERATIONAL, LIST11_PATH,
        change -> event.set(change), DataChangeScope.SUBTREE);

    final WriteTransaction deleteTx = dataBroker.newWriteOnlyTransaction();
    deleteTx.delete(LogicalDatastoreType.OPERATIONAL, LIST11_PATH.augmentation(List11SimpleAugment.class));
    deleteTx.submit().get(5, TimeUnit.SECONDS);

    AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> receivedEvent = event.get();
    assertFalse(receivedEvent.getRemovedPaths().contains(TLL_COMPLEX_AUGMENT_PATH));
}
 

import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; //导入依赖的package包/类
@Test
public void writeTopWithListItemAugmentedListenTopSubtree() {
    TestListener listener = createListener(CONFIGURATION,TOP_PATH, DataChangeScope.SUBTREE, false);

    writeTopWithListItem(CONFIGURATION);

    AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> event = listener.event();

    assertContains(event.getCreatedData(), TOP_PATH);
    assertContains(event.getCreatedData(), path(TOP_FOO_KEY));
    assertContains(event.getCreatedData(), path(TOP_FOO_KEY, TreeComplexUsesAugment.class));
    assertContains(event.getCreatedData(), path(TOP_FOO_KEY, USES_ONE_KEY));
    assertContains(event.getCreatedData(), path(TOP_FOO_KEY, USES_TWO_KEY));

    assertEmpty(event.getUpdatedData());
    assertEmpty(event.getRemovedPaths());
}
 

import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; //导入依赖的package包/类
@Test
public void writeTopWithListItemAugmentedListenAugmentSubtreeWildcarded() {
    TestListener listener = createListener(CONFIGURATION,AUGMENT_WILDCARD, DataChangeScope.SUBTREE, false);
    writeTopWithListItem(CONFIGURATION);

    AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> event = listener.event();

    /*
     * Event should not contain parent nodes
     */
    assertNotContains(event.getCreatedData(), TOP_PATH, path(TOP_FOO_KEY));

    assertContains(event.getCreatedData(), path(TOP_FOO_KEY, TreeComplexUsesAugment.class));
    assertContains(event.getCreatedData(), path(TOP_FOO_KEY, USES_ONE_KEY));
    assertContains(event.getCreatedData(), path(TOP_FOO_KEY, USES_TWO_KEY));

    assertEmpty(event.getUpdatedData());
    assertEmpty(event.getRemovedPaths());
}
 

import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; //导入依赖的package包/类
@Test
public void deleteAugmentChildListenTopSubtree() {
    writeTopWithListItem(CONFIGURATION);
    TestListener listener = createListener(CONFIGURATION, TOP_PATH, DataChangeScope.SUBTREE);
    InstanceIdentifier<?> deletePath = path(TOP_FOO_KEY,USES_ONE_KEY);
    deleteItem(CONFIGURATION,deletePath);

    AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> event = listener.event();


    assertEmpty(event.getCreatedData());

    assertContains(event.getRemovedPaths(), deletePath);

    assertContains(event.getUpdatedData(), TOP_PATH);
    assertContains(event.getUpdatedData(), path(TOP_FOO_KEY));
    assertContains(event.getUpdatedData(), path(TOP_FOO_KEY, TreeComplexUsesAugment.class));

    assertNotContains(event.getCreatedData(), path(TOP_FOO_KEY, USES_TWO_KEY));
}
 

import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; //导入依赖的package包/类
@Test
public void deleteAugmentChildListenAugmentSubtreeWildcarded() {
    writeTopWithListItem(CONFIGURATION);

    TestListener listener = createListener(CONFIGURATION, AUGMENT_WILDCARD, DataChangeScope.SUBTREE, false);
    InstanceIdentifier<?> deletePath = path(TOP_FOO_KEY,USES_ONE_KEY);
    deleteItem(CONFIGURATION,deletePath);
    AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> event = listener.event();

    assertEmpty(event.getCreatedData());

    assertContains(event.getUpdatedData(), path(TOP_FOO_KEY, TreeComplexUsesAugment.class));

    /*
     * Event should not contain parent nodes
     */
    assertNotContains(event.getUpdatedData(), TOP_PATH, path(TOP_FOO_KEY));

    assertContains(event.getRemovedPaths(), deletePath);
}
 

import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; //导入依赖的package包/类
@Override
public <L extends AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>> ListenerRegistration<L> registerChangeListener(
        final YangInstanceIdentifier path, final L listener, final DataChangeScope scope) {

    /*
     * Make sure commit is not occurring right now. Listener has to be
     * registered and its state capture enqueued at a consistent point.
     *
     * FIXME: improve this to read-write lock, such that multiple listener
     * registrations can occur simultaneously
     */
    final DataChangeListenerRegistration<L> reg;
    synchronized (this) {
        LOG.debug("{}: Registering data change listener {} for {}", name, listener, path);

        reg = listenerTree.registerDataChangeListener(path, listener, scope);

        Optional<NormalizedNode<?, ?>> currentState = dataTree.takeSnapshot().readNode(path);
        if (currentState.isPresent()) {
            final NormalizedNode<?, ?> data = currentState.get();

            final DOMImmutableDataChangeEvent event = DOMImmutableDataChangeEvent.builder(DataChangeScope.BASE) //
                    .setAfter(data) //
                    .addCreated(path, data) //
                    .build();

            dataChangeListenerNotificationManager.submitNotification(reg, event);
        }
    }

    return new AbstractListenerRegistration<L>(listener) {
        @Override
        protected void removeRegistration() {
            synchronized (InMemoryDOMDataStore.this) {
                reg.close();
            }
        }
    };
}
 

import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; //导入依赖的package包/类
@Override
public DOMImmutableDataChangeEvent create(final YangInstanceIdentifier path, final NormalizedNode<PathArgument, ?> data) {
    return builder(DataChangeScope.BASE) //
            .setBefore(data) //
            .addRemoved(path, data) //
            .build();
}
 

import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; //导入依赖的package包/类
private ResolveDataChangeState(final YangInstanceIdentifier nodeId,
        final Iterable<Builder> inheritedSub, final Collection<Builder> inheritedOne,
        final Collection<RegistrationTreeNode<DataChangeListenerRegistration<?>>> nodes) {
    this.nodeId = Preconditions.checkNotNull(nodeId);
    this.nodes = Preconditions.checkNotNull(nodes);
    this.inheritedSub = Preconditions.checkNotNull(inheritedSub);
    this.inheritedOne = Preconditions.checkNotNull(inheritedOne);

    /*
     * Collect the nodes which need to be propagated from us to the child.
     */
    final Map<DataChangeListenerRegistration<?>, Builder> sub = new HashMap<>();
    final Map<DataChangeListenerRegistration<?>, Builder> one = new HashMap<>();
    final Map<DataChangeListenerRegistration<?>, Builder> base = new HashMap<>();
    for (RegistrationTreeNode<DataChangeListenerRegistration<?>> n : nodes) {
        for (DataChangeListenerRegistration<?> l : n.getRegistrations()) {
            final Builder b = DOMImmutableDataChangeEvent.builder(DataChangeScope.BASE);
            switch (l.getScope()) {
            case BASE:
                base.put(l, b);
                break;
            case ONE:
                one.put(l, b);
                break;
            case SUBTREE:
                sub.put(l, b);
                break;
            }
        }
    }

    baseBuilders = maybeEmpty(base);
    oneBuilders = maybeEmpty(one);
    subBuilders = maybeEmpty(sub);
}
 

import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; //导入依赖的package包/类
/**
 * Resolves and submits notification tasks to the specified manager.
 */
public synchronized void resolve(final NotificationManager<DataChangeListenerRegistration<?>, DOMImmutableDataChangeEvent> manager) {
    try (final RegistrationTreeSnapshot<DataChangeListenerRegistration<?>> w = listenerRoot.takeSnapshot()) {
        // Defensive: reset internal state
        collectedEvents = ArrayListMultimap.create();

        // Run through the tree
        final ResolveDataChangeState s = ResolveDataChangeState.initial(candidate.getRootPath(), w.getRootNode());
        resolveAnyChangeEvent(s, candidate.getRootNode());

        /*
         * Convert to tasks, but be mindful of multiple values -- those indicate multiple
         * wildcard matches, which need to be merged.
         */
        for (Entry<DataChangeListenerRegistration<?>, Collection<DOMImmutableDataChangeEvent>> e : collectedEvents.asMap().entrySet()) {
            final Collection<DOMImmutableDataChangeEvent> col = e.getValue();
            final DOMImmutableDataChangeEvent event;

            if (col.size() != 1) {
                final Builder b = DOMImmutableDataChangeEvent.builder(DataChangeScope.BASE);
                for (DOMImmutableDataChangeEvent i : col) {
                    b.merge(i);
                }

                event = b.build();
                LOG.trace("Merged events {} into event {}", col, event);
            } else {
                event = col.iterator().next();
            }

            manager.submitNotification(e.getKey(), event);
        }
    }
}
 

import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; //导入依赖的package包/类
private boolean resolveReplacedEvent(final ResolveDataChangeState state,
        final NormalizedNode<?, ?> beforeData, final NormalizedNode<?, ?> afterData) {

    if (beforeData instanceof NormalizedNodeContainer<?, ?, ?>) {
        /*
         * Node is a container (contains a child) and we have interested
         * listeners registered for it, that means we need to do
         * resolution of changes on children level and can not
         * shortcut resolution.
         */
        LOG.trace("Resolving subtree replace event for {} before {}, after {}", state.getPath(), beforeData, afterData);
        @SuppressWarnings("unchecked")
        NormalizedNodeContainer<?, PathArgument, NormalizedNode<PathArgument, ?>> beforeCont = (NormalizedNodeContainer<?, PathArgument, NormalizedNode<PathArgument, ?>>) beforeData;
        @SuppressWarnings("unchecked")
        NormalizedNodeContainer<?, PathArgument, NormalizedNode<PathArgument, ?>> afterCont = (NormalizedNodeContainer<?, PathArgument, NormalizedNode<PathArgument, ?>>) afterData;
        return resolveNodeContainerReplaced(state, beforeCont, afterCont);
    }

    // Node is a Leaf type (does not contain child nodes)
    // so normal equals method is sufficient for determining change.
    if (beforeData.equals(afterData)) {
        LOG.trace("Skipping equal leaf {}", state.getPath());
        return false;
    }

    LOG.trace("Resolving leaf replace event for {} , before {}, after {}", state.getPath(), beforeData, afterData);
    DOMImmutableDataChangeEvent event = DOMImmutableDataChangeEvent.builder(DataChangeScope.BASE).addUpdated(state.getPath(), beforeData, afterData).build();
    state.addEvent(event);
    state.collectEvents(beforeData, afterData, collectedEvents);
    return true;
}
 

import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; //导入依赖的package包/类
@Override
public void registerDataChangeListener(YangInstanceIdentifier path,
        AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> listener, DataChangeScope scope,
        Optional<DataTreeCandidate> initialState,
        Consumer<ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>>
            onRegistration) {
    publisherActor().tell(new ShardDataChangePublisherActor.RegisterListener(path, listener, scope, initialState,
            onRegistration), ActorRef.noSender());
}
 

import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; //导入依赖的package包/类
@Override
public void registerDataChangeListener(YangInstanceIdentifier path,
        AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> listener, DataChangeScope scope,
        Optional<DataTreeCandidate> initialState,
        Consumer<ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>>
                onRegistration) {
    final DataChangeListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>
            registration = dataChangeListenerTree.registerDataChangeListener(path, listener, scope);

    onRegistration.accept(registration);

    if (initialState.isPresent()) {
        notifySingleListener(path, listener, scope, initialState.get(), logContext);
    }
}
 

import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; //导入依赖的package包/类
static void notifySingleListener(final YangInstanceIdentifier path,
        final AsyncDataChangeListener<YangInstanceIdentifier,NormalizedNode<?, ?>> listener,
        final DataChangeScope scope, final DataTreeCandidate initialState, String logContext) {
    DefaultShardDataChangeListenerPublisher publisher = new DefaultShardDataChangeListenerPublisher(logContext);
    publisher.registerDataChangeListener(path, listener, scope, Optional.absent(), noop -> { /* NOOP */ });
    publisher.publishChanges(initialState);
}
 

import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; //导入依赖的package包/类
RegisterListener(final YangInstanceIdentifier path,
        final AsyncDataChangeListener<YangInstanceIdentifier,NormalizedNode<?, ?>> listener,
        final DataChangeScope scope, final Optional<DataTreeCandidate> initialState,
        final Consumer<ListenerRegistration<
            AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>> onRegistration) {
    this.path = Preconditions.checkNotNull(path);
    this.listener = Preconditions.checkNotNull(listener);
    this.scope = Preconditions.checkNotNull(scope);
    this.initialState = Preconditions.checkNotNull(initialState);
    this.onRegistration = Preconditions.checkNotNull(onRegistration);
}
 

import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; //导入依赖的package包/类
void registerDataChangeListener(final YangInstanceIdentifier path,
        final AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> listener,
        final DataChangeScope scope, final Optional<DataTreeCandidate> initialState,
        final Consumer<ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>>
                onRegistration) {
    dataChangeListenerPublisher.registerDataChangeListener(path, listener, scope, initialState, onRegistration);
}