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

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