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

   1 /*
   2  * Copyright (c) 2014 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 com.google.common.util.concurrent.ListenableFuture;
  12 import com.google.common.util.concurrent.SettableFuture;
  13 import io.netty.bootstrap.Bootstrap;
  14 import io.netty.channel.ChannelFuture;
  15 import io.netty.channel.ChannelOption;
  16 import io.netty.channel.EventLoopGroup;
  17 import io.netty.channel.epoll.EpollDatagramChannel;
  18 import io.netty.channel.epoll.EpollEventLoopGroup;
  19 import io.netty.channel.nio.NioEventLoopGroup;
  20 import io.netty.channel.socket.DatagramChannel;
  21 import io.netty.channel.socket.nio.NioDatagramChannel;
  22 import java.net.InetAddress;
  23 import java.net.InetSocketAddress;
  24 import org.opendaylight.openflowjava.protocol.api.connection.ThreadConfiguration;
  25 import org.slf4j.Logger;
  26 import org.slf4j.LoggerFactory;
  27
  28 /**
  29  * Class implementing server over UDP for handling incoming connections.
  30  *
  31  * @author michal.polkorab
  32  */
  33 public final class UdpHandler implements ServerFacade {
  34
  35     private static final Logger LOG = LoggerFactory.getLogger(UdpHandler.class);
  36
  37     private int port;
  38     private EventLoopGroup group;
  39     private final InetAddress startupAddress;
  40     private final Runnable readyRunnable;
  41     private final SettableFuture<Boolean> isOnlineFuture;
  42     private UdpChannelInitializer channelInitializer;
  43     private ThreadConfiguration threadConfig;
  44     private Class<? extends DatagramChannel> datagramChannelClass;
  45
  46     /**
  47      * Constructor of UdpHandler that listens on selected port.
  48      *
  49      * @param port listening port of UdpHandler server
  50      */
  51     public UdpHandler(final int port, Runnable readyRunnable) {
  52         this(null, port, readyRunnable);
  53     }
  54
  55     /**
  56      * Constructor of UdpHandler that listens on selected address and port.
  57      * @param address listening address of UdpHandler server
  58      * @param port listening port of UdpHandler server
  59      */
  60     public UdpHandler(final InetAddress address, final int port, Runnable readyRunnable) {
  61         this.port = port;
  62         this.startupAddress = address;
  63         isOnlineFuture = SettableFuture.create();
  64         this.readyRunnable = readyRunnable;
  65     }
  66
  67     @Override
  68     @SuppressWarnings("checkstyle:IllegalCatch")
  69     public void run() {
  70         final ChannelFuture f;
  71         try {
  72             Bootstrap bootstrap = new Bootstrap();
  73             bootstrap.group(group).channel(datagramChannelClass).option(ChannelOption.SO_BROADCAST, false)
  74                 .handler(channelInitializer);
  75
  76             if (startupAddress != null) {
  77                 f = bootstrap.bind(startupAddress.getHostAddress(), port).sync();
  78             } else {
  79                 f = bootstrap.bind(port).sync();
  80             }
  81         } catch (InterruptedException e) {
  82             LOG.error("Interrupted while binding port {}", port, e);
  83             return;
  84         } catch (Throwable throwable) {
  85             // sync() re-throws exceptions declared as Throwable, so the compiler doesn't see them
  86             LOG.error("Error while binding address {} and port {}", startupAddress, port, throwable);
  87             throw throwable;
  88         }
  89
  90         try {
  91             InetSocketAddress isa = (InetSocketAddress) f.channel().localAddress();
  92             String address = isa.getHostString();
  93
  94             // Update port, as it may have been specified as 0
  95             this.port = isa.getPort();
  96
  97             LOG.debug("Address from udpHandler: {}", address);
  98             isOnlineFuture.set(true);
  99             LOG.info("Switch listener started and ready to accept incoming udp connections on port: {}", port);
 100             readyRunnable.run();
 101             // This waits until this channel is closed, and rethrows the cause of the failure if this future failed.
 102             f.channel().closeFuture().sync();
 103         } catch (InterruptedException e) {
 104             LOG.error("Interrupted while waiting for port {} shutdown", port, e);
 105         } finally {
 106             shutdown();
 107         }
 108     }
 109
 110     @Override
 111     public ListenableFuture<Boolean> shutdown() {
 112         final SettableFuture<Boolean> result = SettableFuture.create();
 113         group.shutdownGracefully().addListener(downResult -> {
 114             result.set(downResult.isSuccess());
 115             if (downResult.cause() != null) {
 116                 result.setException(downResult.cause());
 117             }
 118         });
 119         return result;
 120     }
 121
 122     @Override
 123     public ListenableFuture<Boolean> getIsOnlineFuture() {
 124         return isOnlineFuture;
 125     }
 126
 127     public int getPort() {
 128         return port;
 129     }
 130
 131     public void setChannelInitializer(UdpChannelInitializer channelInitializer) {
 132         this.channelInitializer = channelInitializer;
 133     }
 134
 135     @Override
 136     public void setThreadConfig(ThreadConfiguration threadConfig) {
 137         this.threadConfig = threadConfig;
 138     }
 139
 140     /**
 141      * Initiate event loop groups.
 142      *
 143      * @param threadConfiguration number of threads to be created, if not specified in threadConfig
 144      */
 145     public void initiateEventLoopGroups(ThreadConfiguration threadConfiguration, boolean isEpollEnabled) {
 146
 147         if (isEpollEnabled) {
 148             initiateEpollEventLoopGroups(threadConfiguration);
 149         } else {
 150             initiateNioEventLoopGroups(threadConfiguration);
 151         }
 152     }
 153
 154     /**
 155      * Initiate Nio event loop groups.
 156      *
 157      * @param threadConfiguration number of threads to be created, if not specified in threadConfig
 158      */
 159     public void initiateNioEventLoopGroups(ThreadConfiguration threadConfiguration) {
 160         datagramChannelClass = NioDatagramChannel.class;
 161         if (threadConfiguration != null) {
 162             group = new NioEventLoopGroup(threadConfiguration.getWorkerThreadCount());
 163         } else {
 164             group = new NioEventLoopGroup();
 165         }
 166     }
 167
 168     /**
 169      * Initiate Epoll event loop groups with Nio as fall back.
 170      *
 171      * @param threadConfiguration the ThreadConfiguration
 172      */
 173     @SuppressWarnings("checkstyle:IllegalCatch")
 174     protected void initiateEpollEventLoopGroups(ThreadConfiguration threadConfiguration) {
 175         try {
 176             datagramChannelClass = EpollDatagramChannel.class;
 177             if (threadConfiguration != null) {
 178                 group = new EpollEventLoopGroup(threadConfiguration.getWorkerThreadCount());
 179             } else {
 180                 group = new EpollEventLoopGroup();
 181             }
 182             return;
 183         } catch (RuntimeException ex) {
 184             LOG.debug("Epoll initiation failed");
 185         }
 186
 187         //Fallback mechanism
 188         initiateNioEventLoopGroups(threadConfiguration);
 189     }
 190 }