Make sure RequestContext has a constant XID

[openflowplugin.git] / openflowplugin-impl / src / main / java / org / opendaylight / openflowplugin / impl / services / CommonService.java

diff --git a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/services/CommonService.java b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/services/CommonService.java
index 7ab9586fafd606efff0fcd8b814b324ae249693a..706a46d25f8aa5f42f9e6f650334a5eb78809cf4 100644 (file)
--- a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/services/CommonService.java
+++ b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/services/CommonService.java
@@ -10,16 +10,15 @@ package org.opendaylight.openflowplugin.impl.services;
 import com.google.common.base.Function;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.SettableFuture;
 import java.math.BigInteger;
-import java.util.concurrent.Future;
 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
 import org.opendaylight.openflowplugin.api.openflow.device.RequestContext;
 import org.opendaylight.openflowplugin.api.openflow.device.RequestContextStack;
+import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FeaturesReply;
-import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
+import org.opendaylight.yangtools.yang.common.RpcError;
 import org.opendaylight.yangtools.yang.common.RpcResult;
 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
 import org.slf4j.Logger;
@@ -27,38 +26,59 @@ import org.slf4j.Logger;
 public abstract class CommonService {
     private static final Logger LOG = org.slf4j.LoggerFactory.getLogger(CommonService.class);
     private static final long WAIT_TIME = 2000;
-    protected final static Future<RpcResult<Void>> ERROR_RPC_RESULT = Futures.immediateFuture(RpcResultBuilder
-            .<Void>failed().withError(ErrorType.APPLICATION, "", "Request quota exceeded.").build());
-    protected static final BigInteger PRIMARY_CONNECTION = new BigInteger("0");
+    private static final BigInteger PRIMARY_CONNECTION = BigInteger.ZERO;
 
-    // protected OFRpcTaskContext rpcTaskContext;
-    public short version;
-    public BigInteger datapathId;
-    public RequestContextStack requestContextStack;
-    public DeviceContext deviceContext;
-    public ConnectionAdapter primaryConnectionAdapter;
+    private final short version;
+    private final BigInteger datapathId;
+    private final RequestContextStack requestContextStack;
+    private final DeviceContext deviceContext;
+    private final ConnectionAdapter primaryConnectionAdapter;
+    private final MessageSpy messageSpy;
 
-    /**
-     * @deprecated use {@link #CommonService(RequestContextStack, DeviceContext)}
-     */
-    @Deprecated
-    public CommonService() {
-    }
 
-    public CommonService(final RequestContextStack requestContextStack, DeviceContext deviceContext) {
+    public CommonService(final RequestContextStack requestContextStack, final DeviceContext deviceContext) {
         this.requestContextStack = requestContextStack;
-
         this.deviceContext = deviceContext;
         final FeaturesReply features = this.deviceContext.getPrimaryConnectionContext().getFeatures();
         this.datapathId = features.getDatapathId();
         this.version = features.getVersion();
         this.primaryConnectionAdapter = deviceContext.getPrimaryConnectionContext().getConnectionAdapter();
+        this.messageSpy = deviceContext.getMessageSpy();
+    }
+
+    public static BigInteger getPrimaryConnection() {
+        return PRIMARY_CONNECTION;
+    }
+
+    public short getVersion() {
+        return version;
+    }
+
+    public BigInteger getDatapathId() {
+        return datapathId;
+    }
+
+    public RequestContextStack getRequestContextStack() {
+        return requestContextStack;
+    }
+
+    public DeviceContext getDeviceContext() {
+        return deviceContext;
+    }
+
+    public ConnectionAdapter getPrimaryConnectionAdapter() {
+        return primaryConnectionAdapter;
+    }
+
+    public MessageSpy getMessageSpy() {
+        return messageSpy;
     }
 
     protected long provideWaitTime() {
         return WAIT_TIME;
     }
 
+
     protected ConnectionAdapter provideConnectionAdapter(final BigInteger connectionID) {
         if (connectionID == null) {
             return primaryConnectionAdapter;
@@ -68,7 +88,7 @@ public abstract class CommonService {
         }
 
         final ConnectionContext auxiliaryConnectionContext =
-        deviceContext.getAuxiliaryConnectiobContexts(connectionID);
+                deviceContext.getAuxiliaryConnectiobContexts(connectionID);
         if (auxiliaryConnectionContext != null) {
             return auxiliaryConnectionContext.getConnectionAdapter();
         }
@@ -76,29 +96,46 @@ public abstract class CommonService {
         return primaryConnectionAdapter;
     }
 
-    public <T, F> Future<RpcResult<T>> handleServiceCall(final BigInteger connectionID,
-                                                                            final Function<DataCrate<T>, ListenableFuture<RpcResult<F>>> function) {
-        LOG.debug("Calling the FlowMod RPC method on MessageDispatchService");
 
-        final RequestContext<T> requestContext = requestContextStack.createRequestContext();
-        final SettableFuture<RpcResult<T>> result = requestContextStack.storeOrFail(requestContext);
-        if (!result.isDone()) {
-            final DataCrate<T> dataCrate = DataCrateBuilder.<T>builder().setiDConnection(connectionID)
-                    .setRequestContext(requestContext).build();
-            requestContext.setXid(deviceContext.getNextXid());
+    /**
+     * @param <T>
+     * @param <F>
+     * @param function
+     * @return
+     */
+    public final <T, F> ListenableFuture<RpcResult<T>> handleServiceCall(final Function<RequestContext<T>, ListenableFuture<RpcResult<F>>> function) {
+
+        LOG.trace("Handling general service call");
+        final RequestContext<T> requestContext = createRequestContext();
+        if (requestContext == null) {
+            LOG.trace("Request context refused.");
+            deviceContext.getMessageSpy().spyMessage(null, MessageSpy.STATISTIC_GROUP.TO_SWITCH_DISREGARDED);
+            return failedFuture();
+        }
+
+        if (requestContext.getXid() == null) {
+            deviceContext.getMessageSpy().spyMessage(requestContext.getClass(), MessageSpy.STATISTIC_GROUP.TO_SWITCH_RESERVATION_REJECTED);
+            return RequestContextUtil.closeRequestContextWithRpcError(requestContext, "Outbound queue wasn't able to reserve XID.");
+        }
+        final ListenableFuture<RpcResult<F>> resultFromOFLib;
+
+        LOG.trace("Hooking xid {} to device context - precaution.", requestContext.getXid().getValue());
+        deviceContext.hookRequestCtx(requestContext.getXid(), requestContext);
 
-            LOG.trace("Hooking xid {} to device context - precaution.", requestContext.getXid().getValue());
-            deviceContext.hookRequestCtx(requestContext.getXid(), requestContext);
+        messageSpy.spyMessage(requestContext.getClass(), MessageSpy.STATISTIC_GROUP.TO_SWITCH_READY_FOR_SUBMIT);
+        function.apply(requestContext);
 
-            final ListenableFuture<RpcResult<F>> resultFromOFLib = function.apply(dataCrate);
+        return requestContext.getFuture();
 
-            final OFJResult2RequestCtxFuture<T> OFJResult2RequestCtxFuture = new OFJResult2RequestCtxFuture<>(requestContext, deviceContext);
-            OFJResult2RequestCtxFuture.processResultFromOfJava(resultFromOFLib);
+    }
 
-        } else {
-            RequestContextUtil.closeRequstContext(requestContext);
-        }
-        return result;
+    protected final <T> RequestContext<T> createRequestContext() {
+        return requestContextStack.createRequestContext();
     }
 
+    protected static <T> ListenableFuture<RpcResult<T>> failedFuture() {
+        final RpcResult<T> rpcResult = RpcResultBuilder.<T>failed()
+                .withError(RpcError.ErrorType.APPLICATION, "", "Request quota exceeded").build();
+        return Futures.immediateFuture(rpcResult);
+    }
 }