openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/AbstractConnectionAdapter.java

   1 /*
   2  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   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
   7  */
   8
   9 package org.opendaylight.openflowjava.protocol.impl.core.connection;
  10
  11 import com.google.common.annotations.VisibleForTesting;
  12 import com.google.common.base.Preconditions;
  13 import com.google.common.cache.Cache;
  14 import com.google.common.cache.CacheBuilder;
  15 import com.google.common.cache.RemovalCause;
  16 import com.google.common.cache.RemovalListener;
  17 import com.google.common.cache.RemovalNotification;
  18 import com.google.common.util.concurrent.ListenableFuture;
  19 import com.google.common.util.concurrent.SettableFuture;
  20 import io.netty.channel.Channel;
  21 import io.netty.channel.ChannelFuture;
  22 import io.netty.util.concurrent.GenericFutureListener;
  23 import java.net.InetSocketAddress;
  24 import java.util.concurrent.Future;
  25 import java.util.concurrent.RejectedExecutionException;
  26 import java.util.concurrent.TimeUnit;
  27 import javax.annotation.Nonnull;
  28 import javax.annotation.Nullable;
  29 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.BarrierInput;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.BarrierOutput;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoInput;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoOutput;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoReplyInput;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ExperimenterInput;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowModInput;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetAsyncInput;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetAsyncOutput;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetConfigInput;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetConfigOutput;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetFeaturesInput;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetFeaturesOutput;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetQueueConfigInput;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetQueueConfigOutput;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GroupModInput;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.HelloInput;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MeterModInput;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartRequestInput;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketOutInput;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortModInput;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.RoleRequestInput;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.RoleRequestOutput;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.SetAsyncInput;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.SetConfigInput;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.TableModInput;
  57 import org.opendaylight.yangtools.yang.binding.DataObject;
  58 import org.opendaylight.yangtools.yang.common.RpcResult;
  59 import org.slf4j.Logger;
  60 import org.slf4j.LoggerFactory;
  61
  62 /**
  63  * {@link ConnectionAdapter} interface contains couple of OF message handling approaches.
  64  * {@link AbstractConnectionAdapter} class contains direct RPC processing from OpenflowProtocolService
  65  * {@link org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OpenflowProtocolService}
  66  */
  67 abstract class AbstractConnectionAdapter implements ConnectionAdapter {
  68
  69     private static final Logger LOG = LoggerFactory.getLogger(AbstractConnectionAdapter.class);
  70
  71     /** after this time, RPC future response objects will be thrown away (in minutes) */
  72     private static final int RPC_RESPONSE_EXPIRATION = 1;
  73
  74     private static final Exception QUEUE_FULL_EXCEPTION = new RejectedExecutionException("Output queue is full");
  75
  76     /**
  77      * Default depth of write queue, e.g. we allow these many messages
  78      * to be queued up before blocking producers.
  79      */
  80     private static final int DEFAULT_QUEUE_DEPTH = 1024;
  81
  82     protected static final RemovalListener<RpcResponseKey, ResponseExpectedRpcListener<?>> REMOVAL_LISTENER = new RemovalListener<RpcResponseKey, ResponseExpectedRpcListener<?>>() {
  83         @Override
  84         public void onRemoval(final RemovalNotification<RpcResponseKey, ResponseExpectedRpcListener<?>> notification) {
  85             if (!notification.getCause().equals(RemovalCause.EXPLICIT)) {
  86                 notification.getValue().discard();
  87             }
  88         }
  89     };
  90
  91     protected final Channel channel;
  92     protected final InetSocketAddress address;
  93     protected boolean disconnectOccured = false;
  94     protected final ChannelOutboundQueue output;
  95
  96     /** expiring cache for future rpcResponses */
  97     protected Cache<RpcResponseKey, ResponseExpectedRpcListener<?>> responseCache;
  98
  99
 100     AbstractConnectionAdapter(@Nonnull final Channel channel, @Nullable final InetSocketAddress address) {
 101         this.channel = Preconditions.checkNotNull(channel);
 102         this.address = address;
 103
 104         responseCache = CacheBuilder.newBuilder().concurrencyLevel(1)
 105                 .expireAfterWrite(RPC_RESPONSE_EXPIRATION, TimeUnit.MINUTES).removalListener(REMOVAL_LISTENER).build();
 106         this.output = new ChannelOutboundQueue(channel, DEFAULT_QUEUE_DEPTH, address);
 107         channel.pipeline().addLast(output);
 108     }
 109
 110     @Override
 111     public Future<Boolean> disconnect() {
 112         final ChannelFuture disconnectResult = channel.disconnect();
 113         responseCache.invalidateAll();
 114         disconnectOccured = true;
 115
 116         return handleTransportChannelFuture(disconnectResult);
 117     }
 118
 119     @Override
 120     public Future<RpcResult<BarrierOutput>> barrier(final BarrierInput input) {
 121         return sendToSwitchExpectRpcResultFuture(input, BarrierOutput.class, "barrier-input sending failed");
 122     }
 123
 124     @Override
 125     public Future<RpcResult<EchoOutput>> echo(final EchoInput input) {
 126         return sendToSwitchExpectRpcResultFuture(input, EchoOutput.class, "echo-input sending failed");
 127     }
 128
 129     @Override
 130     public Future<RpcResult<Void>> echoReply(final EchoReplyInput input) {
 131         return sendToSwitchFuture(input, "echo-reply sending failed");
 132     }
 133
 134     @Override
 135     public Future<RpcResult<Void>> experimenter(final ExperimenterInput input) {
 136         return sendToSwitchFuture(input, "experimenter sending failed");
 137     }
 138
 139     @Override
 140     public Future<RpcResult<Void>> flowMod(final FlowModInput input) {
 141         return sendToSwitchFuture(input, "flow-mod sending failed");
 142     }
 143
 144     @Override
 145     public Future<RpcResult<GetConfigOutput>> getConfig(final GetConfigInput input) {
 146         return sendToSwitchExpectRpcResultFuture(input, GetConfigOutput.class, "get-config-input sending failed");
 147     }
 148
 149     @Override
 150     public Future<RpcResult<GetFeaturesOutput>> getFeatures(final GetFeaturesInput input) {
 151         return sendToSwitchExpectRpcResultFuture(input, GetFeaturesOutput.class, "get-features-input sending failed");
 152     }
 153
 154     @Override
 155     public Future<RpcResult<GetQueueConfigOutput>> getQueueConfig(final GetQueueConfigInput input) {
 156         return sendToSwitchExpectRpcResultFuture(input, GetQueueConfigOutput.class,
 157                 "get-queue-config-input sending failed");
 158     }
 159
 160     @Override
 161     public Future<RpcResult<Void>> groupMod(final GroupModInput input) {
 162         return sendToSwitchFuture(input, "group-mod-input sending failed");
 163     }
 164
 165     @Override
 166     public Future<RpcResult<Void>> hello(final HelloInput input) {
 167         return sendToSwitchFuture(input, "hello-input sending failed");
 168     }
 169
 170     @Override
 171     public Future<RpcResult<Void>> meterMod(final MeterModInput input) {
 172         return sendToSwitchFuture(input, "meter-mod-input sending failed");
 173     }
 174
 175     @Override
 176     public Future<RpcResult<Void>> packetOut(final PacketOutInput input) {
 177         return sendToSwitchFuture(input, "packet-out-input sending failed");
 178     }
 179
 180     @Override
 181     public Future<RpcResult<Void>> multipartRequest(final MultipartRequestInput input) {
 182         return sendToSwitchFuture(input, "multi-part-request sending failed");
 183     }
 184
 185     @Override
 186     public Future<RpcResult<Void>> portMod(final PortModInput input) {
 187         return sendToSwitchFuture(input, "port-mod-input sending failed");
 188     }
 189
 190     @Override
 191     public Future<RpcResult<RoleRequestOutput>> roleRequest(final RoleRequestInput input) {
 192         return sendToSwitchExpectRpcResultFuture(input, RoleRequestOutput.class,
 193                 "role-request-config-input sending failed");
 194     }
 195
 196     @Override
 197     public Future<RpcResult<Void>> setConfig(final SetConfigInput input) {
 198         return sendToSwitchFuture(input, "set-config-input sending failed");
 199     }
 200
 201     @Override
 202     public Future<RpcResult<Void>> tableMod(final TableModInput input) {
 203         return sendToSwitchFuture(input, "table-mod-input sending failed");
 204     }
 205
 206     @Override
 207     public Future<RpcResult<GetAsyncOutput>> getAsync(final GetAsyncInput input) {
 208         return sendToSwitchExpectRpcResultFuture(input, GetAsyncOutput.class, "get-async-input sending failed");
 209     }
 210
 211     @Override
 212     public Future<RpcResult<Void>> setAsync(final SetAsyncInput input) {
 213         return sendToSwitchFuture(input, "set-async-input sending failed");
 214     }
 215
 216     @Override
 217     public boolean isAlive() {
 218         return channel.isOpen();
 219     }
 220
 221     @Override
 222     public boolean isAutoRead() {
 223         return channel.config().isAutoRead();
 224     }
 225
 226     @Override
 227     public void setAutoRead(final boolean autoRead) {
 228         channel.config().setAutoRead(autoRead);
 229     }
 230
 231     @Override
 232     public InetSocketAddress getRemoteAddress() {
 233         return (InetSocketAddress) channel.remoteAddress();
 234     }
 235
 236     /**
 237      * Used only for testing purposes
 238      * @param cache replacement
 239      */
 240     @VisibleForTesting
 241     void setResponseCache(final Cache<RpcResponseKey, ResponseExpectedRpcListener<?>> cache) {
 242         this.responseCache = cache;
 243     }
 244
 245     /**
 246      * Return cached RpcListener or {@code null} if not cached
 247      * @return
 248      */
 249     protected ResponseExpectedRpcListener<?> findRpcResponse(final RpcResponseKey key) {
 250         return responseCache.getIfPresent(key);
 251     }
 252
 253     /**
 254      * sends given message to switch, sending result or switch response will be reported via return value
 255      *
 256      * @param input message to send
 257      * @param responseClazz type of response
 258      * @param failureInfo describes, what type of message caused failure by sending
 259      * @return future object,
 260      *         <ul>
 261      *         <li>if send fails, {@link RpcResult} will contain errors and failed status</li>
 262      *         <li>else {@link RpcResult} will be stored in responseCache and wait for particular timeout (
 263      *         {@link ConnectionAdapterImpl#RPC_RESPONSE_EXPIRATION}),
 264      *         <ul>
 265      *         <li>either switch will manage to answer and then corresponding response message will be set into returned
 266      *         future</li>
 267      *         <li>or response in cache will expire and returned future will be cancelled</li>
 268      *         </ul>
 269      *         </li>
 270      *         </ul>
 271      */
 272     protected <IN extends OfHeader, OUT extends OfHeader> ListenableFuture<RpcResult<OUT>> sendToSwitchExpectRpcResultFuture(
 273             final IN input, final Class<OUT> responseClazz, final String failureInfo) {
 274         final RpcResponseKey key = new RpcResponseKey(input.getXid(), responseClazz.getName());
 275         final ResponseExpectedRpcListener<OUT> listener = new ResponseExpectedRpcListener<>(input, failureInfo,
 276                 responseCache, key);
 277         return enqueueMessage(listener);
 278     }
 279
 280     /**
 281      * sends given message to switch, sending result will be reported via return value
 282      *
 283      * @param input message to send
 284      * @param failureInfo describes, what type of message caused failure by sending
 285      * @return future object,
 286      *         <ul>
 287      *         <li>if send successful, {@link RpcResult} without errors and successful status will be returned,</li>
 288      *         <li>else {@link RpcResult} will contain errors and failed status</li>
 289      *         </ul>
 290      */
 291     protected ListenableFuture<RpcResult<Void>> sendToSwitchFuture(final DataObject input, final String failureInfo) {
 292         return enqueueMessage(new SimpleRpcListener(input, failureInfo));
 293     }
 294
 295     private <T> ListenableFuture<RpcResult<T>> enqueueMessage(final AbstractRpcListener<T> promise) {
 296         LOG.debug("Submitting promise {}", promise);
 297
 298         if (!output.enqueue(promise)) {
 299             LOG.debug("Message queue is full, rejecting execution");
 300             promise.failedRpc(QUEUE_FULL_EXCEPTION);
 301         } else {
 302             LOG.debug("Promise enqueued successfully");
 303         }
 304
 305         return promise.getResult();
 306     }
 307
 308     /**
 309      * @param resultFuture
 310      * @param failureInfo
 311      * @param errorSeverity
 312      * @param message
 313      * @return
 314      */
 315     private static SettableFuture<Boolean> handleTransportChannelFuture(final ChannelFuture resultFuture) {
 316
 317         final SettableFuture<Boolean> transportResult = SettableFuture.create();
 318
 319         resultFuture.addListener(new GenericFutureListener<io.netty.util.concurrent.Future<? super Void>>() {
 320
 321             @Override
 322             public void operationComplete(final io.netty.util.concurrent.Future<? super Void> future) throws Exception {
 323                 transportResult.set(future.isSuccess());
 324                 if (!future.isSuccess()) {
 325                     transportResult.setException(future.cause());
 326                 }
 327             }
 328         });
 329         return transportResult;
 330     }
 331 }