import com.google.common.base.Function;
import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.SettableFuture;
+import com.google.common.util.concurrent.ListenableFuture;
+import java.math.BigInteger;
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.device.Xid;
+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.binding.DataObject;
-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;
-import java.math.BigInteger;
-import java.util.concurrent.Future;
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");
-
- // protected OFRpcTaskContext rpcTaskContext;
- public short version;
- public BigInteger datapathId;
- public RequestContextStack requestContextStack;
- public DeviceContext deviceContext;
- public ConnectionAdapter primaryConnectionAdapter;
-
- public CommonService() {
- }
+ private static final BigInteger PRIMARY_CONNECTION = BigInteger.ZERO;
+
+ private final short version;
+ private final BigInteger datapathId;
+ private final RequestContextStack requestContextStack;
+ private final DeviceContext deviceContext;
+ private final ConnectionAdapter primaryConnectionAdapter;
+ private final MessageSpy messageSpy;
- public CommonService(final RequestContextStack requestContextStack, DeviceContext deviceContext) {
- this.requestContextStack = requestContextStack;
+ 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;
return primaryConnectionAdapter;
}
- // TODO uncomment when getAuxiali.... will be merged to APIs
- // final ConnectionContext auxiliaryConnectionContext =
- // deviceContext.getAuxiliaryConnectionContext(connectionID);
- final ConnectionContext auxiliaryConnectionContext = null;
+ final ConnectionContext auxiliaryConnectionContext =
+ deviceContext.getAuxiliaryConnectiobContexts(connectionID);
if (auxiliaryConnectionContext != null) {
return auxiliaryConnectionContext.getConnectionAdapter();
}
return primaryConnectionAdapter;
}
- public <T extends DataObject, F> Future<RpcResult<T>> handleServiceCall(final BigInteger connectionID,
- final Function<DataCrate<T>, Future<RpcResult<F>>> function) {
- LOG.debug("Calling the FlowMod RPC method on MessageDispatchService");
+ /**
+ * @param connectionID connection identifier
+ * @param function data sender
+ * @param <T> rpc result backend type
+ * @param <F> final rpc backend type
+ * @return
+ */
+ public <T, F> ListenableFuture<RpcResult<T>> handleServiceCall(final BigInteger connectionID,
+ final Function<DataCrate<T>, ListenableFuture<RpcResult<F>>> function) {
+ DataCrateBuilder<T> dataCrateBuilder = DataCrateBuilder.<T>builder();
+ return handleServiceCall(function, dataCrateBuilder);
+ }
+
+ public <T, F> ListenableFuture<RpcResult<T>> handleServiceCall(final Function<DataCrate<T>, ListenableFuture<RpcResult<F>>> function) {
+ DataCrateBuilder<T> dataCrateBuilder = DataCrateBuilder.<T>builder();
+ return handleServiceCall(function, dataCrateBuilder);
+ }
- final RequestContext<T> requestContext = requestContextStack.createRequestContext();
- final SettableFuture<RpcResult<T>> result = requestContextStack.storeOrFail(requestContext);
- final DataCrate<T> dataCrate = DataCrateBuilder.<T>builder().setiDConnection(connectionID)
- .setRequestContext(requestContext).build();
- if (!result.isDone()) {
- final Future<RpcResult<F>> resultFromOFLib = function.apply(dataCrate);
+ /**
+ * @param <T>
+ * @param <F>
+ * @param function
+ * @param dataCrateBuilder predefined data
+ * @return
+ */
+ public final <T, F> ListenableFuture<RpcResult<T>> handleServiceCall(final Function<DataCrate<T>, ListenableFuture<RpcResult<F>>> function,
+ final DataCrateBuilder<T> dataCrateBuilder) {
- final RpcResultConvertor<T> rpcResultConvertor = new RpcResultConvertor<>(requestContext, deviceContext);
- rpcResultConvertor.processResultFromOfJava(resultFromOFLib);
+ 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();
+ }
- } else {
- RequestContextUtil.closeRequstContext(requestContext);
+ Long reservedXid = deviceContext.getReservedXid();
+ if (null == reservedXid) {
+ //retry
+ reservedXid = deviceContext.getReservedXid();
+ if (null == reservedXid) {
+ deviceContext.getMessageSpy().spyMessage(requestContext.getClass(), MessageSpy.STATISTIC_GROUP.TO_SWITCH_RESERVATION_REJECTED);
+ return RequestContextUtil.closeRequestContextWithRpcError(requestContext, "Outbound queue wasn't able to reserve XID.");
+ }
}
- return result;
+ final Xid xid = new Xid(reservedXid);
+ requestContext.setXid(xid);
+ DataCrate<T> dataCrate = dataCrateBuilder.setRequestContext(requestContext)
+ .build();
+ final ListenableFuture<RpcResult<F>> resultFromOFLib;
+
+ LOG.trace("Hooking xid {} to device context - precaution.", requestContext.getXid().getValue());
+ deviceContext.hookRequestCtx(xid, requestContext);
+
+ messageSpy.spyMessage(requestContext.getClass(), MessageSpy.STATISTIC_GROUP.TO_SWITCH_READY_FOR_SUBMIT);
+ function.apply(dataCrate);
+
+ return requestContext.getFuture();
+
+ }
+
+ 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);
+ }
}
Code Review / openflowplugin.git / blobdiff