Bug 1316: Fixed loop in global RPC forwarding.
[controller.git] / opendaylight / md-sal / sal-binding-broker / src / main / java / org / opendaylight / controller / sal / binding / impl / connect / dom / BindingIndependentConnector.java
1 /*
2  * Copyright (c) 2013 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.sal.binding.impl.connect.dom;
9
10 import static com.google.common.base.Preconditions.checkArgument;
11 import static com.google.common.base.Preconditions.checkState;
12
13 import java.lang.ref.WeakReference;
14 import java.lang.reflect.InvocationHandler;
15 import java.lang.reflect.Method;
16 import java.lang.reflect.Proxy;
17 import java.util.Collection;
18 import java.util.Collections;
19 import java.util.HashMap;
20 import java.util.HashSet;
21 import java.util.Map;
22 import java.util.Map.Entry;
23 import java.util.Set;
24 import java.util.WeakHashMap;
25 import java.util.concurrent.Callable;
26 import java.util.concurrent.ConcurrentHashMap;
27 import java.util.concurrent.ConcurrentMap;
28 import java.util.concurrent.ExecutionException;
29 import java.util.concurrent.Future;
30
31 import org.opendaylight.controller.md.sal.binding.impl.AbstractForwardedDataBroker;
32 import org.opendaylight.controller.md.sal.common.api.RegistrationListener;
33 import org.opendaylight.controller.md.sal.common.api.TransactionStatus;
34 import org.opendaylight.controller.md.sal.common.api.data.DataCommitHandler;
35 import org.opendaylight.controller.md.sal.common.api.data.DataCommitHandler.DataCommitTransaction;
36 import org.opendaylight.controller.md.sal.common.api.data.DataCommitHandlerRegistration;
37 import org.opendaylight.controller.md.sal.common.api.data.DataModification;
38 import org.opendaylight.controller.md.sal.common.api.routing.RouteChange;
39 import org.opendaylight.controller.md.sal.common.api.routing.RouteChangeListener;
40 import org.opendaylight.controller.md.sal.common.api.routing.RouteChangePublisher;
41 import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
42 import org.opendaylight.controller.sal.binding.api.NotificationProviderService.NotificationInterestListener;
43 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
44 import org.opendaylight.controller.sal.binding.api.data.DataProviderService;
45 import org.opendaylight.controller.sal.binding.api.data.RuntimeDataProvider;
46 import org.opendaylight.controller.sal.binding.api.rpc.RpcContextIdentifier;
47 import org.opendaylight.controller.sal.binding.api.rpc.RpcRouter;
48 import org.opendaylight.controller.sal.binding.impl.DataBrokerImpl;
49 import org.opendaylight.controller.sal.binding.impl.MountPointManagerImpl.BindingMountPointImpl;
50 import org.opendaylight.controller.sal.binding.impl.RpcProviderRegistryImpl;
51 import org.opendaylight.controller.sal.binding.impl.RpcProviderRegistryImpl.GlobalRpcRegistrationListener;
52 import org.opendaylight.controller.sal.binding.impl.RpcProviderRegistryImpl.RouterInstantiationListener;
53 import org.opendaylight.controller.sal.common.util.CommitHandlerTransactions;
54 import org.opendaylight.controller.sal.common.util.Rpcs;
55 import org.opendaylight.controller.sal.core.api.Broker.ProviderSession;
56 import org.opendaylight.controller.sal.core.api.Broker.RoutedRpcRegistration;
57 import org.opendaylight.controller.sal.core.api.Provider;
58 import org.opendaylight.controller.sal.core.api.RpcImplementation;
59 import org.opendaylight.controller.sal.core.api.RpcProvisionRegistry;
60 import org.opendaylight.controller.sal.core.api.RpcRegistrationListener;
61 import org.opendaylight.controller.sal.core.api.data.DataModificationTransaction;
62 import org.opendaylight.controller.sal.core.api.notify.NotificationListener;
63 import org.opendaylight.controller.sal.core.api.notify.NotificationPublishService;
64 import org.opendaylight.yangtools.concepts.CompositeObjectRegistration;
65 import org.opendaylight.yangtools.concepts.CompositeObjectRegistration.CompositeObjectRegistrationBuilder;
66 import org.opendaylight.yangtools.concepts.ListenerRegistration;
67 import org.opendaylight.yangtools.concepts.ObjectRegistration;
68 import org.opendaylight.yangtools.concepts.Registration;
69 import org.opendaylight.yangtools.yang.binding.Augmentable;
70 import org.opendaylight.yangtools.yang.binding.Augmentation;
71 import org.opendaylight.yangtools.yang.binding.BaseIdentity;
72 import org.opendaylight.yangtools.yang.binding.BindingMapping;
73 import org.opendaylight.yangtools.yang.binding.DataContainer;
74 import org.opendaylight.yangtools.yang.binding.DataObject;
75 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
76 import org.opendaylight.yangtools.yang.binding.Notification;
77 import org.opendaylight.yangtools.yang.binding.RpcService;
78 import org.opendaylight.yangtools.yang.binding.util.BindingReflections;
79 import org.opendaylight.yangtools.yang.binding.util.ClassLoaderUtils;
80 import org.opendaylight.yangtools.yang.common.QName;
81 import org.opendaylight.yangtools.yang.common.RpcError;
82 import org.opendaylight.yangtools.yang.common.RpcResult;
83 import org.opendaylight.yangtools.yang.data.api.CompositeNode;
84 import org.opendaylight.yangtools.yang.data.impl.codec.BindingIndependentMappingService;
85 import org.opendaylight.yangtools.yang.data.impl.codec.DeserializationException;
86 import org.slf4j.Logger;
87 import org.slf4j.LoggerFactory;
88
89 import com.google.common.base.Function;
90 import com.google.common.base.Optional;
91 import com.google.common.collect.FluentIterable;
92 import com.google.common.collect.ImmutableSet;
93 import com.google.common.collect.ImmutableSet.Builder;
94 import com.google.common.util.concurrent.Futures;
95 import com.google.common.util.concurrent.ListenableFuture;
96
97 public class BindingIndependentConnector implements //
98         RuntimeDataProvider, //
99         Provider, //
100         AutoCloseable {
101
102     private final Logger LOG = LoggerFactory.getLogger(BindingIndependentConnector.class);
103
104     private static final org.opendaylight.yangtools.yang.data.api.InstanceIdentifier ROOT_BI = org.opendaylight.yangtools.yang.data.api.InstanceIdentifier
105             .builder().toInstance();
106
107     private final static Method EQUALS_METHOD;
108
109     private BindingIndependentMappingService mappingService;
110
111     private org.opendaylight.controller.sal.core.api.data.DataProviderService biDataService;
112
113     private DataProviderService baDataService;
114
115     private final ConcurrentMap<Object, BindingToDomTransaction> domOpenedTransactions = new ConcurrentHashMap<>();
116     private final ConcurrentMap<Object, DomToBindingTransaction> bindingOpenedTransactions = new ConcurrentHashMap<>();
117
118     private final BindingToDomCommitHandler bindingToDomCommitHandler = new BindingToDomCommitHandler();
119     private final DomToBindingCommitHandler domToBindingCommitHandler = new DomToBindingCommitHandler();
120
121     private Registration<DataCommitHandler<InstanceIdentifier<? extends DataObject>, DataObject>> baCommitHandlerRegistration;
122
123     private Registration<DataCommitHandler<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode>> biCommitHandlerRegistration;
124
125     private RpcProvisionRegistry biRpcRegistry;
126     private RpcProviderRegistry baRpcRegistry;
127
128     private ListenerRegistration<DomToBindingRpcForwardingManager> domToBindingRpcManager;
129     // private ListenerRegistration<BindingToDomRpcForwardingManager>
130     // bindingToDomRpcManager;
131
132     private final Function<InstanceIdentifier<?>, org.opendaylight.yangtools.yang.data.api.InstanceIdentifier> toDOMInstanceIdentifier = new Function<InstanceIdentifier<?>, org.opendaylight.yangtools.yang.data.api.InstanceIdentifier>() {
133
134         @Override
135         public org.opendaylight.yangtools.yang.data.api.InstanceIdentifier apply(final InstanceIdentifier<?> input) {
136             return mappingService.toDataDom(input);
137         }
138
139     };
140
141     private boolean rpcForwarding = false;
142
143     private boolean dataForwarding = false;
144
145     private boolean notificationForwarding = false;
146
147     private RpcProviderRegistryImpl baRpcRegistryImpl;
148
149     private NotificationProviderService baNotifyService;
150
151     private NotificationPublishService domNotificationService;
152
153     static {
154         try {
155             EQUALS_METHOD = Object.class.getMethod("equals", Object.class);
156         } catch (Exception e) {
157             throw new RuntimeException(e);
158         }
159     }
160
161     @Override
162     public DataObject readOperationalData(final InstanceIdentifier<? extends DataObject> path) {
163         try {
164             org.opendaylight.yangtools.yang.data.api.InstanceIdentifier biPath = mappingService.toDataDom(path);
165             CompositeNode result = biDataService.readOperationalData(biPath);
166             return potentialAugmentationRead(path, biPath, result);
167         } catch (DeserializationException e) {
168             throw new IllegalStateException(e);
169         }
170     }
171
172     private DataObject potentialAugmentationRead(InstanceIdentifier<? extends DataObject> path,
173             final org.opendaylight.yangtools.yang.data.api.InstanceIdentifier biPath, final CompositeNode result)
174             throws DeserializationException {
175         Class<? extends DataObject> targetType = path.getTargetType();
176         if (Augmentation.class.isAssignableFrom(targetType)) {
177             path = mappingService.fromDataDom(biPath);
178             Class<? extends Augmentation<?>> augmentType = (Class<? extends Augmentation<?>>) targetType;
179             DataObject parentTo = mappingService.dataObjectFromDataDom(path, result);
180             if (parentTo instanceof Augmentable<?>) {
181                 return (DataObject) ((Augmentable) parentTo).getAugmentation(augmentType);
182             }
183         }
184         return mappingService.dataObjectFromDataDom(path, result);
185     }
186
187     @Override
188     public DataObject readConfigurationData(final InstanceIdentifier<? extends DataObject> path) {
189         try {
190             org.opendaylight.yangtools.yang.data.api.InstanceIdentifier biPath = mappingService.toDataDom(path);
191             CompositeNode result = biDataService.readConfigurationData(biPath);
192             return potentialAugmentationRead(path, biPath, result);
193         } catch (DeserializationException e) {
194             throw new IllegalStateException(e);
195         }
196     }
197
198     private DataModificationTransaction createBindingToDomTransaction(
199             final DataModification<InstanceIdentifier<? extends DataObject>, DataObject> source) {
200         DataModificationTransaction target = biDataService.beginTransaction();
201         LOG.debug("Created DOM Transaction {} for {},", target.getIdentifier(), source.getIdentifier());
202         for (InstanceIdentifier<? extends DataObject> entry : source.getRemovedConfigurationData()) {
203             org.opendaylight.yangtools.yang.data.api.InstanceIdentifier biEntry = mappingService.toDataDom(entry);
204             target.removeConfigurationData(biEntry);
205             LOG.debug("Delete of Binding Configuration Data {} is translated to {}", entry, biEntry);
206         }
207         for (InstanceIdentifier<? extends DataObject> entry : source.getRemovedOperationalData()) {
208             org.opendaylight.yangtools.yang.data.api.InstanceIdentifier biEntry = mappingService.toDataDom(entry);
209             target.removeOperationalData(biEntry);
210             LOG.debug("Delete of Binding Operational Data {} is translated to {}", entry, biEntry);
211         }
212         for (Entry<InstanceIdentifier<? extends DataObject>, DataObject> entry : source.getUpdatedConfigurationData()
213                 .entrySet()) {
214             Entry<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> biEntry = mappingService
215                     .toDataDom(entry);
216             target.putConfigurationData(biEntry.getKey(), biEntry.getValue());
217             LOG.debug("Update of Binding Configuration Data {} is translated to {}", entry, biEntry);
218         }
219         for (Entry<InstanceIdentifier<? extends DataObject>, DataObject> entry : source.getUpdatedOperationalData()
220                 .entrySet()) {
221             Entry<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> biEntry = mappingService
222                     .toDataDom(entry);
223             target.putOperationalData(biEntry.getKey(), biEntry.getValue());
224             LOG.debug("Update of Binding Operational Data {} is translated to {}", entry, biEntry);
225         }
226
227         return target;
228     }
229
230     private org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction createDomToBindingTransaction(
231             final DataModification<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> source) {
232         org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction target = baDataService
233                 .beginTransaction();
234         for (org.opendaylight.yangtools.yang.data.api.InstanceIdentifier entry : source.getRemovedConfigurationData()) {
235             try {
236
237                 InstanceIdentifier<?> baEntry = mappingService.fromDataDom(entry);
238                 target.removeConfigurationData(baEntry);
239             } catch (DeserializationException e) {
240                 LOG.error("Ommiting from BA transaction: {}.", entry, e);
241             }
242         }
243         for (org.opendaylight.yangtools.yang.data.api.InstanceIdentifier entry : source.getRemovedOperationalData()) {
244             try {
245
246                 InstanceIdentifier<?> baEntry = mappingService.fromDataDom(entry);
247                 target.removeOperationalData(baEntry);
248             } catch (DeserializationException e) {
249                 LOG.error("Ommiting from BA transaction: {}.", entry, e);
250             }
251         }
252         for (Entry<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> entry : source
253                 .getUpdatedConfigurationData().entrySet()) {
254             try {
255                 InstanceIdentifier<?> baKey = mappingService.fromDataDom(entry.getKey());
256                 DataObject baData = mappingService.dataObjectFromDataDom(baKey, entry.getValue());
257                 target.putConfigurationData(baKey, baData);
258             } catch (DeserializationException e) {
259                 LOG.error("Ommiting from BA transaction: {}.", entry.getKey(), e);
260             }
261         }
262         for (Entry<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> entry : source
263                 .getUpdatedOperationalData().entrySet()) {
264             try {
265
266                 InstanceIdentifier<?> baKey = mappingService.fromDataDom(entry.getKey());
267                 DataObject baData = mappingService.dataObjectFromDataDom(baKey, entry.getValue());
268                 target.putOperationalData(baKey, baData);
269             } catch (DeserializationException e) {
270                 LOG.error("Ommiting from BA transaction: {}.", entry.getKey(), e);
271             }
272         }
273         return target;
274     }
275
276     public org.opendaylight.controller.sal.core.api.data.DataProviderService getBiDataService() {
277         return biDataService;
278     }
279
280     protected void setDomDataService(
281             final org.opendaylight.controller.sal.core.api.data.DataProviderService biDataService) {
282         this.biDataService = biDataService;
283     }
284
285     public DataProviderService getBaDataService() {
286         return baDataService;
287     }
288
289     protected void setBindingDataService(final DataProviderService baDataService) {
290         this.baDataService = baDataService;
291     }
292
293     public RpcProviderRegistry getRpcRegistry() {
294         return baRpcRegistry;
295     }
296
297     protected void setBindingRpcRegistry(final RpcProviderRegistry rpcRegistry) {
298         this.baRpcRegistry = rpcRegistry;
299     }
300
301     public void startDataForwarding() {
302         if (baDataService instanceof AbstractForwardedDataBroker) {
303             dataForwarding = true;
304             return;
305         }
306
307         final DataProviderService baData;
308         if (baDataService instanceof BindingMountPointImpl) {
309             baData = ((BindingMountPointImpl) baDataService).getDataBrokerImpl();
310             LOG.debug("Extracted BA Data provider {} from mount point {}", baData, baDataService);
311         } else {
312             baData = baDataService;
313         }
314
315         if (baData instanceof DataBrokerImpl) {
316             checkState(!dataForwarding, "Connector is already forwarding data.");
317             ((DataBrokerImpl) baData).setDataReadDelegate(this);
318             ((DataBrokerImpl) baData).setRootCommitHandler(bindingToDomCommitHandler);
319             biCommitHandlerRegistration = biDataService.registerCommitHandler(ROOT_BI, domToBindingCommitHandler);
320             baDataService.registerCommitHandlerListener(domToBindingCommitHandler);
321         }
322
323         dataForwarding = true;
324     }
325
326     public void startRpcForwarding() {
327         if (biRpcRegistry != null && baRpcRegistry instanceof RouteChangePublisher<?, ?>) {
328             checkState(!rpcForwarding, "Connector is already forwarding RPCs");
329             final DomToBindingRpcForwardingManager biFwdManager = new DomToBindingRpcForwardingManager();
330
331             domToBindingRpcManager = baRpcRegistry.registerRouteChangeListener(biFwdManager);
332             biRpcRegistry.addRpcRegistrationListener(biFwdManager);
333             if (baRpcRegistry instanceof RpcProviderRegistryImpl) {
334                 baRpcRegistryImpl = (RpcProviderRegistryImpl) baRpcRegistry;
335                 baRpcRegistryImpl.registerRouterInstantiationListener(domToBindingRpcManager.getInstance());
336                 baRpcRegistryImpl.registerGlobalRpcRegistrationListener(domToBindingRpcManager.getInstance());
337             }
338             rpcForwarding = true;
339         }
340     }
341
342     public void startNotificationForwarding() {
343         checkState(!notificationForwarding, "Connector is already forwarding notifications.");
344         if (baNotifyService != null && domNotificationService != null) {
345             baNotifyService.registerInterestListener(new DomToBindingNotificationForwarder());
346
347             notificationForwarding = true;
348         }
349     }
350
351     protected void setMappingService(final BindingIndependentMappingService mappingService) {
352         this.mappingService = mappingService;
353     }
354
355     @Override
356     public Collection<ProviderFunctionality> getProviderFunctionality() {
357         return Collections.emptyList();
358     }
359
360     @Override
361     public void onSessionInitiated(final ProviderSession session) {
362         setDomDataService(session.getService(org.opendaylight.controller.sal.core.api.data.DataProviderService.class));
363         setDomRpcRegistry(session.getService(RpcProvisionRegistry.class));
364
365     }
366
367     public <T extends RpcService> void onRpcRouterCreated(final Class<T> serviceType, final RpcRouter<T> router) {
368
369     }
370
371     public void setDomRpcRegistry(final RpcProvisionRegistry registry) {
372         biRpcRegistry = registry;
373     }
374
375     @Override
376     public void close() throws Exception {
377         if (baCommitHandlerRegistration != null) {
378             baCommitHandlerRegistration.close();
379         }
380         if (biCommitHandlerRegistration != null) {
381             biCommitHandlerRegistration.close();
382         }
383
384     }
385
386     private class DomToBindingTransaction implements
387             DataCommitTransaction<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> {
388
389         private final org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction backing;
390         private final DataModification<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> modification;
391
392         public DomToBindingTransaction(
393                 final org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction backing,
394                 final DataModification<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> modification) {
395             super();
396             this.backing = backing;
397             this.modification = modification;
398             bindingOpenedTransactions.put(backing.getIdentifier(), this);
399         }
400
401         @Override
402         public DataModification<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> getModification() {
403             return modification;
404         }
405
406         @Override
407         public RpcResult<Void> rollback() throws IllegalStateException {
408             // backing.cancel();
409             return Rpcs.<Void> getRpcResult(true, null, Collections.<RpcError> emptySet());
410         }
411
412         @Override
413         public RpcResult<Void> finish() throws IllegalStateException {
414             Future<RpcResult<TransactionStatus>> result = backing.commit();
415             try {
416                 RpcResult<TransactionStatus> baResult = result.get();
417                 return Rpcs.<Void> getRpcResult(baResult.isSuccessful(), null, baResult.getErrors());
418             } catch (InterruptedException e) {
419                 throw new IllegalStateException("", e);
420             } catch (ExecutionException e) {
421                 throw new IllegalStateException("", e);
422             }
423         }
424     }
425
426     private class BindingToDomTransaction implements
427             DataCommitTransaction<InstanceIdentifier<? extends DataObject>, DataObject> {
428
429         private final DataModificationTransaction backing;
430         private final DataModification<InstanceIdentifier<? extends DataObject>, DataObject> modification;
431
432         public BindingToDomTransaction(final DataModificationTransaction backing,
433                 final DataModification<InstanceIdentifier<? extends DataObject>, DataObject> modification) {
434             this.backing = backing;
435             this.modification = modification;
436             domOpenedTransactions.put(backing.getIdentifier(), this);
437         }
438
439         @Override
440         public DataModification<InstanceIdentifier<? extends DataObject>, DataObject> getModification() {
441             return modification;
442         }
443
444         @Override
445         public RpcResult<Void> finish() throws IllegalStateException {
446             Future<RpcResult<TransactionStatus>> result = backing.commit();
447             try {
448                 RpcResult<TransactionStatus> biResult = result.get();
449                 return Rpcs.<Void> getRpcResult(biResult.isSuccessful(), null, biResult.getErrors());
450             } catch (InterruptedException e) {
451                 throw new IllegalStateException("", e);
452             } catch (ExecutionException e) {
453                 throw new IllegalStateException("", e);
454             } finally {
455                 domOpenedTransactions.remove(backing.getIdentifier());
456             }
457         }
458
459         @Override
460         public RpcResult<Void> rollback() throws IllegalStateException {
461             domOpenedTransactions.remove(backing.getIdentifier());
462             return Rpcs.<Void> getRpcResult(true, null, Collections.<RpcError> emptySet());
463         }
464     }
465
466     private class BindingToDomCommitHandler implements
467             DataCommitHandler<InstanceIdentifier<? extends DataObject>, DataObject> {
468
469         @Override
470         public org.opendaylight.controller.md.sal.common.api.data.DataCommitHandler.DataCommitTransaction<InstanceIdentifier<? extends DataObject>, DataObject> requestCommit(
471                 final DataModification<InstanceIdentifier<? extends DataObject>, DataObject> bindingTransaction) {
472
473             /**
474              * Transaction was created as DOM transaction, in that case we do
475              * not need to forward it back.
476              */
477             if (bindingOpenedTransactions.containsKey(bindingTransaction.getIdentifier())) {
478
479                 return CommitHandlerTransactions.allwaysSuccessfulTransaction(bindingTransaction);
480             }
481             DataModificationTransaction domTransaction = createBindingToDomTransaction(bindingTransaction);
482             BindingToDomTransaction wrapped = new BindingToDomTransaction(domTransaction, bindingTransaction);
483             LOG.trace("Forwarding Binding Transaction: {} as DOM Transaction: {} .",
484                     bindingTransaction.getIdentifier(), domTransaction.getIdentifier());
485             return wrapped;
486         }
487     }
488
489     private class DomToBindingCommitHandler implements //
490             RegistrationListener<DataCommitHandlerRegistration<InstanceIdentifier<? extends DataObject>, DataObject>>, //
491             DataCommitHandler<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> {
492
493         @Override
494         public void onRegister(
495                 final DataCommitHandlerRegistration<InstanceIdentifier<? extends DataObject>, DataObject> registration) {
496
497             mappingService.toDataDom(registration
498                     .getPath());
499
500         }
501
502         @Override
503         public void onUnregister(
504                 final DataCommitHandlerRegistration<InstanceIdentifier<? extends DataObject>, DataObject> registration) {
505             // NOOP for now
506             // FIXME: do registration based on only active commit handlers.
507         }
508
509         @Override
510         public org.opendaylight.controller.md.sal.common.api.data.DataCommitHandler.DataCommitTransaction<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> requestCommit(
511                 final DataModification<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> domTransaction) {
512             Object identifier = domTransaction.getIdentifier();
513
514             /**
515              * We checks if the transcation was originated in this mapper. If it
516              * was originated in this mapper we are returing allways success
517              * commit hanlder to prevent creating loop in two-phase commit and
518              * duplicating data.
519              */
520             if (domOpenedTransactions.containsKey(identifier)) {
521                 return CommitHandlerTransactions.allwaysSuccessfulTransaction(domTransaction);
522             }
523
524             org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction baTransaction = createDomToBindingTransaction(domTransaction);
525             DomToBindingTransaction forwardedTransaction = new DomToBindingTransaction(baTransaction, domTransaction);
526             LOG.trace("Forwarding DOM Transaction: {} as Binding Transaction: {}.", domTransaction.getIdentifier(),
527                     baTransaction.getIdentifier());
528             return forwardedTransaction;
529         }
530     }
531
532     /**
533      * Manager responsible for instantiating forwarders responsible for
534      * forwarding of RPC invocations from DOM Broker to Binding Aware Broker
535      *
536      */
537     private class DomToBindingRpcForwardingManager implements
538             RouteChangeListener<RpcContextIdentifier, InstanceIdentifier<?>>, RouterInstantiationListener,
539             GlobalRpcRegistrationListener, RpcRegistrationListener {
540
541         private final Map<Class<? extends RpcService>, DomToBindingRpcForwarder> forwarders = new WeakHashMap<>();
542         private RpcProviderRegistryImpl registryImpl;
543
544         public RpcProviderRegistryImpl getRegistryImpl() {
545             return registryImpl;
546         }
547
548         public void setRegistryImpl(final RpcProviderRegistryImpl registryImpl) {
549             this.registryImpl = registryImpl;
550         }
551
552         @Override
553         public void onGlobalRpcRegistered(final Class<? extends RpcService> cls) {
554             getRpcForwarder(cls, null).registerToDOMBroker();
555         }
556
557         @Override
558         public void onGlobalRpcUnregistered(final Class<? extends RpcService> cls) {
559             // NOOP
560         }
561
562         @Override
563         public void onRpcRouterCreated(final RpcRouter<?> router) {
564             Class<? extends BaseIdentity> ctx = router.getContexts().iterator().next();
565             getRpcForwarder(router.getServiceType(), ctx);
566         }
567
568         @Override
569         public void onRouteChange(final RouteChange<RpcContextIdentifier, InstanceIdentifier<?>> change) {
570             for (Entry<RpcContextIdentifier, Set<InstanceIdentifier<?>>> entry : change.getAnnouncements().entrySet()) {
571                 bindingRoutesAdded(entry);
572             }
573         }
574
575         private void bindingRoutesAdded(final Entry<RpcContextIdentifier, Set<InstanceIdentifier<?>>> entry) {
576             Class<? extends BaseIdentity> context = entry.getKey().getRoutingContext();
577             Class<? extends RpcService> service = entry.getKey().getRpcService();
578             if (context != null) {
579                 getRpcForwarder(service, context).registerPaths(context, service, entry.getValue());
580             }
581         }
582
583         private DomToBindingRpcForwarder getRpcForwarder(final Class<? extends RpcService> service,
584                 final Class<? extends BaseIdentity> context) {
585             DomToBindingRpcForwarder potential = forwarders.get(service);
586             if (potential != null) {
587                 return potential;
588             }
589             if (context == null) {
590                 potential = new DomToBindingRpcForwarder(service);
591             } else {
592                 potential = new DomToBindingRpcForwarder(service, context);
593             }
594
595             forwarders.put(service, potential);
596             return potential;
597         }
598
599         @Override
600         public void onRpcImplementationAdded(final QName name) {
601
602             final Optional<Class<? extends RpcService>> rpcInterface = mappingService.getRpcServiceClassFor(
603                     name.getNamespace().toString(), name.getFormattedRevision());
604             if (rpcInterface.isPresent()) {
605                 getRpcForwarder(rpcInterface.get(), null).registerToBidningBroker();
606             }
607         }
608
609         @Override
610         public void onRpcImplementationRemoved(final QName name) {
611
612         }
613     }
614
615     private class DomToBindingRpcForwarder implements RpcImplementation, InvocationHandler {
616
617         private final Set<QName> supportedRpcs;
618         private final WeakReference<Class<? extends RpcService>> rpcServiceType;
619         private Set<org.opendaylight.controller.sal.core.api.Broker.RoutedRpcRegistration> registrations;
620         private final Map<QName, RpcInvocationStrategy> strategiesByQName = new HashMap<>();
621         private final WeakHashMap<Method, RpcInvocationStrategy> strategiesByMethod = new WeakHashMap<>();
622         private final RpcService proxy;
623         private ObjectRegistration<?> forwarderRegistration;
624
625         public DomToBindingRpcForwarder(final Class<? extends RpcService> service) {
626             this.rpcServiceType = new WeakReference<Class<? extends RpcService>>(service);
627             this.supportedRpcs = mappingService.getRpcQNamesFor(service);
628
629             Class<?> cls = rpcServiceType.get();
630             ClassLoader clsLoader = cls.getClassLoader();
631             proxy =(RpcService) Proxy.newProxyInstance(clsLoader, new Class<?>[] { cls }, this);
632             createStrategies();
633         }
634
635         /**
636          * Constructor for Routed RPC Forwareder.
637          *
638          * @param service
639          * @param context
640          */
641         public DomToBindingRpcForwarder(final Class<? extends RpcService> service,
642                                         final Class<? extends BaseIdentity> context) {
643             this(service);
644             Builder<RoutedRpcRegistration> registrationsBuilder = ImmutableSet
645                     .<org.opendaylight.controller.sal.core.api.Broker.RoutedRpcRegistration> builder();
646             try {
647                 for (QName rpc : supportedRpcs) {
648                     registrationsBuilder.add(biRpcRegistry.addRoutedRpcImplementation(rpc, this));
649                 }
650                 createDefaultDomForwarder();
651             } catch (Exception e) {
652                 LOG.error("Could not forward Rpcs of type {}", service.getName(), e);
653             }
654             registrations = registrationsBuilder.build();
655         }
656
657
658
659         private void createStrategies() {
660             try {
661                 for (QName rpc : supportedRpcs) {
662                     RpcInvocationStrategy strategy = createInvocationStrategy(rpc, rpcServiceType.get());
663                     strategiesByMethod.put(strategy.targetMethod, strategy);
664                     strategiesByQName.put(rpc, strategy);
665                 }
666             } catch (Exception e) {
667                 LOG.error("Could not forward Rpcs of type {}", rpcServiceType.get(), e);
668             }
669
670         }
671
672         /**
673          * Registers RPC Forwarder to DOM Broker,
674          * this means Binding Aware Broker has implementation of RPC
675          * which is registered to it.
676          *
677          * If RPC Forwarder was previously registered to DOM Broker
678          * or to Bidning Broker this method is noop to prevent
679          * creating forwarding loop.
680          *
681          */
682         public void registerToDOMBroker() {
683             if(forwarderRegistration == null) {
684                 CompositeObjectRegistrationBuilder<DomToBindingRpcForwarder> builder = CompositeObjectRegistration.builderFor(this);
685                 try {
686                     for (QName rpc : supportedRpcs) {
687                         builder.add(biRpcRegistry.addRpcImplementation(rpc, this));
688                     }
689                 } catch (Exception e) {
690                     LOG.error("Could not forward Rpcs of type {}", rpcServiceType.get(), e);
691                 }
692                 this.forwarderRegistration = builder.toInstance();
693             }
694         }
695
696
697         public void registerPaths(final Class<? extends BaseIdentity> context,
698                 final Class<? extends RpcService> service, final Set<InstanceIdentifier<?>> set) {
699             QName ctx = BindingReflections.findQName(context);
700             for (org.opendaylight.yangtools.yang.data.api.InstanceIdentifier path : FluentIterable.from(set).transform(
701                     toDOMInstanceIdentifier)) {
702                 for (org.opendaylight.controller.sal.core.api.Broker.RoutedRpcRegistration reg : registrations) {
703                     reg.registerPath(ctx, path);
704                 }
705             }
706         }
707
708         @Override
709         public Object invoke(final Object proxy, final Method method, final Object[] args) throws Throwable {
710             if (EQUALS_METHOD.equals(method)) {
711                 return false;
712             }
713             RpcInvocationStrategy strategy = strategiesByMethod.get(method);
714             checkState(strategy != null);
715             checkArgument(args.length <= 2);
716             if (args.length == 1) {
717                 checkArgument(args[0] instanceof DataObject);
718                 return strategy.forwardToDomBroker((DataObject) args[0]);
719             }
720             return strategy.forwardToDomBroker(null);
721         }
722
723         public void removePaths(final Class<? extends BaseIdentity> context, final Class<? extends RpcService> service,
724                 final Set<InstanceIdentifier<?>> set) {
725             QName ctx = BindingReflections.findQName(context);
726             for (org.opendaylight.yangtools.yang.data.api.InstanceIdentifier path : FluentIterable.from(set).transform(
727                     toDOMInstanceIdentifier)) {
728                 for (org.opendaylight.controller.sal.core.api.Broker.RoutedRpcRegistration reg : registrations) {
729                     reg.unregisterPath(ctx, path);
730                 }
731             }
732         }
733
734         @Override
735         public Set<QName> getSupportedRpcs() {
736             return supportedRpcs;
737         }
738
739         @SuppressWarnings({ "unchecked", "rawtypes" })
740         public void createDefaultDomForwarder() {
741             if (baRpcRegistryImpl != null) {
742                 Class<?> cls = rpcServiceType.get();
743                 ClassLoader clsLoader = cls.getClassLoader();
744                 RpcService proxy = (RpcService) Proxy.newProxyInstance(clsLoader, new Class<?>[] { cls }, this);
745
746                 RpcRouter rpcRouter = baRpcRegistryImpl.getRpcRouter(rpcServiceType.get());
747                 rpcRouter.registerDefaultService(proxy);
748             }
749         }
750
751         @Override
752         public ListenableFuture<RpcResult<CompositeNode>> invokeRpc(final QName rpc, final CompositeNode domInput) {
753             checkArgument(rpc != null);
754             checkArgument(domInput != null);
755
756             Class<? extends RpcService> rpcType = rpcServiceType.get();
757             checkState(rpcType != null);
758             RpcService rpcService = baRpcRegistry.getRpcService(rpcType);
759             checkState(rpcService != null);
760             CompositeNode domUnwrappedInput = domInput.getFirstCompositeByName(QName.create(rpc, "input"));
761
762             try {
763                 return Futures.immediateFuture(resolveInvocationStrategy(rpc).invokeOn(rpcService, domUnwrappedInput));
764             } catch (Exception e) {
765                 return Futures.immediateFailedFuture(e);
766             }
767         }
768
769         private RpcInvocationStrategy resolveInvocationStrategy(final QName rpc) {
770             return strategiesByQName.get(rpc);
771         }
772
773         private RpcInvocationStrategy createInvocationStrategy(final QName rpc,
774                 final Class<? extends RpcService> rpcType) throws Exception {
775             return ClassLoaderUtils.withClassLoader(rpcType.getClassLoader(), new Callable<RpcInvocationStrategy>() {
776                 @Override
777                 public RpcInvocationStrategy call() throws Exception {
778                     String methodName = BindingMapping.getMethodName(rpc);
779                     Method targetMethod = null;
780                     for (Method possibleMethod : rpcType.getMethods()) {
781                         if (possibleMethod.getName().equals(methodName)
782                                 && BindingReflections.isRpcMethod(possibleMethod)) {
783                             targetMethod = possibleMethod;
784                             break;
785                         }
786                     }
787                     checkState(targetMethod != null, "Rpc method not found");
788                     return  new RpcInvocationStrategy(rpc,targetMethod, mappingService, biRpcRegistry);
789                 }
790
791             });
792         }
793
794         /**
795          * Registers RPC Forwarder to Binding Broker,
796          * this means DOM Broekr has implementation of RPC
797          * which is registered to it.
798          *
799          * If RPC Forwarder was previously registered to DOM Broker
800          * or to Bidning Broker this method is noop to prevent
801          * creating forwarding loop.
802          *
803          */
804         public void registerToBidningBroker() {
805             if(forwarderRegistration == null) {
806                try {
807                    this.forwarderRegistration = baRpcRegistry.addRpcImplementation((Class)rpcServiceType.get(), proxy);
808                } catch (Exception e) {
809                    LOG.error("Unable to forward RPCs for {}",rpcServiceType.get(),e);
810                }
811             }
812         }
813     }
814
815     public boolean isRpcForwarding() {
816         return rpcForwarding;
817     }
818
819     public boolean isDataForwarding() {
820         return dataForwarding;
821     }
822
823     public boolean isNotificationForwarding() {
824         return notificationForwarding;
825     }
826
827     public BindingIndependentMappingService getMappingService() {
828         return mappingService;
829     }
830
831     public void setBindingNotificationService(final NotificationProviderService baService) {
832         this.baNotifyService = baService;
833
834     }
835
836     public void setDomNotificationService(final NotificationPublishService domService) {
837         this.domNotificationService = domService;
838     }
839
840     private class DomToBindingNotificationForwarder implements NotificationInterestListener, NotificationListener {
841
842         private final ConcurrentMap<QName, WeakReference<Class<? extends Notification>>> notifications = new ConcurrentHashMap<>();
843         private final Set<QName> supportedNotifications = new HashSet<>();
844
845         @Override
846         public Set<QName> getSupportedNotifications() {
847             return Collections.unmodifiableSet(supportedNotifications);
848         }
849
850         @Override
851         public void onNotification(final CompositeNode notification) {
852             QName qname = notification.getNodeType();
853             WeakReference<Class<? extends Notification>> potential = notifications.get(qname);
854             if (potential != null) {
855                 Class<? extends Notification> potentialClass = potential.get();
856                 if (potentialClass != null) {
857                     final DataContainer baNotification = mappingService.dataObjectFromDataDom(potentialClass,
858                             notification);
859
860                     if (baNotification instanceof Notification) {
861                         baNotifyService.publish((Notification) baNotification);
862                     }
863                 }
864             }
865         }
866
867         @Override
868         public void onNotificationSubscribtion(final Class<? extends Notification> notificationType) {
869             QName qname = BindingReflections.findQName(notificationType);
870             if (qname != null) {
871                 WeakReference<Class<? extends Notification>> already = notifications.putIfAbsent(qname,
872                         new WeakReference<Class<? extends Notification>>(notificationType));
873                 if (already == null) {
874                     domNotificationService.addNotificationListener(qname, this);
875                     supportedNotifications.add(qname);
876                 }
877             }
878         }
879     }
880 }

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