3fe3933fe72c08b84f33b747760ce6e3f6f17aa9
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / DataChangeListenerRegistrationProxy.java
1 /*
2  * Copyright (c) 2014 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
9 package org.opendaylight.controller.cluster.datastore;
10
11 import akka.actor.ActorRef;
12 import akka.actor.ActorSelection;
13 import akka.actor.PoisonPill;
14 import akka.dispatch.OnComplete;
15 import com.google.common.annotations.VisibleForTesting;
16 import com.google.common.base.Preconditions;
17 import org.opendaylight.controller.cluster.datastore.exceptions.LocalShardNotFoundException;
18 import org.opendaylight.controller.cluster.datastore.messages.CloseDataTreeNotificationListenerRegistration;
19 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
20 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
21 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
22 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker;
23 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
24 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
25 import org.opendaylight.controller.md.sal.dom.api.ClusteredDOMDataChangeListener;
26 import org.opendaylight.yangtools.concepts.ListenerRegistration;
27 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
28 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
29 import org.slf4j.Logger;
30 import org.slf4j.LoggerFactory;
31 import scala.concurrent.Future;
32
33 /**
34  * ListenerRegistrationProxy acts as a proxy for a ListenerRegistration that was done on a remote shard
35  * <p>
36  * Registering a DataChangeListener on the Data Store creates a new instance of the ListenerRegistrationProxy
37  * The ListenerRegistrationProxy talks to a remote ListenerRegistration actor.
38  * </p>
39  */
40 @SuppressWarnings("rawtypes")
41 public class DataChangeListenerRegistrationProxy implements ListenerRegistration {
42
43     private static final Logger LOG = LoggerFactory.getLogger(DataChangeListenerRegistrationProxy.class);
44
45     private final AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> listener;
46     private final String shardName;
47     private final ActorContext actorContext;
48     private ActorRef dataChangeListenerActor;
49     private volatile ActorSelection listenerRegistrationActor;
50     private boolean closed = false;
51
52     public <L extends AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>
53             DataChangeListenerRegistrationProxy(String shardName, ActorContext actorContext, L listener) {
54         this.shardName = Preconditions.checkNotNull(shardName);
55         this.actorContext = Preconditions.checkNotNull(actorContext);
56         this.listener = Preconditions.checkNotNull(listener);
57     }
58
59     @VisibleForTesting
60     ActorSelection getListenerRegistrationActor() {
61         return listenerRegistrationActor;
62     }
63
64     @VisibleForTesting
65     ActorRef getDataChangeListenerActor() {
66         return dataChangeListenerActor;
67     }
68
69     @Override
70     public Object getInstance() {
71         return listener;
72     }
73
74     private void setListenerRegistrationActor(ActorSelection listenerRegistrationActor) {
75         if (listenerRegistrationActor == null) {
76             return;
77         }
78
79         boolean sendCloseMessage = false;
80         synchronized (this) {
81             if (closed) {
82                 sendCloseMessage = true;
83             } else {
84                 this.listenerRegistrationActor = listenerRegistrationActor;
85             }
86         }
87
88         if (sendCloseMessage) {
89             listenerRegistrationActor.tell(CloseDataTreeNotificationListenerRegistration.getInstance(),
90                     ActorRef.noSender());
91         }
92     }
93
94     public void init(final YangInstanceIdentifier path, final AsyncDataBroker.DataChangeScope scope) {
95
96         dataChangeListenerActor = actorContext.getActorSystem().actorOf(
97                 DataChangeListener.props(listener, path).withDispatcher(actorContext.getNotificationDispatcherPath()));
98
99         Future<ActorRef> findFuture = actorContext.findLocalShardAsync(shardName);
100         findFuture.onComplete(new OnComplete<ActorRef>() {
101             @Override
102             public void onComplete(Throwable failure, ActorRef shard) {
103                 if (failure instanceof LocalShardNotFoundException) {
104                     LOG.debug("No local shard found for {} - DataChangeListener {} at path {} "
105                             + "cannot be registered", shardName, listener, path);
106                 } else if (failure != null) {
107                     LOG.error("Failed to find local shard {} - DataChangeListener {} at path {} "
108                             + "cannot be registered: {}", shardName, listener, path, failure);
109                 } else {
110                     doRegistration(shard, path, scope);
111                 }
112             }
113         }, actorContext.getClientDispatcher());
114     }
115
116     private void doRegistration(ActorRef shard, final YangInstanceIdentifier path,
117             DataChangeScope scope) {
118
119         Future<Object> future = actorContext.executeOperationAsync(shard,
120                 new RegisterChangeListener(path, dataChangeListenerActor, scope,
121                     listener instanceof ClusteredDOMDataChangeListener),
122                 actorContext.getDatastoreContext().getShardInitializationTimeout());
123
124         future.onComplete(new OnComplete<Object>() {
125             @Override
126             public void onComplete(Throwable failure, Object result) {
127                 if (failure != null) {
128                     LOG.error("Failed to register DataChangeListener {} at path {}",
129                             listener, path.toString(), failure);
130                 } else {
131                     RegisterChangeListenerReply reply = (RegisterChangeListenerReply) result;
132                     setListenerRegistrationActor(actorContext.actorSelection(
133                             reply.getListenerRegistrationPath()));
134                 }
135             }
136         }, actorContext.getClientDispatcher());
137     }
138
139     @Override
140     public void close() {
141
142         boolean sendCloseMessage;
143         synchronized (this) {
144             sendCloseMessage = !closed && listenerRegistrationActor != null;
145             closed = true;
146         }
147
148         if (sendCloseMessage) {
149             listenerRegistrationActor.tell(CloseDataTreeNotificationListenerRegistration.getInstance(),
150                     ActorRef.noSender());
151             listenerRegistrationActor = null;
152         }
153
154         if (dataChangeListenerActor != null) {
155             dataChangeListenerActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
156             dataChangeListenerActor = null;
157         }
158     }
159 }

©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.