+++ /dev/null
-/*
- * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.md.sal.dom.store.impl;
-
-import com.google.common.base.Preconditions;
-
-import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent;
-import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
-import org.opendaylight.yangtools.util.concurrent.NotificationManager;
-import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
-import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-class ChangeListenerNotifyTask implements Runnable {
- private static final Logger LOG = LoggerFactory.getLogger(ChangeListenerNotifyTask.class);
-
- @SuppressWarnings("rawtypes")
- private final NotificationManager<AsyncDataChangeListener,AsyncDataChangeEvent> notificationMgr;
- private final AsyncDataChangeEvent<YangInstanceIdentifier, NormalizedNode<?, ?>> event;
- private final DataChangeListenerRegistration<?> listener;
-
- @SuppressWarnings("rawtypes")
- public ChangeListenerNotifyTask(final DataChangeListenerRegistration<?> listener,
- final AsyncDataChangeEvent<YangInstanceIdentifier, NormalizedNode<?, ?>> event,
- final NotificationManager<AsyncDataChangeListener,AsyncDataChangeEvent> notificationMgr) {
- this.notificationMgr = Preconditions.checkNotNull(notificationMgr);
- this.listener = Preconditions.checkNotNull(listener);
- this.event = Preconditions.checkNotNull(event);
- }
-
- @Override
- public void run() {
- final AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> l = listener.getInstance();
- if (l == null) {
- LOG.trace("Skipping event delivery to unregistered listener {}", l);
- return;
- }
- LOG.trace("Listener {} event {}", l, event);
-
- // FIXME: Yo dawg I heard you like queues, so this was queued to be queued
- notificationMgr.submitNotification(l, event);
- }
-
- @Override
- public String toString() {
- return "ChangeListenerNotifyTask [listener=" + listener + ", event=" + event + "]";
- }
-}
*/
package org.opendaylight.controller.md.sal.dom.store.impl;
+import static com.google.common.base.Preconditions.checkState;
+
import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.annotation.concurrent.GuardedBy;
+
import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
-import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent;
import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
import org.opendaylight.controller.md.sal.common.api.data.OptimisticLockFailedException;
import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
import org.opendaylight.controller.md.sal.dom.store.impl.SnapshotBackedWriteTransaction.TransactionReadyPrototype;
-import org.opendaylight.yangtools.util.ExecutorServiceUtil;
-import org.opendaylight.yangtools.util.concurrent.NotificationManager;
-import org.opendaylight.yangtools.util.concurrent.QueuedNotificationManager;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.ConflictingModificationAppliedException;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException;
import org.opendaylight.controller.md.sal.dom.store.impl.tree.ListenerTree;
-import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory;
import org.opendaylight.controller.sal.core.spi.data.DOMStore;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
import org.opendaylight.yangtools.concepts.AbstractListenerRegistration;
import org.opendaylight.yangtools.concepts.Identifiable;
import org.opendaylight.yangtools.concepts.ListenerRegistration;
+import org.opendaylight.yangtools.util.ExecutorServiceUtil;
+import org.opendaylight.yangtools.util.concurrent.NotificationManager;
+import org.opendaylight.yangtools.util.concurrent.QueuedNotificationManager;
+import org.opendaylight.yangtools.util.concurrent.QueuedNotificationManager.Invoker;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.ConflictingModificationAppliedException;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException;
+import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
import org.opendaylight.yangtools.yang.model.api.SchemaContextListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.annotation.concurrent.GuardedBy;
-
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
-
-import static com.google.common.base.Preconditions.checkState;
-
/**
* In-memory DOM Data Store
*
TransactionReadyPrototype,AutoCloseable {
private static final Logger LOG = LoggerFactory.getLogger(InMemoryDOMDataStore.class);
- @SuppressWarnings("rawtypes")
- private static final QueuedNotificationManager.Invoker<AsyncDataChangeListener,
- AsyncDataChangeEvent> DCL_NOTIFICATION_MGR_INVOKER =
- new QueuedNotificationManager.Invoker<AsyncDataChangeListener,
- AsyncDataChangeEvent>() {
-
- @SuppressWarnings("unchecked")
+ private static final Invoker<DataChangeListenerRegistration<?>, DOMImmutableDataChangeEvent> DCL_NOTIFICATION_MGR_INVOKER =
+ new Invoker<DataChangeListenerRegistration<?>, DOMImmutableDataChangeEvent>() {
@Override
- public void invokeListener( AsyncDataChangeListener listener,
- AsyncDataChangeEvent notification ) {
- listener.onDataChanged(notification);
+ public void invokeListener(final DataChangeListenerRegistration<?> listener,
+ final DOMImmutableDataChangeEvent notification ) {
+ final AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> inst = listener.getInstance();
+ if (inst != null) {
+ inst.onDataChanged(notification);
+ }
}
};
private final AtomicLong txCounter = new AtomicLong(0);
private final ListeningExecutorService listeningExecutor;
- @SuppressWarnings("rawtypes")
- private final NotificationManager<AsyncDataChangeListener,AsyncDataChangeEvent>
- dataChangeListenerNotificationManager;
+ private final NotificationManager<DataChangeListenerRegistration<?>, DOMImmutableDataChangeEvent> dataChangeListenerNotificationManager;
private final ExecutorService dataChangeListenerExecutor;
private final String name;
}
public InMemoryDOMDataStore(final String name, final ListeningExecutorService listeningExecutor,
- final ExecutorService dataChangeListenerExecutor, int maxDataChangeListenerQueueSize) {
+ final ExecutorService dataChangeListenerExecutor, final int maxDataChangeListenerQueueSize) {
this.name = Preconditions.checkNotNull(name);
this.listeningExecutor = Preconditions.checkNotNull(listeningExecutor);
-
this.dataChangeListenerExecutor = Preconditions.checkNotNull(dataChangeListenerExecutor);
dataChangeListenerNotificationManager =
.addCreated(path, data) //
.build();
- new ChangeListenerNotifyTask(reg, event,
- dataChangeListenerNotificationManager).run();
+ dataChangeListenerNotificationManager.submitNotification(reg, event);
}
}
}
public synchronized void onTransactionCommited(final SnapshotBackedWriteTransaction transaction) {
- // If commited transaction is latestOutstandingTx we clear
+ // If committed transaction is latestOutstandingTx we clear
// latestOutstandingTx
// field in order to base new transactions on Datastore Data Tree
// directly.
@Override
public Void call() {
candidate = dataTree.prepare(modification);
- listenerResolver = ResolveDataChangeEventsTask.create(candidate, listenerTree,
- dataChangeListenerNotificationManager);
+ listenerResolver = ResolveDataChangeEventsTask.create(candidate, listenerTree);
return null;
}
});
*/
synchronized (this) {
dataTree.commit(candidate);
-
- for (ChangeListenerNotifyTask task : listenerResolver.call()) {
- LOG.trace("Scheduling invocation of listeners: {}", task);
- task.run();
- }
+ listenerResolver.resolve(dataChangeListenerNotificationManager);
}
return Futures.immediateFuture(null);
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Multimap;
-import java.util.ArrayList;
import java.util.Collection;
import java.util.Map.Entry;
-import java.util.concurrent.Callable;
import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
-import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent;
-import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
import org.opendaylight.controller.md.sal.dom.store.impl.DOMImmutableDataChangeEvent.Builder;
import org.opendaylight.controller.md.sal.dom.store.impl.DOMImmutableDataChangeEvent.SimpleEventFactory;
import org.opendaylight.controller.md.sal.dom.store.impl.tree.ListenerTree;
* Computes data change events for all affected registered listeners in data
* tree.
*/
-final class ResolveDataChangeEventsTask implements Callable<Iterable<ChangeListenerNotifyTask>> {
+final class ResolveDataChangeEventsTask {
private static final Logger LOG = LoggerFactory.getLogger(ResolveDataChangeEventsTask.class);
- @SuppressWarnings("rawtypes")
- private final NotificationManager<AsyncDataChangeListener, AsyncDataChangeEvent> notificationMgr;
private final DataTreeCandidate candidate;
private final ListenerTree listenerRoot;
private Multimap<DataChangeListenerRegistration<?>, DOMImmutableDataChangeEvent> collectedEvents;
- @SuppressWarnings("rawtypes")
- public ResolveDataChangeEventsTask(final DataTreeCandidate candidate, final ListenerTree listenerTree,
- final NotificationManager<AsyncDataChangeListener, AsyncDataChangeEvent> notificationMgr) {
+ public ResolveDataChangeEventsTask(final DataTreeCandidate candidate, final ListenerTree listenerTree) {
this.candidate = Preconditions.checkNotNull(candidate);
this.listenerRoot = Preconditions.checkNotNull(listenerTree);
- this.notificationMgr = Preconditions.checkNotNull(notificationMgr);
}
/**
- * Resolves and creates Notification Tasks
- *
- * Implementation of done as Map-Reduce with two steps: 1. resolving events
- * and their mapping to listeners 2. merging events affecting same listener
- *
- * @return An {@link Iterable} of Notification Tasks which needs to be executed in
- * order to delivery data change events.
+ * Resolves and submits notification tasks to the specified manager.
*/
- @Override
- public synchronized Iterable<ChangeListenerNotifyTask> call() {
+ public synchronized void resolve(final NotificationManager<DataChangeListenerRegistration<?>, DOMImmutableDataChangeEvent> manager) {
try (final Walker w = listenerRoot.getWalker()) {
// Defensive: reset internal state
collectedEvents = ArrayListMultimap.create();
* Convert to tasks, but be mindful of multiple values -- those indicate multiple
* wildcard matches, which need to be merged.
*/
- final Collection<ChangeListenerNotifyTask> ret = new ArrayList<>();
for (Entry<DataChangeListenerRegistration<?>, Collection<DOMImmutableDataChangeEvent>> e : collectedEvents.asMap().entrySet()) {
final Collection<DOMImmutableDataChangeEvent> col = e.getValue();
final DOMImmutableDataChangeEvent event;
event = col.iterator().next();
}
- ret.add(new ChangeListenerNotifyTask(e.getKey(), event, notificationMgr));
+ manager.submitNotification(e.getKey(), event);
}
-
- // FIXME: so now we have tasks to submit tasks... Inception-style!
- LOG.debug("Created tasks {}", ret);
- return ret;
}
}
return scope != null;
}
- @SuppressWarnings("rawtypes")
- public static ResolveDataChangeEventsTask create(final DataTreeCandidate candidate,
- final ListenerTree listenerTree,
- final NotificationManager<AsyncDataChangeListener,AsyncDataChangeEvent> notificationMgr) {
- return new ResolveDataChangeEventsTask(candidate, listenerTree, notificationMgr);
+ public static ResolveDataChangeEventsTask create(final DataTreeCandidate candidate, final ListenerTree listenerTree) {
+ return new ResolveDataChangeEventsTask(candidate, listenerTree);
}
}
Code Review / controller.git / commitdiff