-/**
+/*
* Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
*
* This program and the accompanying materials are made available under the
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
-import java.math.BigInteger;
-import javax.annotation.Nonnull;
+import java.util.function.Function;
+import org.eclipse.jdt.annotation.NonNull;
+import org.eclipse.jdt.annotation.Nullable;
import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueue;
import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
import org.opendaylight.openflowplugin.api.openflow.device.Xid;
import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.EventIdentifier;
import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy;
+import org.opendaylight.openflowplugin.impl.services.util.RequestContextUtil;
+import org.opendaylight.openflowplugin.impl.services.util.ServiceException;
import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
import org.opendaylight.yangtools.yang.binding.DataContainer;
-import org.opendaylight.yangtools.yang.common.RpcError;
+import org.opendaylight.yangtools.yang.common.ErrorTag;
+import org.opendaylight.yangtools.yang.common.ErrorType;
import org.opendaylight.yangtools.yang.common.RpcResult;
import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
+import org.opendaylight.yangtools.yang.common.Uint32;
+import org.opendaylight.yangtools.yang.common.Uint64;
+import org.opendaylight.yangtools.yang.common.Uint8;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-abstract class AbstractService<I, O> {
+public abstract class AbstractService<I, O> {
private static final Logger LOG = LoggerFactory.getLogger(AbstractService.class);
- private static final long WAIT_TIME = 2000;
- private static final BigInteger PRIMARY_CONNECTION = BigInteger.ZERO;
- private final short version;
- private final BigInteger datapathId;
+ private final Uint8 version;
+ private final Uint64 datapathId;
private final RequestContextStack requestContextStack;
private final DeviceContext deviceContext;
private final MessageSpy messageSpy;
this.messageSpy = deviceContext.getMessageSpy();
}
+ public boolean canUseSingleLayerSerialization() {
+ return deviceContext.canUseSingleLayerSerialization();
+ }
+
public EventIdentifier getEventIdentifier() {
return eventIdentifier;
}
this.eventIdentifier = eventIdentifier;
}
- public short getVersion() {
+ public final Uint8 getVersion() {
return version;
}
- public BigInteger getDatapathId() {
+ public final Uint64 getDatapathId() {
return datapathId;
}
return deviceContext;
}
- protected DeviceRegistry getDeviceRegistry() {return deviceContext;}
+ public DeviceRegistry getDeviceRegistry() {
+ return deviceContext;
+ }
- public DeviceInfo getDeviceInfo() {return deviceContext.getDeviceInfo();}
+ public DeviceInfo getDeviceInfo() {
+ return deviceContext.getDeviceInfo();
+ }
- public TxFacade getTxFacade() {return deviceContext;}
+ public TxFacade getTxFacade() {
+ return deviceContext;
+ }
public MessageSpy getMessageSpy() {
return messageSpy;
}
- protected abstract OfHeader buildRequest(Xid xid, I input) throws Exception;
+ protected abstract OfHeader buildRequest(Xid xid, I input) throws ServiceException;
protected abstract FutureCallback<OfHeader> createCallback(RequestContext<O> context, Class<?> requestType);
- public final ListenableFuture<RpcResult<O>> handleServiceCall(@Nonnull final I input) {
+ public ListenableFuture<RpcResult<O>> handleServiceCall(@NonNull final I input) {
+ return handleServiceCall(input, null);
+ }
+
+ public ListenableFuture<RpcResult<O>> handleServiceCall(@NonNull final I input,
+ @Nullable final Function<OfHeader, Boolean> isComplete) {
Preconditions.checkNotNull(input);
- final Class<?> requestType;
- if (input instanceof DataContainer) {
- requestType = ((DataContainer) input).getImplementedInterface();
- } else {
- requestType = input.getClass();
- }
- getMessageSpy().spyMessage(requestType, MessageSpy.STATISTIC_GROUP.TO_SWITCH_ENTERED);
+ final Class<?> requestType = input instanceof DataContainer
+ ? ((DataContainer) input).implementedInterface()
+ : input.getClass();
+
+ getMessageSpy().spyMessage(requestType, MessageSpy.StatisticsGroup.TO_SWITCH_ENTERED);
LOG.trace("Handling general service call");
final RequestContext<O> requestContext = requestContextStack.createRequestContext();
+
if (requestContext == null) {
LOG.trace("Request context refused.");
- getMessageSpy().spyMessage(AbstractService.class, MessageSpy.STATISTIC_GROUP.TO_SWITCH_DISREGARDED);
- return failedFuture();
+ getMessageSpy().spyMessage(AbstractService.class, MessageSpy.StatisticsGroup.TO_SWITCH_DISREGARDED);
+ return Futures.immediateFuture(RpcResultBuilder
+ .<O>failed()
+ .withError(ErrorType.APPLICATION, ErrorTag.ACCESS_DENIED, "Request quota exceeded")
+ .build());
}
if (requestContext.getXid() == null) {
- getMessageSpy().spyMessage(requestContext.getClass(), MessageSpy.STATISTIC_GROUP.TO_SWITCH_RESERVATION_REJECTED);
- return RequestContextUtil.closeRequestContextWithRpcError(requestContext, "Outbound queue wasn't able to reserve XID.");
+ getMessageSpy().spyMessage(requestContext.getClass(),
+ MessageSpy.StatisticsGroup.TO_SWITCH_RESERVATION_REJECTED);
+ return RequestContextUtil
+ .closeRequestContextWithRpcError(requestContext, "Outbound queue wasn't able to reserve XID.");
}
- getMessageSpy().spyMessage(requestContext.getClass(), MessageSpy.STATISTIC_GROUP.TO_SWITCH_READY_FOR_SUBMIT);
+ getMessageSpy().spyMessage(requestContext.getClass(), MessageSpy.StatisticsGroup.TO_SWITCH_READY_FOR_SUBMIT);
final Xid xid = requestContext.getXid();
OfHeader request = null;
try {
request = buildRequest(xid, input);
- Verify.verify(xid.getValue().equals(request.getXid()), "Expected XID %s got %s", xid.getValue(), request.getXid());
- } catch (Exception e) {
- LOG.error("Failed to build request for {}, forfeiting request {}", input, xid.getValue(), e);
- RequestContextUtil.closeRequestContextWithRpcError(requestContext, "failed to build request input: " + e.getMessage());
+ Verify.verify(xid.getValue().equals(request.getXid()),
+ "Expected XID %s got %s",
+ xid.getValue(),
+ request.getXid());
+ } catch (ServiceException ex) {
+ LOG.error("Failed to build request for {}, forfeiting request {}", input, xid.getValue(), ex);
+ RequestContextUtil.closeRequestContextWithRpcError(requestContext,
+ "failed to build request input: " + ex.getMessage());
} finally {
- final OutboundQueue outboundQueue = getDeviceContext().getPrimaryConnectionContext().getOutboundQueueProvider();
- outboundQueue.commitEntry(xid.getValue(), request, createCallback(requestContext, requestType));
+ final OutboundQueue outboundQueue =
+ getDeviceContext().getPrimaryConnectionContext().getOutboundQueueProvider();
+
+ final Uint32 queueXid = xid.getValue();
+ if (isComplete != null) {
+ outboundQueue.commitEntry(queueXid, request, createCallback(requestContext, requestType), isComplete);
+ } else {
+ outboundQueue.commitEntry(queueXid, request, createCallback(requestContext, requestType));
+ }
}
return requestContext.getFuture();
}
-
- 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