openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/TcpHandler.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
   9 package org.opendaylight.openflowjava.protocol.impl.core;
  10
  11 import io.netty.bootstrap.ServerBootstrap;
  12 import io.netty.channel.ChannelFuture;
  13 import io.netty.channel.ChannelOption;
  14 import io.netty.channel.nio.NioEventLoopGroup;
  15 import io.netty.channel.socket.nio.NioServerSocketChannel;
  16 import io.netty.handler.logging.LogLevel;
  17 import io.netty.handler.logging.LoggingHandler;
  18 import io.netty.util.concurrent.GenericFutureListener;
  19
  20 import java.net.InetAddress;
  21 import java.net.InetSocketAddress;
  22
  23 import org.opendaylight.openflowjava.protocol.api.connection.SwitchConnectionHandler;
  24 import org.opendaylight.openflowjava.protocol.impl.connection.ServerFacade;
  25 import org.opendaylight.openflowjava.protocol.impl.deserialization.DeserializationFactory;
  26 import org.opendaylight.openflowjava.protocol.impl.serialization.SerializationFactory;
  27 import org.slf4j.Logger;
  28 import org.slf4j.LoggerFactory;
  29
  30 import com.google.common.util.concurrent.ListenableFuture;
  31 import com.google.common.util.concurrent.SettableFuture;
  32
  33 /**
  34  * Class implementing server over TCP for handling incoming connections.
  35  *
  36  * @author michal.polkorab
  37  */
  38 public class TcpHandler implements ServerFacade {
  39     /*
  40      * High/low write watermarks, in KiB.
  41      */
  42     private static final int DEFAULT_WRITE_HIGH_WATERMARK = 64;
  43     private static final int DEFAULT_WRITE_LOW_WATERMARK = 32;
  44     /*
  45      * Write spin count. This tells netty to immediately retry a non-blocking
  46      * write this many times before moving on to selecting.
  47      */
  48     private static final int DEFAULT_WRITE_SPIN_COUNT = 16;
  49
  50     private static final Logger LOGGER = LoggerFactory.getLogger(TcpHandler.class);
  51
  52     private int port;
  53     private String address;
  54     private final InetAddress startupAddress;
  55     private NioEventLoopGroup workerGroup;
  56     private NioEventLoopGroup bossGroup;
  57     private final SettableFuture<Boolean> isOnlineFuture;
  58
  59     private final PublishingChannelInitializer channelInitializer;
  60
  61     /**
  62      * Enum used for storing names of used components (in pipeline).
  63      */
  64     public static enum COMPONENT_NAMES {
  65
  66         /**
  67          * Detects switch idle state
  68          */
  69         IDLE_HANDLER,
  70         /**
  71          * Detects TLS connections
  72          */
  73         TLS_DETECTOR,
  74         /**
  75          * Component for handling TLS frames
  76          */
  77         SSL_HANDLER,
  78         /**
  79          * Decodes incoming messages into message frames
  80          */
  81         OF_FRAME_DECODER,
  82         /**
  83          * Detects version of incoming OpenFlow Protocol message
  84          */
  85         OF_VERSION_DETECTOR,
  86         /**
  87          * Transforms OpenFlow Protocol byte messages into POJOs
  88          */
  89         OF_DECODER,
  90         /**
  91          * Transforms POJOs into OpenFlow Protocol byte messages
  92          */
  93         OF_ENCODER,
  94         /**
  95          * Delegates translated POJOs into MessageConsumer
  96          */
  97         DELEGATING_INBOUND_HANDLER,
  98     }
  99
 100
 101     /**
 102      * Constructor of TCPHandler that listens on selected port.
 103      *
 104      * @param port listening port of TCPHandler server
 105      */
 106     public TcpHandler(final int port) {
 107         this(null, port);
 108     }
 109
 110     /**
 111      * Constructor of TCPHandler that listens on selected address and port.
 112      * @param address listening address of TCPHandler server
 113      * @param port listening port of TCPHandler server
 114      */
 115     public TcpHandler(final InetAddress address, final int port) {
 116         this.port = port;
 117         this.startupAddress = address;
 118         channelInitializer = new PublishingChannelInitializer();
 119         isOnlineFuture = SettableFuture.create();
 120     }
 121
 122     /**
 123      * Starts server on selected port.
 124      */
 125     @Override
 126     public void run() {
 127         bossGroup = new NioEventLoopGroup();
 128         workerGroup = new NioEventLoopGroup();
 129
 130         /*
 131          * We generally do not perform IO-unrelated tasks, so we want to have
 132          * all outstanding tasks completed before the executing thread goes
 133          * back into select.
 134          *
 135          * Any other setting means netty will measure the time it spent selecting
 136          * and spend roughly proportional time executing tasks.
 137          */
 138         workerGroup.setIoRatio(100);
 139
 140         final ChannelFuture f;
 141         try {
 142             ServerBootstrap b = new ServerBootstrap();
 143             b.group(bossGroup, workerGroup)
 144                     .channel(NioServerSocketChannel.class)
 145                     .handler(new LoggingHandler(LogLevel.DEBUG))
 146                     .childHandler(channelInitializer)
 147                     .option(ChannelOption.SO_BACKLOG, 128)
 148                     .option(ChannelOption.SO_REUSEADDR, true)
 149                     .childOption(ChannelOption.SO_KEEPALIVE, true)
 150                     .childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, DEFAULT_WRITE_HIGH_WATERMARK * 1024)
 151                     .childOption(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, DEFAULT_WRITE_LOW_WATERMARK * 1024)
 152                     .childOption(ChannelOption.WRITE_SPIN_COUNT, DEFAULT_WRITE_SPIN_COUNT);
 153
 154             if (startupAddress != null) {
 155                 f = b.bind(startupAddress.getHostAddress(), port).sync();
 156             } else {
 157                 f = b.bind(port).sync();
 158             }
 159         } catch (InterruptedException e) {
 160             LOGGER.error("Interrupted while binding port {}", port, e);
 161             return;
 162         }
 163
 164         try {
 165             InetSocketAddress isa = (InetSocketAddress) f.channel().localAddress();
 166             address = isa.getHostString();
 167
 168             // Update port, as it may have been specified as 0
 169             this.port = isa.getPort();
 170
 171             LOGGER.debug("address from tcphandler: {}", address);
 172             isOnlineFuture.set(true);
 173             LOGGER.info("Switch listener started and ready to accept incoming connections on port: {}", port);
 174             f.channel().closeFuture().sync();
 175         } catch (InterruptedException e) {
 176             LOGGER.error("Interrupted while waiting for port {} shutdown", port, e);
 177         } finally {
 178             shutdown();
 179         }
 180     }
 181
 182     /**
 183      * Shuts down {@link TcpHandler}}
 184      */
 185     @Override
 186     public ListenableFuture<Boolean> shutdown() {
 187         final SettableFuture<Boolean> result = SettableFuture.create();
 188         workerGroup.shutdownGracefully();
 189         // boss will shutdown as soon, as worker is down
 190         bossGroup.shutdownGracefully().addListener(new GenericFutureListener<io.netty.util.concurrent.Future<Object>>() {
 191
 192             @Override
 193             public void operationComplete(
 194                     final io.netty.util.concurrent.Future<Object> downResult) throws Exception {
 195                 result.set(downResult.isSuccess());
 196                 if (downResult.cause() != null) {
 197                     result.setException(downResult.cause());
 198                 }
 199             }
 200
 201         });
 202         return result;
 203     }
 204
 205     /**
 206      *
 207      * @return number of connected clients / channels
 208      */
 209     public int getNumberOfConnections() {
 210         return channelInitializer.size();
 211     }
 212
 213     /**
 214      * @return channelInitializer providing channels
 215      */
 216     public PublishingChannelInitializer getChannelInitializer() {
 217         return channelInitializer;
 218     }
 219
 220     @Override
 221     public ListenableFuture<Boolean> getIsOnlineFuture() {
 222         return isOnlineFuture;
 223     }
 224
 225     /**
 226      * @return the port
 227      */
 228     public int getPort() {
 229         return port;
 230     }
 231
 232     /**
 233      * @return the address
 234      */
 235     public String getAddress() {
 236         return address;
 237     }
 238
 239     /**
 240      * @param switchConnectionHandler
 241      */
 242     public void setSwitchConnectionHandler(
 243             final SwitchConnectionHandler switchConnectionHandler) {
 244         channelInitializer.setSwitchConnectionHandler(switchConnectionHandler);
 245     }
 246
 247     /**
 248      * @param switchIdleTimeout in milliseconds
 249      */
 250     public void setSwitchIdleTimeout(final long switchIdleTimeout) {
 251         channelInitializer.setSwitchIdleTimeout(switchIdleTimeout);
 252     }
 253
 254     /**
 255      * @param tlsSupported
 256      */
 257     public void setEncryption(final boolean tlsSupported) {
 258         channelInitializer.setEncryption(tlsSupported);
 259     }
 260
 261     /**
 262      * @param sf serialization factory
 263      */
 264     public void setSerializationFactory(final SerializationFactory sf) {
 265         channelInitializer.setSerializationFactory(sf);
 266     }
 267
 268     /**
 269      * @param factory
 270      */
 271     public void setDeserializationFactory(final DeserializationFactory factory) {
 272         channelInitializer.setDeserializationFactory(factory);
 273     }
 274
 275 }