2 * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
8 package org.opendaylight.openflowplugin.impl.services;
10 import com.google.common.base.Preconditions;
11 import com.google.common.base.Verify;
12 import com.google.common.util.concurrent.FutureCallback;
13 import com.google.common.util.concurrent.Futures;
14 import com.google.common.util.concurrent.ListenableFuture;
15 import java.math.BigInteger;
16 import javax.annotation.Nonnull;
17 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
18 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueue;
19 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
20 import org.opendaylight.openflowplugin.api.openflow.device.RequestContext;
21 import org.opendaylight.openflowplugin.api.openflow.device.RequestContextStack;
22 import org.opendaylight.openflowplugin.api.openflow.device.Xid;
23 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.EventIdentifier;
24 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy;
25 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FeaturesReply;
26 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
27 import org.opendaylight.yangtools.yang.binding.DataContainer;
28 import org.opendaylight.yangtools.yang.common.RpcError;
29 import org.opendaylight.yangtools.yang.common.RpcResult;
30 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
31 import org.slf4j.Logger;
32 import org.slf4j.LoggerFactory;
34 abstract class AbstractService<I, O> {
35 private static final Logger LOG = LoggerFactory.getLogger(AbstractService.class);
36 private static final long WAIT_TIME = 2000;
37 private static final BigInteger PRIMARY_CONNECTION = BigInteger.ZERO;
39 private final short version;
40 private final BigInteger datapathId;
41 private final RequestContextStack requestContextStack;
42 private final DeviceContext deviceContext;
43 private final ConnectionAdapter primaryConnectionAdapter;
44 private final MessageSpy messageSpy;
45 private EventIdentifier eventIdentifier;
47 public AbstractService(final RequestContextStack requestContextStack, final DeviceContext deviceContext) {
48 this.requestContextStack = requestContextStack;
49 this.deviceContext = deviceContext;
50 final FeaturesReply features = this.deviceContext.getPrimaryConnectionContext().getFeatures();
51 this.datapathId = features.getDatapathId();
52 this.version = features.getVersion();
53 this.primaryConnectionAdapter = deviceContext.getPrimaryConnectionContext().getConnectionAdapter();
54 this.messageSpy = deviceContext.getMessageSpy();
57 public EventIdentifier getEventIdentifier() {
58 return eventIdentifier;
61 public void setEventIdentifier(final EventIdentifier eventIdentifier) {
62 this.eventIdentifier = eventIdentifier;
65 public short getVersion() {
69 public BigInteger getDatapathId() {
73 public RequestContextStack getRequestContextStack() {
74 return requestContextStack;
77 public DeviceContext getDeviceContext() {
81 public MessageSpy getMessageSpy() {
85 protected abstract OfHeader buildRequest(Xid xid, I input) throws Exception;
87 protected abstract FutureCallback<OfHeader> createCallback(RequestContext<O> context, Class<?> requestType);
89 public final ListenableFuture<RpcResult<O>> handleServiceCall(@Nonnull final I input) {
90 Preconditions.checkNotNull(input);
92 final Class<?> requestType;
93 if (input instanceof DataContainer) {
94 requestType = ((DataContainer) input).getImplementedInterface();
96 requestType = input.getClass();
98 getMessageSpy().spyMessage(requestType, MessageSpy.STATISTIC_GROUP.TO_SWITCH_ENTERED);
100 LOG.trace("Handling general service call");
101 final RequestContext<O> requestContext = requestContextStack.createRequestContext();
102 if (requestContext == null) {
103 LOG.trace("Request context refused.");
104 deviceContext.getMessageSpy().spyMessage(AbstractService.class, MessageSpy.STATISTIC_GROUP.TO_SWITCH_DISREGARDED);
105 return failedFuture();
108 if (requestContext.getXid() == null) {
109 deviceContext.getMessageSpy().spyMessage(requestContext.getClass(), MessageSpy.STATISTIC_GROUP.TO_SWITCH_RESERVATION_REJECTED);
110 return RequestContextUtil.closeRequestContextWithRpcError(requestContext, "Outbound queue wasn't able to reserve XID.");
113 messageSpy.spyMessage(requestContext.getClass(), MessageSpy.STATISTIC_GROUP.TO_SWITCH_READY_FOR_SUBMIT);
115 final Xid xid = requestContext.getXid();
116 OfHeader request = null;
118 request = buildRequest(xid, input);
119 Verify.verify(xid.getValue().equals(request.getXid()), "Expected XID %s got %s", xid.getValue(), request.getXid());
120 } catch (Exception e) {
121 LOG.error("Failed to build request for {}, forfeiting request {}", input, xid.getValue(), e);
122 RequestContextUtil.closeRequestContextWithRpcError(requestContext, "failed to build request input: " + e.getMessage());
124 final OutboundQueue outboundQueue = getDeviceContext().getPrimaryConnectionContext().getOutboundQueueProvider();
125 outboundQueue.commitEntry(xid.getValue(), request, createCallback(requestContext, requestType));
128 return requestContext.getFuture();
131 protected static <T> ListenableFuture<RpcResult<T>> failedFuture() {
132 final RpcResult<T> rpcResult = RpcResultBuilder.<T>failed()
133 .withError(RpcError.ErrorType.APPLICATION, "", "Request quota exceeded").build();
134 return Futures.immediateFuture(rpcResult);