9b2beccdad99740cbcba027e4a893d8cf9555133
[controller.git] / 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.Collection;
15 import java.util.Collections;
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 Collections.unmodifiableCollection(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 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.