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

   1 /*
   2  * Copyright (c) 2013 Pantheon Technologies s.r.o. 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 package org.opendaylight.openflowjava.protocol.impl.core;
   9
  10 import com.google.common.util.concurrent.FutureCallback;
  11 import com.google.common.util.concurrent.Futures;
  12 import com.google.common.util.concurrent.ListenableFuture;
  13 import com.google.common.util.concurrent.ListeningExecutorService;
  14 import com.google.common.util.concurrent.MoreExecutors;
  15 import io.netty.channel.epoll.Epoll;
  16 import java.util.Map;
  17 import org.eclipse.jdt.annotation.Nullable;
  18 import org.opendaylight.infrautils.diagstatus.DiagStatusService;
  19 import org.opendaylight.infrautils.diagstatus.ServiceDescriptor;
  20 import org.opendaylight.infrautils.diagstatus.ServiceRegistration;
  21 import org.opendaylight.infrautils.diagstatus.ServiceState;
  22 import org.opendaylight.infrautils.utils.concurrent.Executors;
  23 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionConfiguration;
  24 import org.opendaylight.openflowjava.protocol.api.connection.SwitchConnectionHandler;
  25 import org.opendaylight.openflowjava.protocol.api.extensibility.DeserializerRegistry;
  26 import org.opendaylight.openflowjava.protocol.api.extensibility.OFDeserializer;
  27 import org.opendaylight.openflowjava.protocol.api.extensibility.OFGeneralDeserializer;
  28 import org.opendaylight.openflowjava.protocol.api.extensibility.OFGeneralSerializer;
  29 import org.opendaylight.openflowjava.protocol.api.extensibility.OFSerializer;
  30 import org.opendaylight.openflowjava.protocol.api.extensibility.SerializerRegistry;
  31 import org.opendaylight.openflowjava.protocol.api.keys.ActionSerializerKey;
  32 import org.opendaylight.openflowjava.protocol.api.keys.ExperimenterActionDeserializerKey;
  33 import org.opendaylight.openflowjava.protocol.api.keys.ExperimenterDeserializerKey;
  34 import org.opendaylight.openflowjava.protocol.api.keys.ExperimenterIdDeserializerKey;
  35 import org.opendaylight.openflowjava.protocol.api.keys.ExperimenterIdMeterSubTypeSerializerKey;
  36 import org.opendaylight.openflowjava.protocol.api.keys.ExperimenterIdSerializerKey;
  37 import org.opendaylight.openflowjava.protocol.api.keys.ExperimenterInstructionDeserializerKey;
  38 import org.opendaylight.openflowjava.protocol.api.keys.ExperimenterSerializerKey;
  39 import org.opendaylight.openflowjava.protocol.api.keys.InstructionSerializerKey;
  40 import org.opendaylight.openflowjava.protocol.api.keys.MatchEntryDeserializerKey;
  41 import org.opendaylight.openflowjava.protocol.api.keys.MatchEntrySerializerKey;
  42 import org.opendaylight.openflowjava.protocol.api.keys.MessageCodeKey;
  43 import org.opendaylight.openflowjava.protocol.api.keys.MessageTypeKey;
  44 import org.opendaylight.openflowjava.protocol.api.keys.TypeToClassKey;
  45 import org.opendaylight.openflowjava.protocol.impl.deserialization.DeserializationFactory;
  46 import org.opendaylight.openflowjava.protocol.impl.deserialization.DeserializerRegistryImpl;
  47 import org.opendaylight.openflowjava.protocol.impl.serialization.SerializationFactory;
  48 import org.opendaylight.openflowjava.protocol.impl.serialization.SerializerRegistryImpl;
  49 import org.opendaylight.openflowjava.protocol.spi.connection.SwitchConnectionProvider;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.config.rev140630.TransportProtocol;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.oxm.rev150225.MatchField;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.oxm.rev150225.OxmClassBase;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ErrorMessage;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.experimenter.core.ExperimenterDataOfChoice;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.meter.band.header.meter.band.MeterBandExperimenterCase;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.queue.property.header.QueueProperty;
  57 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.table.features.properties.grouping.TableFeatureProperties;
  58 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflow._switch.connection.config.rev160506.SwitchConnectionConfig;
  59 import org.osgi.service.component.annotations.Activate;
  60 import org.osgi.service.component.annotations.Component;
  61 import org.osgi.service.component.annotations.Deactivate;
  62 import org.osgi.service.component.annotations.Reference;
  63 import org.slf4j.Logger;
  64 import org.slf4j.LoggerFactory;
  65
  66 /**
  67  * Exposed class for server handling. <br>
  68  * C - {@link MatchEntrySerializerKey} parameter representing oxm_class (see specification)<br>
  69  * F - {@link MatchEntrySerializerKey} parameter representing oxm_field (see specification)
  70  * @author mirehak
  71  * @author michal.polkorab
  72  */
  73 @Component(service = SwitchConnectionProvider.class, factory = SwitchConnectionProviderImpl.FACTORY_NAME)
  74 public class SwitchConnectionProviderImpl implements SwitchConnectionProvider, ConnectionInitializer, AutoCloseable {
  75     private static final Logger LOG = LoggerFactory.getLogger(SwitchConnectionProviderImpl.class);
  76
  77     private static final String THREAD_NAME_PREFIX = "OFP-SwitchConnectionProvider-Udp/TcpHandler";
  78     private static final String OPENFLOW_JAVA_SERVICE_NAME_PREFIX = "OPENFLOW_SERVER";
  79
  80     // OSGi DS Component Factory name
  81     public static final String FACTORY_NAME =
  82         "org.opendaylight.openflowjava.protocol.impl.core.SwitchConnectionProviderImpl";
  83     public static final String PROP_CONFIG = ".config";
  84
  85     private final ConnectionConfiguration connConfig;
  86     private final SerializationFactory serializationFactory;
  87     private final SerializerRegistry serializerRegistry;
  88     private final DeserializerRegistry deserializerRegistry;
  89     private final DeserializationFactory deserializationFactory;
  90     private final ListeningExecutorService listeningExecutorService;
  91     private final String diagStatusIdentifier;
  92     private final String threadName;
  93
  94     private TcpConnectionInitializer connectionInitializer;
  95     private ServerFacade serverFacade;
  96     // FIXME: clean this up when no longer needed
  97     private final ServiceRegistration diagReg;
  98
  99     public SwitchConnectionProviderImpl(final DiagStatusService diagStatus,
 100             final @Nullable ConnectionConfiguration connConfig) {
 101         this.connConfig = connConfig;
 102         String connectionSuffix = createConnectionSuffix(connConfig);
 103         diagStatusIdentifier = OPENFLOW_JAVA_SERVICE_NAME_PREFIX + connectionSuffix;
 104         diagReg = diagStatus.register(diagStatusIdentifier);
 105
 106         threadName = THREAD_NAME_PREFIX + connectionSuffix;
 107         listeningExecutorService = Executors.newListeningSingleThreadExecutor(threadName, LOG);
 108         serializerRegistry = new SerializerRegistryImpl();
 109         if (connConfig != null) {
 110             serializerRegistry.setGroupAddModConfig(connConfig.isGroupAddModEnabled());
 111         }
 112         serializerRegistry.init();
 113         serializationFactory = new SerializationFactory(serializerRegistry);
 114         deserializerRegistry = new DeserializerRegistryImpl();
 115         deserializerRegistry.init();
 116         deserializationFactory = new DeserializationFactory(deserializerRegistry);
 117     }
 118
 119     @Activate
 120     public SwitchConnectionProviderImpl(@Reference final DiagStatusService diagStatus,
 121             final Map<String, Object> props) {
 122         this(diagStatus, new ConnectionConfigurationImpl((SwitchConnectionConfig) props.get(PROP_CONFIG)));
 123     }
 124
 125     @Override
 126     @Deactivate
 127     public void close() {
 128         shutdown();
 129         diagReg.close();
 130     }
 131
 132     // ID based, on configuration, used for diagstatus serviceIdentifier (ServiceDescriptor moduleServiceName)
 133     private static String createConnectionSuffix(final @Nullable ConnectionConfiguration config) {
 134         return config == null ? "-null-config" : "_" + config.getPort();
 135     }
 136
 137     @Override
 138     public ListenableFuture<Void> shutdown() {
 139         LOG.debug("Shutdown summoned");
 140         if (serverFacade == null) {
 141             LOG.warn("Can not shutdown - not configured or started");
 142             throw new IllegalStateException("SwitchConnectionProvider is not started or not configured.");
 143         }
 144         final var serverFacadeShutdownFuture = serverFacade.shutdown();
 145         Executors.shutdownAndAwaitTermination(listeningExecutorService);
 146         return serverFacadeShutdownFuture;
 147     }
 148
 149     @Override
 150     @SuppressWarnings("checkstyle:IllegalCatch")
 151     public ListenableFuture<Void> startup(final SwitchConnectionHandler connectionHandler) {
 152         LOG.debug("Startup summoned");
 153         if (connConfig == null) {
 154             return Futures.immediateFailedFuture(new IllegalStateException("Connection not configured"));
 155         }
 156         if (connectionHandler == null) {
 157             return Futures.immediateFailedFuture(new IllegalStateException("SwitchConnectionHandler is not set"));
 158         }
 159
 160         try {
 161             serverFacade = createAndConfigureServer(connectionHandler);
 162             Futures.addCallback(listeningExecutorService.submit(serverFacade), new FutureCallback<Object>() {
 163                 @Override
 164                 public void onFailure(final Throwable throwable) {
 165                     diagReg.report(new ServiceDescriptor(diagStatusIdentifier, throwable));
 166                 }
 167
 168                 @Override
 169                 public void onSuccess(final Object result) {
 170                     diagReg.report(new ServiceDescriptor(diagStatusIdentifier, ServiceState.ERROR,
 171                         threadName + " terminated"));
 172                 }
 173             }, MoreExecutors.directExecutor());
 174             return serverFacade.getIsOnlineFuture();
 175         } catch (RuntimeException e) {
 176             return Futures.immediateFailedFuture(e);
 177         }
 178     }
 179
 180     private ServerFacade createAndConfigureServer(final SwitchConnectionHandler connectionHandler) {
 181         LOG.debug("Configuring ..");
 182
 183         final var transportProtocol = (TransportProtocol) connConfig.getTransferProtocol();
 184         if (transportProtocol == null) {
 185             throw new IllegalStateException("No transport protocol received in " + connConfig);
 186         }
 187
 188         final var factory = new ChannelInitializerFactory();
 189         factory.setSwitchConnectionHandler(connectionHandler);
 190         factory.setSwitchIdleTimeout(connConfig.getSwitchIdleTimeout());
 191         factory.setTlsConfig(connConfig.getTlsConfiguration());
 192         factory.setSerializationFactory(serializationFactory);
 193         factory.setDeserializationFactory(deserializationFactory);
 194         factory.setUseBarrier(connConfig.useBarrier());
 195         factory.setChannelOutboundQueueSize(connConfig.getChannelOutboundQueueSize());
 196         // Check if Epoll native transport is available.
 197         // TODO : Add option to disable Epoll.
 198         boolean isEpollEnabled = Epoll.isAvailable();
 199
 200         return switch (transportProtocol) {
 201             case TCP, TLS -> {
 202                 final var tcpHandler = new TcpHandler(connConfig.getAddress(), connConfig.getPort(),
 203                     () -> diagReg.report(new ServiceDescriptor(diagStatusIdentifier, ServiceState.OPERATIONAL)));
 204                 final var channelInitializer = factory.createPublishingChannelInitializer();
 205                 tcpHandler.setChannelInitializer(channelInitializer);
 206                 tcpHandler.initiateEventLoopGroups(connConfig.getThreadConfiguration(), isEpollEnabled);
 207                 final var workerGroupFromTcpHandler = tcpHandler.getWorkerGroup();
 208                 connectionInitializer = new TcpConnectionInitializer(workerGroupFromTcpHandler, channelInitializer,
 209                     isEpollEnabled);
 210                 yield tcpHandler;
 211             }
 212             case UDP -> {
 213                 final var udpHandler = new UdpHandler(connConfig.getAddress(), connConfig.getPort(),
 214                     () -> diagReg.report(new ServiceDescriptor(diagStatusIdentifier, ServiceState.OPERATIONAL)));
 215                 udpHandler.initiateEventLoopGroups(connConfig.getThreadConfiguration(), isEpollEnabled);
 216                 udpHandler.setChannelInitializer(factory.createUdpChannelInitializer());
 217                 yield udpHandler;
 218             }
 219         };
 220     }
 221
 222     public ServerFacade getServerFacade() {
 223         return serverFacade;
 224     }
 225
 226     @Override
 227     public boolean unregisterSerializer(final ExperimenterSerializerKey key) {
 228         return serializerRegistry.unregisterSerializer((MessageTypeKey<?>) key);
 229     }
 230
 231     @Override
 232     public boolean unregisterDeserializer(final ExperimenterDeserializerKey key) {
 233         return deserializerRegistry.unregisterDeserializer((MessageCodeKey) key);
 234     }
 235
 236     @Override
 237     public void registerActionSerializer(final ActionSerializerKey<?> key,
 238             final OFGeneralSerializer serializer) {
 239         serializerRegistry.registerSerializer(key, serializer);
 240     }
 241
 242     @Override
 243     public void registerActionDeserializer(final ExperimenterActionDeserializerKey key,
 244             final OFGeneralDeserializer deserializer) {
 245         deserializerRegistry.registerDeserializer(key, deserializer);
 246     }
 247
 248     @Override
 249     public void registerInstructionSerializer(final InstructionSerializerKey<?> key,
 250             final OFGeneralSerializer serializer) {
 251         serializerRegistry.registerSerializer(key, serializer);
 252     }
 253
 254     @Override
 255     public void registerInstructionDeserializer(final ExperimenterInstructionDeserializerKey key,
 256             final OFGeneralDeserializer deserializer) {
 257         deserializerRegistry.registerDeserializer(key, deserializer);
 258     }
 259
 260     @Override
 261     public <C extends OxmClassBase, F extends MatchField> void registerMatchEntrySerializer(
 262             final MatchEntrySerializerKey<C, F> key, final OFGeneralSerializer serializer) {
 263         serializerRegistry.registerSerializer(key, serializer);
 264     }
 265
 266     @Override
 267     public void registerMatchEntryDeserializer(final MatchEntryDeserializerKey key,
 268             final OFGeneralDeserializer deserializer) {
 269         deserializerRegistry.registerDeserializer(key, deserializer);
 270     }
 271
 272     @Override
 273     public void registerErrorDeserializer(final ExperimenterIdDeserializerKey key,
 274             final OFDeserializer<ErrorMessage> deserializer) {
 275         deserializerRegistry.registerDeserializer(key, deserializer);
 276     }
 277
 278     @Override
 279     public void registerExperimenterMessageDeserializer(final ExperimenterIdDeserializerKey key,
 280             final OFDeserializer<? extends ExperimenterDataOfChoice> deserializer) {
 281         deserializerRegistry.registerDeserializer(key, deserializer);
 282     }
 283
 284     @Override
 285     public void registerMultipartReplyMessageDeserializer(final ExperimenterIdDeserializerKey key,
 286             final OFDeserializer<? extends ExperimenterDataOfChoice> deserializer) {
 287         deserializerRegistry.registerDeserializer(key, deserializer);
 288     }
 289
 290     @Override
 291     public void registerMultipartReplyTFDeserializer(final ExperimenterIdDeserializerKey key,
 292             final OFGeneralDeserializer deserializer) {
 293         deserializerRegistry.registerDeserializer(key, deserializer);
 294     }
 295
 296     @Override
 297     public void registerQueuePropertyDeserializer(final ExperimenterIdDeserializerKey key,
 298             final OFDeserializer<QueueProperty> deserializer) {
 299         deserializerRegistry.registerDeserializer(key, deserializer);
 300     }
 301
 302     @Override
 303     public void registerMeterBandDeserializer(final ExperimenterIdDeserializerKey key,
 304             final OFDeserializer<MeterBandExperimenterCase> deserializer) {
 305         deserializerRegistry.registerDeserializer(key, deserializer);
 306     }
 307
 308     @Override
 309     public void registerExperimenterMessageSerializer(
 310             final ExperimenterIdSerializerKey<? extends ExperimenterDataOfChoice> key,
 311             final OFSerializer<? extends ExperimenterDataOfChoice> serializer) {
 312         serializerRegistry.registerSerializer(key, serializer);
 313     }
 314
 315     @Override
 316     public void registerMultipartRequestSerializer(
 317             final ExperimenterIdSerializerKey<? extends ExperimenterDataOfChoice> key,
 318             final OFSerializer<? extends ExperimenterDataOfChoice> serializer) {
 319         serializerRegistry.registerSerializer(key, serializer);
 320     }
 321
 322     @Override
 323     public void registerMultipartRequestTFSerializer(final ExperimenterIdSerializerKey<TableFeatureProperties> key,
 324             final OFGeneralSerializer serializer) {
 325         serializerRegistry.registerSerializer(key, serializer);
 326     }
 327
 328     /**
 329      * Deprecated.
 330      *
 331      * @deprecated Since we have used ExperimenterIdMeterSubTypeSerializerKey as MeterBandSerializer's key, in order
 332      *     to avoid the occurrence of an error, we should discard this function.
 333      */
 334     @Override
 335     @Deprecated
 336     public void registerMeterBandSerializer(final ExperimenterIdSerializerKey<MeterBandExperimenterCase> key,
 337             final OFSerializer<MeterBandExperimenterCase> serializer) {
 338         serializerRegistry.registerSerializer(key, serializer);
 339     }
 340
 341     @Override
 342     public void registerMeterBandSerializer(
 343             final ExperimenterIdMeterSubTypeSerializerKey<MeterBandExperimenterCase> key,
 344             final OFSerializer<MeterBandExperimenterCase> serializer) {
 345         serializerRegistry.registerSerializer(key, serializer);
 346     }
 347
 348     @Override
 349     public void initiateConnection(final String host, final int port) {
 350         connectionInitializer.initiateConnection(host, port);
 351     }
 352
 353     @Override
 354     public ConnectionConfiguration getConfiguration() {
 355         return connConfig;
 356     }
 357
 358     @Override
 359     public <K> void registerSerializer(final MessageTypeKey<K> key, final OFGeneralSerializer serializer) {
 360         serializerRegistry.registerSerializer(key, serializer);
 361     }
 362
 363     @Override
 364     public void registerDeserializer(final MessageCodeKey key, final OFGeneralDeserializer deserializer) {
 365         deserializerRegistry.registerDeserializer(key, deserializer);
 366     }
 367
 368     @Override
 369     public void registerDeserializerMapping(final TypeToClassKey key, final Class<?> clazz) {
 370         deserializationFactory.registerMapping(key, clazz);
 371     }
 372
 373     @Override
 374     public boolean unregisterDeserializerMapping(final TypeToClassKey key) {
 375         return deserializationFactory.unregisterMapping(key);
 376     }
 377 }