Merge "Bug 509: Improve logging in InMemoryDataStore."
[controller.git] / opendaylight / md-sal / sal-binding-broker / src / main / java / org / opendaylight / controller / sal / binding / impl / connect / dom / BindingIndependentConnector.java
index e48ebbc0577f1b6101a772284ba9e07b3e6580cf..a398abc75a3254c3160ae685f9d829920f726f6c 100644 (file)
@@ -28,13 +28,13 @@ import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 
+import org.opendaylight.controller.md.sal.binding.impl.AbstractForwardedDataBroker;
 import org.opendaylight.controller.md.sal.common.api.RegistrationListener;
 import org.opendaylight.controller.md.sal.common.api.TransactionStatus;
 import org.opendaylight.controller.md.sal.common.api.data.DataCommitHandler;
 import org.opendaylight.controller.md.sal.common.api.data.DataCommitHandler.DataCommitTransaction;
 import org.opendaylight.controller.md.sal.common.api.data.DataCommitHandlerRegistration;
 import org.opendaylight.controller.md.sal.common.api.data.DataModification;
-import org.opendaylight.controller.md.sal.common.api.data.DataReader;
 import org.opendaylight.controller.md.sal.common.api.routing.RouteChange;
 import org.opendaylight.controller.md.sal.common.api.routing.RouteChangeListener;
 import org.opendaylight.controller.md.sal.common.api.routing.RouteChangePublisher;
@@ -45,6 +45,8 @@ import org.opendaylight.controller.sal.binding.api.data.DataProviderService;
 import org.opendaylight.controller.sal.binding.api.data.RuntimeDataProvider;
 import org.opendaylight.controller.sal.binding.api.rpc.RpcContextIdentifier;
 import org.opendaylight.controller.sal.binding.api.rpc.RpcRouter;
+import org.opendaylight.controller.sal.binding.impl.DataBrokerImpl;
+import org.opendaylight.controller.sal.binding.impl.MountPointManagerImpl.BindingMountPointImpl;
 import org.opendaylight.controller.sal.binding.impl.RpcProviderRegistryImpl;
 import org.opendaylight.controller.sal.binding.impl.RpcProviderRegistryImpl.GlobalRpcRegistrationListener;
 import org.opendaylight.controller.sal.binding.impl.RpcProviderRegistryImpl.RouterInstantiationListener;
@@ -89,6 +91,7 @@ import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.ImmutableSet.Builder;
 import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
 
 public class BindingIndependentConnector implements //
         RuntimeDataProvider, //
@@ -99,8 +102,6 @@ public class BindingIndependentConnector implements //
 
     private final Logger LOG = LoggerFactory.getLogger(BindingIndependentConnector.class);
 
-    @SuppressWarnings("deprecation")
-    private static final InstanceIdentifier<? extends DataObject> ROOT = InstanceIdentifier.builder().toInstance();
 
     private static final org.opendaylight.yangtools.yang.data.api.InstanceIdentifier ROOT_BI = org.opendaylight.yangtools.yang.data.api.InstanceIdentifier
             .builder().toInstance();
@@ -133,14 +134,12 @@ public class BindingIndependentConnector implements //
     private final Function<InstanceIdentifier<?>, org.opendaylight.yangtools.yang.data.api.InstanceIdentifier> toDOMInstanceIdentifier = new Function<InstanceIdentifier<?>, org.opendaylight.yangtools.yang.data.api.InstanceIdentifier>() {
 
         @Override
-        public org.opendaylight.yangtools.yang.data.api.InstanceIdentifier apply(InstanceIdentifier<?> input) {
+        public org.opendaylight.yangtools.yang.data.api.InstanceIdentifier apply(final InstanceIdentifier<?> input) {
             return mappingService.toDataDom(input);
         }
 
     };
 
-    private Registration<DataReader<InstanceIdentifier<? extends DataObject>, DataObject>> baDataReaderRegistration;
-
     private boolean rpcForwarding = false;
 
     private boolean dataForwarding = false;
@@ -162,7 +161,7 @@ public class BindingIndependentConnector implements //
     }
 
     @Override
