681132e660d9608f3dea75e58eb4a9f3809f8dd6
[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.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.slf4j.Logger;
28 import org.slf4j.LoggerFactory;
29 import scala.concurrent.Future;
30
31 /**
32  * ListenerRegistrationProxy acts as a proxy for a ListenerRegistration that was done on a remote shard
33  * <p>
34  * Registering a DataChangeListener on the Data Store creates a new instance of the ListenerRegistrationProxy
35  * The ListenerRegistrationProxy talks to a remote ListenerRegistration actor.
36  * </p>
37  */
38 @SuppressWarnings("rawtypes")
39 public class DataChangeListenerRegistrationProxy implements ListenerRegistration {
40
41     private static final Logger LOG = LoggerFactory.getLogger(DataChangeListenerRegistrationProxy.class);
42
43     private volatile ActorSelection listenerRegistrationActor;
44     private final AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> listener;
45     private ActorRef dataChangeListenerActor;
46     private final String shardName;
47     private final ActorContext actorContext;
48     private boolean closed = false;
49
50     public <L extends AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>
51                                                               DataChangeListenerRegistrationProxy (
52             String shardName, ActorContext actorContext, L listener) {
53         this.shardName = shardName;
54         this.actorContext = actorContext;
55         this.listener = listener;
56     }
57
58     @VisibleForTesting
59     ActorSelection getListenerRegistrationActor() {
60         return listenerRegistrationActor;
61     }
62
63     @VisibleForTesting
64     ActorRef getDataChangeListenerActor() {
65         return dataChangeListenerActor;
66     }
67
68     @Override
69     public Object getInstance() {
70         return listener;
71     }
72
73     private void setListenerRegistrationActor(ActorSelection listenerRegistrationActor) {
74         if(listenerRegistrationActor == null) {
75             return;
76         }
77
78         boolean sendCloseMessage = false;
79         synchronized(this) {
80             if(closed) {
81                 sendCloseMessage = true;
82             } else {
83                 this.listenerRegistrationActor = listenerRegistrationActor;
84             }
85         }
86
87         if(sendCloseMessage) {
88             listenerRegistrationActor.tell(new
89                 CloseDataChangeListenerRegistration().toSerializable(), 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.path(), scope),
120                 actorContext.getDatastoreContext().getShardInitializationTimeout());
121
122         future.onComplete(new OnComplete<Object>(){
123             @Override
124             public void onComplete(Throwable failure, Object result) {
125                 if(failure != null) {
126                     LOG.error("Failed to register DataChangeListener {} at path {}",
127                             listener, path.toString(), failure);
128                 } else {
129                     RegisterChangeListenerReply reply = (RegisterChangeListenerReply) result;
130                     setListenerRegistrationActor(actorContext.actorSelection(
131                             reply.getListenerRegistrationPath()));
132                 }
133             }
134         }, actorContext.getClientDispatcher());
135     }
136
137     @Override
138     public void close() {
139
140         boolean sendCloseMessage;
141         synchronized(this) {
142             sendCloseMessage = !closed && listenerRegistrationActor != null;
143             closed = true;
144         }
145
146         if(sendCloseMessage) {
147             listenerRegistrationActor.tell(new CloseDataChangeListenerRegistration().toSerializable(),
148                     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.