534beead0ed4d256c4626870fbee54582add8d9f
[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 org.opendaylight.controller.cluster.datastore.exceptions.LocalShardNotFoundException;
17 import org.opendaylight.controller.cluster.datastore.messages.CloseDataChangeListenerRegistration;
18 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
19 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
20 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
21 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker;
22 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
23 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
24 import org.opendaylight.controller.md.sal.dom.api.ClusteredDOMDataChangeListener;
25 import org.opendaylight.yangtools.concepts.ListenerRegistration;
26 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
27 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
28 import org.slf4j.Logger;
29 import org.slf4j.LoggerFactory;
30 import scala.concurrent.Future;
31
32 /**
33  * ListenerRegistrationProxy acts as a proxy for a ListenerRegistration that was done on a remote shard
34  * <p>
35  * Registering a DataChangeListener on the Data Store creates a new instance of the ListenerRegistrationProxy
36  * The ListenerRegistrationProxy talks to a remote ListenerRegistration actor.
37  * </p>
38  */
39 @SuppressWarnings("rawtypes")
40 public class DataChangeListenerRegistrationProxy implements ListenerRegistration {
41
42     private static final Logger LOG = LoggerFactory.getLogger(DataChangeListenerRegistrationProxy.class);
43
44     private volatile ActorSelection listenerRegistrationActor;
45     private final AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> listener;
46     private ActorRef dataChangeListenerActor;
47     private final String shardName;
48     private final ActorContext actorContext;
49     private boolean closed = false;
50
51     public <L extends AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>
52                                                               DataChangeListenerRegistrationProxy (
53             String shardName, ActorContext actorContext, L listener) {
54         this.shardName = shardName;
55         this.actorContext = actorContext;
56         this.listener = 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(CloseDataChangeListenerRegistration.INSTANCE, null);
90         }
91     }
92
93     public void init(final YangInstanceIdentifier path, final AsyncDataBroker.DataChangeScope scope) {
94
95         dataChangeListenerActor = actorContext.getActorSystem().actorOf(
96                 DataChangeListener.props(listener).withDispatcher(actorContext.getNotificationDispatcherPath()));
97
98         Future<ActorRef> findFuture = actorContext.findLocalShardAsync(shardName);
99         findFuture.onComplete(new OnComplete<ActorRef>() {
100             @Override
101             public void onComplete(Throwable failure, ActorRef shard) {
102                 if(failure instanceof LocalShardNotFoundException) {
103                     LOG.debug("No local shard found for {} - DataChangeListener {} at path {} " +
104                             "cannot be registered", shardName, listener, path);
105                 } else if(failure != null) {
106                     LOG.error("Failed to find local shard {} - DataChangeListener {} at path {} " +
107                             "cannot be registered: {}", shardName, listener, path, failure);
108                 } else {
109                     doRegistration(shard, path, scope);
110                 }
111             }
112         }, actorContext.getClientDispatcher());
113     }
114
115     private void doRegistration(ActorRef shard, final YangInstanceIdentifier path,
116             DataChangeScope scope) {
117
118         Future<Object> future = actorContext.executeOperationAsync(shard,
119                 new RegisterChangeListener(path, dataChangeListenerActor, scope,
120                     listener instanceof ClusteredDOMDataChangeListener),
121                 actorContext.getDatastoreContext().getShardInitializationTimeout());
122
123         future.onComplete(new OnComplete<Object>(){
124             @Override
125             public void onComplete(Throwable failure, Object result) {
126                 if(failure != null) {
127                     LOG.error("Failed to register DataChangeListener {} at path {}",
128                             listener, path.toString(), failure);
129                 } else {
130                     RegisterChangeListenerReply reply = (RegisterChangeListenerReply) result;
131                     setListenerRegistrationActor(actorContext.actorSelection(
132                             reply.getListenerRegistrationPath()));
133                 }
134             }
135         }, actorContext.getClientDispatcher());
136     }
137
138     @Override
139     public void close() {
140
141         boolean sendCloseMessage;
142         synchronized(this) {
143             sendCloseMessage = !closed && listenerRegistrationActor != null;
144             closed = true;
145         }
146
147         if(sendCloseMessage) {
148             listenerRegistrationActor.tell(CloseDataChangeListenerRegistration.INSTANCE, ActorRef.noSender());
149             listenerRegistrationActor = null;
150         }
151
152         if(dataChangeListenerActor != null) {
153             dataChangeListenerActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
154             dataChangeListenerActor = null;
155         }
156     }
157 }

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