import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
-import org.opendaylight.controller.md.sal.dom.store.impl.tree.ListenerRegistrationNode;
-import org.opendaylight.controller.md.sal.dom.store.impl.tree.ListenerRegistrationNode.DataChangeListenerRegistration;
+import org.opendaylight.controller.md.sal.dom.store.impl.tree.ListenerTree;
import org.opendaylight.controller.md.sal.dom.store.impl.tree.ModificationType;
import org.opendaylight.controller.md.sal.dom.store.impl.tree.NodeModification;
import org.opendaylight.controller.md.sal.dom.store.impl.tree.StoreMetadataNode;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransaction;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
+import org.opendaylight.yangtools.concepts.AbstractListenerRegistration;
import org.opendaylight.yangtools.concepts.Identifiable;
import org.opendaylight.yangtools.concepts.ListenerRegistration;
import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
private final ListeningExecutorService executor;
private final String name;
private final AtomicLong txCounter = new AtomicLong(0);
- private final ListenerRegistrationNode listenerTree;
+ private final ListenerTree listenerTree;
private final AtomicReference<DataAndMetadataSnapshot> snapshot;
private ModificationApplyOperation operationTree;
public InMemoryDOMDataStore(final String name, final ListeningExecutorService executor) {
this.name = Preconditions.checkNotNull(name);
this.executor = Preconditions.checkNotNull(executor);
- this.listenerTree = ListenerRegistrationNode.createRoot();
+ this.listenerTree = ListenerTree.create();
this.snapshot = new AtomicReference<DataAndMetadataSnapshot>(DataAndMetadataSnapshot.createEmpty());
this.operationTree = new AlwaysFailOperation();
}
@Override
public <L extends AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>> ListenerRegistration<L> registerChangeListener(
final InstanceIdentifier path, final L listener, final DataChangeScope scope) {
- LOG.debug("{}: Registering data change listener {} for {}",name,listener,path);
- ListenerRegistrationNode listenerNode = listenerTree;
- for(PathArgument arg : path.getPath()) {
- listenerNode = listenerNode.ensureChild(arg);
- }
/*
* Make sure commit is not occurring right now. Listener has to be registered and its
*/
final DataChangeListenerRegistration<L> reg;
synchronized (this) {
- reg = listenerNode.registerDataChangeListener(path, listener, scope);
+ LOG.debug("{}: Registering data change listener {} for {}",name,listener,path);
+
+ reg = listenerTree.registerDataChangeListener(path, listener, scope);
Optional<StoreMetadataNode> currentState = snapshot.get().read(path);
if (currentState.isPresent()) {
final NormalizedNode<?, ?> data = currentState.get().getData();
- final DOMImmutableDataChangeEvent event = DOMImmutableDataChangeEvent.builder() //
+ final DOMImmutableDataChangeEvent event = DOMImmutableDataChangeEvent.builder(DataChangeScope.BASE) //
.setAfter(data) //
.addCreated(path, data) //
.build();
}
}
- return reg;
+ return new AbstractListenerRegistration<L>(listener) {
+ @Override
+ protected void removeRegistration() {
+ synchronized (InMemoryDOMDataStore.this) {
+ reg.close();
+ }
+ }
+ };
}
private synchronized DOMStoreThreePhaseCommitCohort submit(
}
private void commit(final DataAndMetadataSnapshot currentSnapshot,
- final StoreMetadataNode newDataTree, final Iterable<ChangeListenerNotifyTask> listenerTasks) {
+ final StoreMetadataNode newDataTree, final ResolveDataChangeEventsTask listenerResolver) {
LOG.debug("Updating Store snaphot version: {} with version:{}",currentSnapshot.getMetadataTree().getSubtreeVersion(),newDataTree.getSubtreeVersion());
if(LOG.isTraceEnabled()) {
final boolean success = snapshot.compareAndSet(currentSnapshot, newSnapshot);
checkState(success, "Store snapshot and transaction snapshot differ. This should never happen.");
- for (ChangeListenerNotifyTask task : listenerTasks) {
+ for (ChangeListenerNotifyTask task : listenerResolver.call()) {
+ LOG.trace("Scheduling invocation of listeners: {}",task);
executor.submit(task);
}
}
@Override
public void write(final InstanceIdentifier path, final NormalizedNode<?, ?> data) {
checkNotReady();
- mutableTree.write(path, data);
+ try {
+ LOG.trace("Tx: {} Write: {}:{}",getIdentifier(),path,data);
+ mutableTree.write(path, data);
+ // FIXME: Add checked exception
+ } catch (Exception e) {
+ LOG.error("Tx: {}, failed to write {}:{} in {}",getIdentifier(),path,data,mutableTree,e);
+ }
}
@Override
public void delete(final InstanceIdentifier path) {
checkNotReady();
- mutableTree.delete(path);
+ try {
+ LOG.trace("Tx: {} Delete: {}",getIdentifier(),path);
+ mutableTree.delete(path);
+ // FIXME: Add checked exception
+ } catch (Exception e) {
+ LOG.error("Tx: {}, failed to delete {} in {}",getIdentifier(),path,mutableTree,e);
+ }
}
protected final boolean isReady() {
@Override
public ListenableFuture<Optional<NormalizedNode<?, ?>>> read(final InstanceIdentifier path) {
- return Futures.immediateFuture(getMutatedView().read(path));
+ LOG.trace("Tx: {} Read: {}",getIdentifier(),path);
+ try {
+ return Futures.immediateFuture(getMutatedView().read(path));
+ } catch (Exception e) {
+ LOG.error("Tx: {} Failed Read of {}",getIdentifier(),path,e);
+ throw e;
+ }
}
}
private DataAndMetadataSnapshot storeSnapshot;
private Optional<StoreMetadataNode> proposedSubtree;
- private Iterable<ChangeListenerNotifyTask> listenerTasks;
+ private ResolveDataChangeEventsTask listenerResolver;
public ThreePhaseCommitImpl(final SnaphostBackedWriteTransaction writeTransaction) {
this.transaction = writeTransaction;
proposedSubtree = operationTree.apply(modification, Optional.of(metadataTree),
increase(metadataTree.getSubtreeVersion()));
- listenerTasks = DataChangeEventResolver.create() //
+ listenerResolver = ResolveDataChangeEventsTask.create() //
.setRootPath(PUBLIC_ROOT_PATH) //
.setBeforeRoot(Optional.of(metadataTree)) //
.setAfterRoot(proposedSubtree) //
.setModificationRoot(modification) //
- .setListenerRoot(listenerTree) //
- .resolve();
+ .setListenerRoot(listenerTree);
return null;
}
checkState(proposedSubtree != null,"Proposed subtree must be computed");
checkState(storeSnapshot != null,"Proposed subtree must be computed");
// return ImmediateFuture<>;
- InMemoryDOMDataStore.this.commit(storeSnapshot, proposedSubtree.get(),listenerTasks);
+ InMemoryDOMDataStore.this.commit(storeSnapshot, proposedSubtree.get(),listenerResolver);
return Futures.<Void> immediateFuture(null);
}