opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DataChangeListenerSupport.java

   1 /*
   2  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.datastore;
   9
  10 import akka.actor.ActorRef;
  11 import akka.actor.ActorSelection;
  12 import com.google.common.base.Optional;
  13 import com.google.common.collect.Sets;
  14 import java.util.ArrayList;
  15 import java.util.Collection;
  16 import java.util.Map.Entry;
  17 import java.util.Set;
  18 import org.opendaylight.controller.cluster.datastore.messages.EnableNotification;
  19 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
  20 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
  21 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
  22 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
  23 import org.opendaylight.controller.md.sal.dom.store.impl.DataChangeListenerRegistration;
  24 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  25 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  26 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  27 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
  28
  29 final class DataChangeListenerSupport extends AbstractDataListenerSupport<
  30         AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>, RegisterChangeListener,
  31             DelayedDataChangeListenerRegistration, DataChangeListenerRegistration<
  32                     AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>> {
  33
  34     private final Set<ActorSelection> listenerActors = Sets.newConcurrentHashSet();
  35
  36     DataChangeListenerSupport(final Shard shard) {
  37         super(shard);
  38     }
  39
  40     Collection<ActorSelection> getListenerActors() {
  41         return new ArrayList<>(listenerActors);
  42     }
  43
  44     @Override
  45     DataChangeListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>
  46             createDelegate(final RegisterChangeListener message) {
  47         final ActorSelection dataChangeListenerPath = selectActor(message.getDataChangeListenerPath());
  48
  49         // Notify the listener if notifications should be enabled or not
  50         // If this shard is the leader then it will enable notifications else
  51         // it will not
  52         dataChangeListenerPath.tell(new EnableNotification(true), getSelf());
  53
  54         // Now store a reference to the data change listener so it can be notified
  55         // at a later point if notifications should be enabled or disabled
  56         if (!message.isRegisterOnAllInstances()) {
  57             addListenerActor(dataChangeListenerPath);
  58         }
  59
  60         AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> listener =
  61                 new DataChangeListenerProxy(dataChangeListenerPath);
  62
  63         log().debug("{}: Registering for path {}", persistenceId(), message.getPath());
  64
  65         Entry<DataChangeListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>,
  66                 Optional<DataTreeCandidate>> regEntry = getShard().getDataStore().registerChangeListener(
  67                         message.getPath(), listener, message.getScope());
  68
  69         getShard().getDataStore().notifyOfInitialData(regEntry.getKey(), regEntry.getValue());
  70
  71         listenerActors.add(dataChangeListenerPath);
  72         final DataChangeListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>
  73             delegate = regEntry.getKey();
  74         return new DataChangeListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier,
  75                 NormalizedNode<?,?>>>() {
  76             @Override
  77             public void close() {
  78                 listenerActors.remove(dataChangeListenerPath);
  79                 delegate.close();
  80             }
  81
  82             @Override
  83             public AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> getInstance() {
  84                 return delegate.getInstance();
  85             }
  86
  87             @Override
  88             public YangInstanceIdentifier getPath() {
  89                 return delegate.getPath();
  90             }
  91
  92             @Override
  93             public DataChangeScope getScope() {
  94                 return delegate.getScope();
  95             }
  96         };
  97     }
  98
  99     @Override
 100     protected DelayedDataChangeListenerRegistration newDelayedListenerRegistration(RegisterChangeListener message) {
 101         return new DelayedDataChangeListenerRegistration(message);
 102     }
 103
 104     @Override
 105     protected ActorRef newRegistrationActor(
 106             ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>> registration) {
 107         return createActor(DataChangeListenerRegistrationActor.props(registration));
 108     }
 109
 110     @Override
 111     protected Object newRegistrationReplyMessage(ActorRef registrationActor) {
 112         return new RegisterChangeListenerReply(registrationActor);
 113     }
 114
 115     @Override
 116     protected String logName() {
 117         return "registerDataChangeListener";
 118     }
 119 }