-    public DataObject readOperationalData(InstanceIdentifier<? extends DataObject> path) {
+    public DataObject readOperationalData(final InstanceIdentifier<? extends DataObject> path) {
         try {
             org.opendaylight.yangtools.yang.data.api.InstanceIdentifier biPath = mappingService.toDataDom(path);
             CompositeNode result = biDataService.readOperationalData(biPath);
@@ -173,7 +172,7 @@ public class BindingIndependentConnector implements //
     }
 
     private DataObject potentialAugmentationRead(InstanceIdentifier<? extends DataObject> path,
-            org.opendaylight.yangtools.yang.data.api.InstanceIdentifier biPath, CompositeNode result)
+            final org.opendaylight.yangtools.yang.data.api.InstanceIdentifier biPath, final CompositeNode result)
             throws DeserializationException {
         Class<? extends DataObject> targetType = path.getTargetType();
         if (Augmentation.class.isAssignableFrom(targetType)) {
@@ -188,7 +187,7 @@ public class BindingIndependentConnector implements //
     }
 
     @Override
-    public DataObject readConfigurationData(InstanceIdentifier<? extends DataObject> path) {
+    public DataObject readConfigurationData(final InstanceIdentifier<? extends DataObject> path) {
         try {
             org.opendaylight.yangtools.yang.data.api.InstanceIdentifier biPath = mappingService.toDataDom(path);
             CompositeNode result = biDataService.readConfigurationData(biPath);
@@ -199,9 +198,19 @@ public class BindingIndependentConnector implements //
     }
 
     private DataModificationTransaction createBindingToDomTransaction(
-            DataModification<InstanceIdentifier<? extends DataObject>, DataObject> source) {
+            final DataModification<InstanceIdentifier<? extends DataObject>, DataObject> source) {
         DataModificationTransaction target = biDataService.beginTransaction();
         LOG.debug("Created DOM Transaction {} for {},", target.getIdentifier(),source.getIdentifier());
+        for (InstanceIdentifier<? extends DataObject> entry : source.getRemovedConfigurationData()) {
+            org.opendaylight.yangtools.yang.data.api.InstanceIdentifier biEntry = mappingService.toDataDom(entry);
+            target.removeConfigurationData(biEntry);
+            LOG.debug("Delete of Binding Configuration Data {} is translated to {}",entry,biEntry);
+        }
+        for (InstanceIdentifier<? extends DataObject> entry : source.getRemovedOperationalData()) {
+            org.opendaylight.yangtools.yang.data.api.InstanceIdentifier biEntry = mappingService.toDataDom(entry);
+            target.removeOperationalData(biEntry);
+            LOG.debug("Delete of Binding Operational Data {} is translated to {}",entry,biEntry);
+        }
         for (Entry<InstanceIdentifier<? extends DataObject>, DataObject> entry : source.getUpdatedConfigurationData()
                 .entrySet()) {
             Entry<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> biEntry = mappingService
@@ -216,23 +225,32 @@ public class BindingIndependentConnector implements //
             target.putOperationalData(biEntry.getKey(), biEntry.getValue());
             LOG.debug("Update of Binding Operational Data {} is translated to {}",entry,biEntry);
         }
-        for (InstanceIdentifier<? extends DataObject> entry : source.getRemovedConfigurationData()) {
-            org.opendaylight.yangtools.yang.data.api.InstanceIdentifier biEntry = mappingService.toDataDom(entry);
-            target.removeConfigurationData(biEntry);
-            LOG.debug("Delete of Binding Configuration Data {} is translated to {}",entry,biEntry);
-        }
-        for (InstanceIdentifier<? extends DataObject> entry : source.getRemovedOperationalData()) {
-            org.opendaylight.yangtools.yang.data.api.InstanceIdentifier biEntry = mappingService.toDataDom(entry);
-            target.removeOperationalData(biEntry);
-            LOG.debug("Delete of Binding Operational Data {} is translated to {}",entry,biEntry);
-        }
+
         return target;
     }
 
     private org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction createDomToBindingTransaction(
-            DataModification<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> source) {
+            final DataModification<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> source) {
         org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction target = baDataService
                 .beginTransaction();
+        for (org.opendaylight.yangtools.yang.data.api.InstanceIdentifier entry : source.getRemovedConfigurationData()) {
+            try {
+
+                InstanceIdentifier<?> baEntry = mappingService.fromDataDom(entry);
+                target.removeConfigurationData(baEntry);
+            } catch (DeserializationException e) {
+                LOG.error("Ommiting from BA transaction: {}.", entry, e);
+            }
+        }
+        for (org.opendaylight.yangtools.yang.data.api.InstanceIdentifier entry : source.getRemovedOperationalData()) {
+            try {
+
+                InstanceIdentifier<?> baEntry = mappingService.fromDataDom(entry);
+                target.removeOperationalData(baEntry);
+            } catch (DeserializationException e) {
+                LOG.error("Ommiting from BA transaction: {}.", entry, e);
+            }
+        }
         for (Entry<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> entry : source
                 .getUpdatedConfigurationData().entrySet()) {
             try {
@@ -254,24 +272,6 @@ public class BindingIndependentConnector implements //
                 LOG.error("Ommiting from BA transaction: {}.", entry.getKey(), e);
             }
         }
-        for (org.opendaylight.yangtools.yang.data.api.InstanceIdentifier entry : source.getRemovedConfigurationData()) {
-            try {
-
-                InstanceIdentifier<?> baEntry = mappingService.fromDataDom(entry);
-                target.removeConfigurationData(baEntry);
-            } catch (DeserializationException e) {
-                LOG.error("Ommiting from BA transaction: {}.", entry, e);
-            }
-        }
-        for (org.opendaylight.yangtools.yang.data.api.InstanceIdentifier entry : source.getRemovedOperationalData()) {
-            try {
-
-                InstanceIdentifier<?> baEntry = mappingService.fromDataDom(entry);
-                target.removeOperationalData(baEntry);
-            } catch (DeserializationException e) {
-                LOG.error("Ommiting from BA transaction: {}.", entry, e);
-            }
-        }
         return target;
     }
 
@@ -279,7 +279,7 @@ public class BindingIndependentConnector implements //
         return biDataService;
     }
 
-    protected void setDomDataService(org.opendaylight.controller.sal.core.api.data.DataProviderService biDataService) {
+    protected void setDomDataService(final org.opendaylight.controller.sal.core.api.data.DataProviderService biDataService) {
         this.biDataService = biDataService;
     }
 
@@ -287,7 +287,7 @@ public class BindingIndependentConnector implements //
         return baDataService;
     }
 
-    protected void setBindingDataService(DataProviderService baDataService) {
+    protected void setBindingDataService(final DataProviderService baDataService) {
         this.baDataService = baDataService;
     }
 
@@ -295,21 +295,37 @@ public class BindingIndependentConnector implements //
         return baRpcRegistry;
     }
 
-    protected void setBindingRpcRegistry(RpcProviderRegistry rpcRegistry) {
+    protected void setBindingRpcRegistry(final RpcProviderRegistry rpcRegistry) {
         this.baRpcRegistry = rpcRegistry;
     }
 
     public void startDataForwarding() {
-        checkState(!dataForwarding, "Connector is already forwarding data.");
-        baDataReaderRegistration = baDataService.registerDataReader(ROOT, this);
-        baCommitHandlerRegistration = baDataService.registerCommitHandler(ROOT, bindingToDomCommitHandler);
-        biCommitHandlerRegistration = biDataService.registerCommitHandler(ROOT_BI, domToBindingCommitHandler);
-        baDataService.registerCommitHandlerListener(domToBindingCommitHandler);
+        if(baDataService instanceof AbstractForwardedDataBroker) {
+            dataForwarding = true;
+            return;
+        }
+
+        final DataProviderService baData;
+        if (baDataService instanceof BindingMountPointImpl) {
+            baData = ((BindingMountPointImpl)baDataService).getDataBrokerImpl();
+            LOG.debug("Extracted BA Data provider {} from mount point {}", baData, baDataService);
+        } else {
+            baData = baDataService;
+        }
+
+        if (baData instanceof DataBrokerImpl) {
+            checkState(!dataForwarding, "Connector is already forwarding data.");
+            ((DataBrokerImpl) baData).setDataReadDelegate(this);
+            ((DataBrokerImpl) baData).setRootCommitHandler(bindingToDomCommitHandler);
+            biCommitHandlerRegistration = biDataService.registerCommitHandler(ROOT_BI, domToBindingCommitHandler);
+            baDataService.registerCommitHandlerListener(domToBindingCommitHandler);
+        }
+
         dataForwarding = true;
     }
 
     public void startRpcForwarding() {
-        if (baRpcRegistry != null && biRpcRegistry != null && baRpcRegistry instanceof RouteChangePublisher<?, ?>) {
+        if (biRpcRegistry != null && baRpcRegistry instanceof RouteChangePublisher<?, ?>) {
             checkState(!rpcForwarding, "Connector is already forwarding RPCs");
             domToBindingRpcManager = baRpcRegistry.registerRouteChangeListener(new DomToBindingRpcForwardingManager());
             if (baRpcRegistry instanceof RpcProviderRegistryImpl) {
@@ -330,7 +346,7 @@ public class BindingIndependentConnector implements //
         }
     }
 
-    protected void setMappingService(BindingIndependentMappingService mappingService) {
+    protected void setMappingService(final BindingIndependentMappingService mappingService) {
         this.mappingService = mappingService;
     }
 
@@ -340,17 +356,17 @@ public class BindingIndependentConnector implements //
     }
 
     @Override
-    public void onSessionInitiated(ProviderSession session) {
+    public void onSessionInitiated(final ProviderSession session) {
         setDomDataService(session.getService(org.opendaylight.controller.sal.core.api.data.DataProviderService.class));
         setDomRpcRegistry(session.getService(RpcProvisionRegistry.class));
 
     }
 
-    public <T extends RpcService> void onRpcRouterCreated(Class<T> serviceType, RpcRouter<T> router) {
+    public <T extends RpcService> void onRpcRouterCreated(final Class<T> serviceType, final RpcRouter<T> router) {
 
     }
 
-    public void setDomRpcRegistry(RpcProvisionRegistry registry) {
+    public void setDomRpcRegistry(final RpcProvisionRegistry registry) {
         biRpcRegistry = registry;
     }
 
@@ -372,8 +388,8 @@ public class BindingIndependentConnector implements //
         private final DataModification<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> modification;
 
         public DomToBindingTransaction(
-                org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction backing,
-                DataModification<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> modification) {
+                final org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction backing,
+                final DataModification<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> modification) {
             super();
             this.backing = backing;
             this.modification = modification;
@@ -411,8 +427,8 @@ public class BindingIndependentConnector implements //
         private final DataModificationTransaction backing;
         private final DataModification<InstanceIdentifier<? extends DataObject>, DataObject> modification;
 
-        public BindingToDomTransaction(DataModificationTransaction backing,
-                DataModification<InstanceIdentifier<? extends DataObject>, DataObject> modification) {
+        public BindingToDomTransaction(final DataModificationTransaction backing,
+                final DataModification<InstanceIdentifier<? extends DataObject>, DataObject> modification) {
             this.backing = backing;
             this.modification = modification;
             domOpenedTransactions.put(backing.getIdentifier(), this);
@@ -450,7 +466,7 @@ public class BindingIndependentConnector implements //
 
         @Override
         public org.opendaylight.controller.md.sal.common.api.data.DataCommitHandler.DataCommitTransaction<InstanceIdentifier<? extends DataObject>, DataObject> requestCommit(
-                DataModification<InstanceIdentifier<? extends DataObject>, DataObject> bindingTransaction) {
+                final DataModification<InstanceIdentifier<? extends DataObject>, DataObject> bindingTransaction) {
 
             /**
              * Transaction was created as DOM transaction, in that case we do
@@ -473,7 +489,7 @@ public class BindingIndependentConnector implements //
             DataCommitHandler<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> {
 
         @Override
-        public void onRegister(DataCommitHandlerRegistration<InstanceIdentifier<? extends DataObject>, DataObject> registration) {
+        public void onRegister(final DataCommitHandlerRegistration<InstanceIdentifier<? extends DataObject>, DataObject> registration) {
 
             org.opendaylight.yangtools.yang.data.api.InstanceIdentifier domPath = mappingService.toDataDom(registration
                     .getPath());
@@ -481,14 +497,14 @@ public class BindingIndependentConnector implements //
         }
 
         @Override
-        public void onUnregister(DataCommitHandlerRegistration<InstanceIdentifier<? extends DataObject>, DataObject> registration) {
+        public void onUnregister(final DataCommitHandlerRegistration<InstanceIdentifier<? extends DataObject>, DataObject> registration) {
             // NOOP for now
             // FIXME: do registration based on only active commit handlers.
         }
 
         @Override
         public org.opendaylight.controller.md.sal.common.api.data.DataCommitHandler.DataCommitTransaction<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> requestCommit(
-                DataModification<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> domTransaction) {
+                final DataModification<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, CompositeNode> domTransaction) {
             Object identifier = domTransaction.getIdentifier();
 
             /**
@@ -526,34 +542,34 @@ public class BindingIndependentConnector implements //
             return registryImpl;
         }
 
-        public void setRegistryImpl(RpcProviderRegistryImpl registryImpl) {
+        public void setRegistryImpl(final RpcProviderRegistryImpl registryImpl) {
             this.registryImpl = registryImpl;
         }
 
         @Override
-        public void onGlobalRpcRegistered(Class<? extends RpcService> cls) {
+        public void onGlobalRpcRegistered(final Class<? extends RpcService> cls) {
             getRpcForwarder(cls, null);
         }
 
         @Override
-        public void onGlobalRpcUnregistered(Class<? extends RpcService> cls) {
+        public void onGlobalRpcUnregistered(final Class<? extends RpcService> cls) {
             // NOOP
         }
 
         @Override
-        public void onRpcRouterCreated(RpcRouter<?> router) {
+        public void onRpcRouterCreated(final RpcRouter<?> router) {
             Class<? extends BaseIdentity> ctx = router.getContexts().iterator().next();
             getRpcForwarder(router.getServiceType(), ctx);
         }
 
         @Override
-        public void onRouteChange(RouteChange<RpcContextIdentifier, InstanceIdentifier<?>> change) {
+        public void onRouteChange(final RouteChange<RpcContextIdentifier, InstanceIdentifier<?>> change) {
             for (Entry<RpcContextIdentifier, Set<InstanceIdentifier<?>>> entry : change.getAnnouncements().entrySet()) {
                 bindingRoutesAdded(entry);
             }
         }
 
-        private void bindingRoutesAdded(Entry<RpcContextIdentifier, Set<InstanceIdentifier<?>>> entry) {
+        private void bindingRoutesAdded(final Entry<RpcContextIdentifier, Set<InstanceIdentifier<?>>> entry) {
             Class<? extends BaseIdentity> context = entry.getKey().getRoutingContext();
             Class<? extends RpcService> service = entry.getKey().getRpcService();
             if (context != null) {
@@ -561,8 +577,8 @@ public class BindingIndependentConnector implements //
             }
         }
 
-        private DomToBindingRpcForwarder getRpcForwarder(Class<? extends RpcService> service,
-                Class<? extends BaseIdentity> context) {
+        private DomToBindingRpcForwarder getRpcForwarder(final Class<? extends RpcService> service,
+                final Class<? extends BaseIdentity> context) {
             DomToBindingRpcForwarder potential = forwarders.get(service);
             if (potential != null) {
                 return potential;
@@ -587,7 +603,7 @@ public class BindingIndependentConnector implements //
         private final Map<QName, RpcInvocationStrategy> strategiesByQName = new HashMap<>();
         private final WeakHashMap<Method, RpcInvocationStrategy> strategiesByMethod = new WeakHashMap<>();
 
-        public DomToBindingRpcForwarder(Class<? extends RpcService> service) {
+        public DomToBindingRpcForwarder(final Class<? extends RpcService> service) {
             this.rpcServiceType = new WeakReference<Class<? extends RpcService>>(service);
             this.supportedRpcs = mappingService.getRpcQNamesFor(service);
             try {
@@ -610,7 +626,7 @@ public class BindingIndependentConnector implements //
          * @param service
          * @param context
          */
-        public DomToBindingRpcForwarder(Class<? extends RpcService> service, Class<? extends BaseIdentity> context) {
+        public DomToBindingRpcForwarder(final Class<? extends RpcService> service, final Class<? extends BaseIdentity> context) {
             this.rpcServiceType = new WeakReference<Class<? extends RpcService>>(service);
             this.supportedRpcs = mappingService.getRpcQNamesFor(service);
             Builder<RoutedRpcRegistration> registrationsBuilder = ImmutableSet
@@ -629,8 +645,8 @@ public class BindingIndependentConnector implements //
             registrations = registrationsBuilder.build();
         }
 
-        public void registerPaths(Class<? extends BaseIdentity> context, Class<? extends RpcService> service,
-                Set<InstanceIdentifier<?>> set) {
+        public void registerPaths(final Class<? extends BaseIdentity> context, final Class<? extends RpcService> service,
+                final Set<InstanceIdentifier<?>> set) {
             QName ctx = BindingReflections.findQName(context);
             for (org.opendaylight.yangtools.yang.data.api.InstanceIdentifier path : FluentIterable.from(set).transform(
                     toDOMInstanceIdentifier)) {
@@ -642,7 +658,7 @@ public class BindingIndependentConnector implements //
 
 
         @Override
-        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
+        public Object invoke(final Object proxy, final Method method, final Object[] args) throws Throwable {
             if (EQUALS_METHOD.equals(method)) {
                 return false;
             }
@@ -656,8 +672,8 @@ public class BindingIndependentConnector implements //
             return strategy.forwardToDomBroker(null);
         }
 
-        public void removePaths(Class<? extends BaseIdentity> context, Class<? extends RpcService> service,
-                Set<InstanceIdentifier<?>> set) {
+        public void removePaths(final Class<? extends BaseIdentity> context, final Class<? extends RpcService> service,
+                final Set<InstanceIdentifier<?>> set) {
             QName ctx = BindingReflections.findQName(context);
             for (org.opendaylight.yangtools.yang.data.api.InstanceIdentifier path : FluentIterable.from(set).transform(
                     toDOMInstanceIdentifier)) {
@@ -684,8 +700,9 @@ public class BindingIndependentConnector implements //
             }
         }
 
+
         @Override
-        public RpcResult<CompositeNode> invokeRpc(QName rpc, CompositeNode domInput) {
+        public ListenableFuture<RpcResult<CompositeNode>> invokeRpc(final QName rpc, final CompositeNode domInput) {
             checkArgument(rpc != null);
             checkArgument(domInput != null);
 
@@ -694,14 +711,15 @@ public class BindingIndependentConnector implements //
             RpcService rpcService = baRpcRegistry.getRpcService(rpcType);
             checkState(rpcService != null);
             CompositeNode domUnwrappedInput = domInput.getFirstCompositeByName(QName.create(rpc, "input"));
+
             try {
-                return resolveInvocationStrategy(rpc).invokeOn(rpcService, domUnwrappedInput);
+                return Futures.immediateFuture(resolveInvocationStrategy(rpc).invokeOn(rpcService, domUnwrappedInput));
             } catch (Exception e) {
-                throw new IllegalStateException(e);
+                return Futures.immediateFailedFuture(e);
             }
         }
 
-        private RpcInvocationStrategy resolveInvocationStrategy(QName rpc) {
+        private RpcInvocationStrategy resolveInvocationStrategy(final QName rpc) {
             return strategiesByQName.get(rpc);
         }
 
@@ -749,7 +767,7 @@ public class BindingIndependentConnector implements //
         protected final Method targetMethod;
         protected final QName rpc;
 
-        public RpcInvocationStrategy(QName rpc, Method targetMethod) {
+        public RpcInvocationStrategy(final QName rpc, final Method targetMethod) {
             this.targetMethod = targetMethod;
             this.rpc = rpc;
         }
@@ -759,7 +777,7 @@ public class BindingIndependentConnector implements //
         public abstract RpcResult<CompositeNode> uncheckedInvoke(RpcService rpcService, CompositeNode domInput)
                 throws Exception;
 
-        public RpcResult<CompositeNode> invokeOn(RpcService rpcService, CompositeNode domInput) throws Exception {
+        public RpcResult<CompositeNode> invokeOn(final RpcService rpcService, final CompositeNode domInput) throws Exception {
             return uncheckedInvoke(rpcService, domInput);
         }
     }
@@ -773,8 +791,8 @@ public class BindingIndependentConnector implements //
         private final WeakReference<Class> outputClass;
 
         @SuppressWarnings({ "rawtypes", "unchecked" })
-        public DefaultInvocationStrategy(QName rpc, Method targetMethod, Class<?> outputClass,
-                Class<? extends DataContainer> inputClass) {
+        public DefaultInvocationStrategy(final QName rpc, final Method targetMethod, final Class<?> outputClass,
+                final Class<? extends DataContainer> inputClass) {
             super(rpc, targetMethod);
             this.outputClass = new WeakReference(outputClass);
             this.inputClass = new WeakReference(inputClass);
@@ -782,7 +800,7 @@ public class BindingIndependentConnector implements //
 
         @SuppressWarnings("unchecked")
         @Override
-        public RpcResult<CompositeNode> uncheckedInvoke(RpcService rpcService, CompositeNode domInput) throws Exception {
+        public RpcResult<CompositeNode> uncheckedInvoke(final RpcService rpcService, final CompositeNode domInput) throws Exception {
             DataContainer bindingInput = mappingService.dataObjectFromDataDom(inputClass.get(), domInput);
             Future<RpcResult<?>> futureResult = (Future<RpcResult<?>>) targetMethod.invoke(rpcService, bindingInput);
             if (futureResult == null) {
@@ -798,31 +816,35 @@ public class BindingIndependentConnector implements //
         }
 
         @Override
-        public Future<RpcResult<?>> forwardToDomBroker(DataObject input) {
-            if(biRpcRegistry != null) {
-                CompositeNode xml = mappingService.toDataDom(input);
-                CompositeNode wrappedXml = ImmutableCompositeNode.create(rpc, ImmutableList.<Node<?>> of(xml));
-                RpcResult<CompositeNode> result = biRpcRegistry.invokeRpc(rpc, wrappedXml);
-                Object baResultValue = null;
-                if (result.getResult() != null) {
-                    baResultValue = mappingService.dataObjectFromDataDom(outputClass.get(), result.getResult());
-                }
-                RpcResult<?> baResult = Rpcs.getRpcResult(result.isSuccessful(), baResultValue, result.getErrors());
-                return Futures.<RpcResult<?>> immediateFuture(baResult);
+        public ListenableFuture<RpcResult<?>> forwardToDomBroker(final DataObject input) {
+            if(biRpcRegistry == null) {
+                return Futures.<RpcResult<?>> immediateFuture(Rpcs.getRpcResult(false));
             }
-            return Futures.<RpcResult<?>> immediateFuture(Rpcs.getRpcResult(false));
-        }
 
+            CompositeNode xml = mappingService.toDataDom(input);
+            CompositeNode wrappedXml = ImmutableCompositeNode.create(rpc, ImmutableList.<Node<?>> of(xml));
+
+            return Futures.transform(biRpcRegistry.invokeRpc(rpc, wrappedXml), new Function<RpcResult<CompositeNode>, RpcResult<?>>() {
+                @Override
+                public RpcResult<?> apply(RpcResult<CompositeNode> input) {
+                    Object baResultValue = null;
+                    if (input.getResult() != null) {
+                        baResultValue = mappingService.dataObjectFromDataDom(outputClass.get(), input.getResult());
+                    }
+                    return Rpcs.getRpcResult(input.isSuccessful(), baResultValue, input.getErrors());
+                }
+            });
+        }
     }
 
     private class NoInputNoOutputInvocationStrategy extends RpcInvocationStrategy {
 
-        public NoInputNoOutputInvocationStrategy(QName rpc, Method targetMethod) {
+        public NoInputNoOutputInvocationStrategy(final QName rpc, final Method targetMethod) {
             super(rpc, targetMethod);
         }
 
         @Override
-        public RpcResult<CompositeNode> uncheckedInvoke(RpcService rpcService, CompositeNode domInput) throws Exception {
+        public RpcResult<CompositeNode> uncheckedInvoke(final RpcService rpcService, final CompositeNode domInput) throws Exception {
             @SuppressWarnings("unchecked")
             Future<RpcResult<Void>> result = (Future<RpcResult<Void>>) targetMethod.invoke(rpcService);
             RpcResult<Void> bindingResult = result.get();
@@ -830,7 +852,7 @@ public class BindingIndependentConnector implements //
         }
 
         @Override
-        public Future<RpcResult<?>> forwardToDomBroker(DataObject input) {
+        public Future<RpcResult<?>> forwardToDomBroker(final DataObject input) {
             return Futures.immediateFuture(null);
         }
     }
@@ -842,15 +864,15 @@ public class BindingIndependentConnector implements //
         private final WeakReference<Class> inputClass;
 
         @SuppressWarnings({ "rawtypes", "unchecked" })
-        public NoOutputInvocationStrategy(QName rpc, Method targetMethod,
-                Class<? extends DataContainer> inputClass) {
+        public NoOutputInvocationStrategy(final QName rpc, final Method targetMethod,
+                final Class<? extends DataContainer> inputClass) {
             super(rpc,targetMethod);
             this.inputClass = new WeakReference(inputClass);
         }
 
 
         @Override
-        public RpcResult<CompositeNode> uncheckedInvoke(RpcService rpcService, CompositeNode domInput) throws Exception {
+        public RpcResult<CompositeNode> uncheckedInvoke(final RpcService rpcService, final CompositeNode domInput) throws Exception {
             DataContainer bindingInput = mappingService.dataObjectFromDataDom(inputClass.get(), domInput);
             Future<RpcResult<?>> result = (Future<RpcResult<?>>) targetMethod.invoke(rpcService, bindingInput);
             if (result == null) {
@@ -861,18 +883,21 @@ public class BindingIndependentConnector implements //
         }
 
         @Override
-        public Future<RpcResult<?>> forwardToDomBroker(DataObject input) {
-            if(biRpcRegistry != null) {
-                CompositeNode xml = mappingService.toDataDom(input);
-                CompositeNode wrappedXml = ImmutableCompositeNode.create(rpc,ImmutableList.<Node<?>>of(xml));
-                RpcResult<CompositeNode> result = biRpcRegistry.invokeRpc(rpc, wrappedXml);
-                Object baResultValue = null;
-                RpcResult<?> baResult = Rpcs.<Void>getRpcResult(result.isSuccessful(), null, result.getErrors());
-                return Futures.<RpcResult<?>>immediateFuture(baResult);
+        public ListenableFuture<RpcResult<?>> forwardToDomBroker(final DataObject input) {
+            if(biRpcRegistry == null) {
+                return Futures.<RpcResult<?>> immediateFuture(Rpcs.getRpcResult(false));
             }
-            return Futures.<RpcResult<?>>immediateFuture(Rpcs.getRpcResult(false));
-        }
 
+            CompositeNode xml = mappingService.toDataDom(input);
+            CompositeNode wrappedXml = ImmutableCompositeNode.create(rpc,ImmutableList.<Node<?>>of(xml));
+
+            return Futures.transform(biRpcRegistry.invokeRpc(rpc, wrappedXml), new Function<RpcResult<CompositeNode>, RpcResult<?>>() {
+                @Override
+                public RpcResult<?> apply(RpcResult<CompositeNode> input) {
+                    return Rpcs.<Void>getRpcResult(input.isSuccessful(), null, input.getErrors());
+                }
+            });
+        }
     }
 
     public boolean isRpcForwarding() {
@@ -891,12 +916,12 @@ public class BindingIndependentConnector implements //
         return mappingService;
     }
 
-    public void setBindingNotificationService(NotificationProviderService baService) {
+    public void setBindingNotificationService(final NotificationProviderService baService) {
         this.baNotifyService = baService;
 
     }
 
-    public void setDomNotificationService(NotificationPublishService domService) {
+    public void setDomNotificationService(final NotificationPublishService domService) {
         this.domNotificationService = domService;
     }
 
@@ -911,7 +936,7 @@ public class BindingIndependentConnector implements //
         }
 
         @Override
-        public void onNotification(CompositeNode notification) {
+        public void onNotification(final CompositeNode notification) {
             QName qname = notification.getNodeType();
             WeakReference<Class<? extends Notification>> potential = notifications.get(qname);
             if (potential != null) {
@@ -928,7 +953,7 @@ public class BindingIndependentConnector implements //
         }
 
         @Override
-        public void onNotificationSubscribtion(Class<? extends Notification> notificationType) {
+        public void onNotificationSubscribtion(final Class<? extends Notification> notificationType) {
             QName qname = BindingReflections.findQName(notificationType);
             if (qname != null) {
                 WeakReference<Class<? extends Notification>> already = notifications.putIfAbsent(qname,