Merge "Implement PortNumberBuilder"

[controller.git] / opendaylight / md-sal / sal-dom-broker / src / main / java / org / opendaylight / controller / sal / dom / broker / impl / SchemaAwareRpcBroker.java

diff --git a/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/sal/dom/broker/impl/SchemaAwareRpcBroker.java b/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/sal/dom/broker/impl/SchemaAwareRpcBroker.java
index 598361c3ae3cbf7e41eed5a69ea9cf0d74727702..32139308b11230b806467f95d078c1608077a1f6 100644 (file)
--- a/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/sal/dom/broker/impl/SchemaAwareRpcBroker.java
+++ b/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/sal/dom/broker/impl/SchemaAwareRpcBroker.java
@@ -14,6 +14,7 @@ import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
+import com.google.common.base.Preconditions;
 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.impl.routing.RoutingUtils;
@@ -46,6 +47,7 @@ import com.google.common.base.Optional;
 import com.google.common.collect.FluentIterable;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
+import com.google.common.util.concurrent.ListenableFuture;
 
 public class SchemaAwareRpcBroker implements RpcRouter, Identifiable<String>, RoutedRpcDefaultImplementation {
 
@@ -85,16 +87,16 @@ public class SchemaAwareRpcBroker implements RpcRouter, Identifiable<String>, Ro
         this.schemaProvider = schemaProvider;
     }
 
-  public RoutedRpcDefaultImplementation getRoutedRpcDefaultDelegate() {
-    return defaultDelegate;
-  }
+    public RoutedRpcDefaultImplementation getRoutedRpcDefaultDelegate() {
+        return defaultDelegate;
+    }
 
     @Override
-  public void setRoutedRpcDefaultDelegate(RoutedRpcDefaultImplementation defaultDelegate) {
-    this.defaultDelegate = defaultDelegate;
-  }
+    public void setRoutedRpcDefaultDelegate(RoutedRpcDefaultImplementation defaultDelegate) {
+        this.defaultDelegate = defaultDelegate;
+    }
 
-  @Override
+    @Override
     public RoutedRpcRegistration addRoutedRpcImplementation(QName rpcType, RpcImplementation implementation) {
         checkArgument(rpcType != null, "RPC Type should not be null");
         checkArgument(implementation != null, "RPC Implementatoin should not be null");
@@ -139,17 +141,34 @@ public class SchemaAwareRpcBroker implements RpcRouter, Identifiable<String>, Ro
         RpcDefinition definition = findRpcDefinition(rpcType);
         checkArgument(!isRoutedRpc(definition), "RPC Type must not be routed.");
         GlobalRpcRegistration reg = new GlobalRpcRegistration(rpcType, implementation, this);
-        implementations.putIfAbsent(rpcType, implementation);
+        final RpcImplementation previous = implementations.putIfAbsent(rpcType, implementation);
+        Preconditions.checkState(previous == null, "Rpc %s is already registered.",rpcType);
+        notifyRpcAdded(rpcType);
         return reg;
     }
 
+    private void notifyRpcAdded(QName rpcType) {
+        for (ListenerRegistration<RpcRegistrationListener> listener : rpcRegistrationListeners) {
+            try {
+                listener.getInstance().onRpcImplementationAdded(rpcType);
+            } catch (Exception ex) {
+                LOG.error("Unhandled exception during invoking listener {}", listener.getInstance(), ex);
+            }
+
+        }
+    }
+
     private boolean isRoutedRpc(RpcDefinition definition) {
         return getRoutingStrategy(definition) instanceof RoutedRpcStrategy;
     }
 
     @Override
     public ListenerRegistration<RpcRegistrationListener> addRpcRegistrationListener(RpcRegistrationListener listener) {
-        return rpcRegistrationListeners.register(listener);
+        ListenerRegistration<RpcRegistrationListener> reg = rpcRegistrationListeners.register(listener);
+        for (QName impl : implementations.keySet()) {
+            listener.onRpcImplementationAdded(impl);
+        }
+        return reg;
     }
 
     @Override
@@ -163,7 +182,7 @@ public class SchemaAwareRpcBroker implements RpcRouter, Identifiable<String>, Ro
     }
 
     @Override
-    public RpcResult<CompositeNode> invokeRpc(QName rpc, CompositeNode input) {
+    public ListenableFuture<RpcResult<CompositeNode>> invokeRpc(QName rpc, CompositeNode input) {
         return findRpcImplemention(rpc).invokeRpc(rpc, input);
     }
 
@@ -173,8 +192,12 @@ public class SchemaAwareRpcBroker implements RpcRouter, Identifiable<String>, Ro
         if (potentialImpl != null) {
             return potentialImpl;
         }
+
         potentialImpl = defaultImplementation;
-        checkState(potentialImpl != null, "Implementation is not available.");
+        if( potentialImpl == null ) {
+            throw new UnsupportedOperationException( "No implementation for this operation is available." );
+        }
+
         return potentialImpl;
     }
 
@@ -224,7 +247,6 @@ public class SchemaAwareRpcBroker implements RpcRouter, Identifiable<String>, Ro
             if (CONTEXT_REFERENCE.equals(extension.getNodeType())) {
                 return Optional.fromNullable(extension.getQName());
             }
-            ;
         }
         return Optional.absent();
     }
@@ -235,7 +257,7 @@ public class SchemaAwareRpcBroker implements RpcRouter, Identifiable<String>, Ro
     }
 
     @Override
-    public RpcResult<CompositeNode> invokeRpc(QName rpc, InstanceIdentifier identifier, CompositeNode input) {
+    public ListenableFuture<RpcResult<CompositeNode>> invokeRpc(QName rpc, InstanceIdentifier identifier, CompositeNode input) {
       checkState(defaultDelegate != null);
       return defaultDelegate.invokeRpc(rpc, identifier, input);
     }
@@ -319,13 +341,14 @@ public class SchemaAwareRpcBroker implements RpcRouter, Identifiable<String>, Ro
         }
 
         @Override
-        public RpcResult<CompositeNode> invokeRpc(QName rpc, CompositeNode input) {
+        public ListenableFuture<RpcResult<CompositeNode>> invokeRpc(QName rpc, CompositeNode input) {
             CompositeNode inputContainer = input.getFirstCompositeByName(QName.create(rpc,"input"));
             checkArgument(inputContainer != null, "Rpc payload must contain input element");
             SimpleNode<?> routeContainer = inputContainer.getFirstSimpleByName(strategy.getLeaf());
             checkArgument(routeContainer != null, "Leaf %s must be set with value", strategy.getLeaf());
             Object route = routeContainer.getValue();
-            checkArgument(route instanceof InstanceIdentifier);
+            checkArgument(route instanceof InstanceIdentifier,
+                          "The routed node %s is not an instance identifier", route);
             RpcImplementation potential = null;
             if (route != null) {
                 RoutedRpcRegImpl potentialReg = implementations.get(route);
@@ -456,7 +479,7 @@ public class SchemaAwareRpcBroker implements RpcRouter, Identifiable<String>, Ro
         try {
         listener.onRouteChange(initial);
         } catch (Exception e) {
-            LOG.error("Unhandled exception during sending initial route change event {} to {}",initial,listener);
+            LOG.error("Unhandled exception during sending initial route change event {} to {}",initial,listener, e);
         }
         return reg;
     }