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 = SettableFuture.create();
  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         this.readyRunnable = readyRunnable;
  64     }
  65
  66     @Override
  67     @SuppressWarnings("checkstyle:IllegalCatch")
  68     public void run() {
  69         final ChannelFuture f;
  70         try {
  71             Bootstrap bootstrap = new Bootstrap();
  72             bootstrap.group(group).channel(datagramChannelClass).option(ChannelOption.SO_BROADCAST, false)
  73                 .handler(channelInitializer);
  74
  75             if (startupAddress != null) {
  76                 f = bootstrap.bind(startupAddress.getHostAddress(), port).sync();
  77             } else {
  78                 f = bootstrap.bind(port).sync();
  79             }
  80         } catch (InterruptedException e) {
  81             LOG.error("Interrupted while binding port {}", port, e);
  82             return;
  83         } catch (Throwable throwable) {
  84             // sync() re-throws exceptions declared as Throwable, so the compiler doesn't see them
  85             LOG.error("Error while binding address {} and port {}", startupAddress, port, throwable);
  86             throw throwable;
  87         }
  88
  89         try {
  90             InetSocketAddress isa = (InetSocketAddress) f.channel().localAddress();
  91             String address = isa.getHostString();
  92
  93             // Update port, as it may have been specified as 0
  94             this.port = isa.getPort();
  95
  96             LOG.debug("Address from udpHandler: {}", address);
  97             LOG.info("Switch listener started and ready to accept incoming udp connections on port: {}", port);
  98             readyRunnable.run();
  99             isOnlineFuture.set(true);
 100
 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         if (isEpollEnabled) {
 147             initiateEpollEventLoopGroups(threadConfiguration);
 148         } else {
 149             initiateNioEventLoopGroups(threadConfiguration);
 150         }
 151     }
 152
 153     /**
 154      * Initiate Nio event loop groups.
 155      *
 156      * @param threadConfiguration number of threads to be created, if not specified in threadConfig
 157      */
 158     public void initiateNioEventLoopGroups(ThreadConfiguration threadConfiguration) {
 159         datagramChannelClass = NioDatagramChannel.class;
 160         if (threadConfiguration != null) {
 161             group = new NioEventLoopGroup(threadConfiguration.getWorkerThreadCount());
 162         } else {
 163             group = new NioEventLoopGroup();
 164         }
 165     }
 166
 167     /**
 168      * Initiate Epoll event loop groups with Nio as fall back.
 169      *
 170      * @param threadConfiguration the ThreadConfiguration
 171      */
 172     @SuppressWarnings("checkstyle:IllegalCatch")
 173     protected void initiateEpollEventLoopGroups(ThreadConfiguration threadConfiguration) {
 174         try {
 175             datagramChannelClass = EpollDatagramChannel.class;
 176             if (threadConfiguration != null) {
 177                 group = new EpollEventLoopGroup(threadConfiguration.getWorkerThreadCount());
 178             } else {
 179                 group = new EpollEventLoopGroup();
 180             }
 181             return;
 182         } catch (RuntimeException ex) {
 183             LOG.debug("Epoll initiation failed");
 184         }
 185
 186         //Fallback mechanism
 187         initiateNioEventLoopGroups(threadConfiguration);
 188     }
 189 }