netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/AbstractNetconfDispatcher.java

   1 /*
   2  * Copyright (c) 2013 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 package org.opendaylight.netconf.nettyutil;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import io.netty.bootstrap.Bootstrap;
  13 import io.netty.bootstrap.ServerBootstrap;
  14 import io.netty.buffer.PooledByteBufAllocator;
  15 import io.netty.channel.Channel;
  16 import io.netty.channel.ChannelFuture;
  17 import io.netty.channel.ChannelInitializer;
  18 import io.netty.channel.ChannelOption;
  19 import io.netty.channel.EventLoopGroup;
  20 import io.netty.channel.ServerChannel;
  21 import io.netty.channel.local.LocalServerChannel;
  22 import io.netty.channel.socket.SocketChannel;
  23 import io.netty.channel.socket.nio.NioServerSocketChannel;
  24 import io.netty.channel.socket.nio.NioSocketChannel;
  25 import io.netty.util.concurrent.DefaultPromise;
  26 import io.netty.util.concurrent.EventExecutor;
  27 import io.netty.util.concurrent.Future;
  28 import io.netty.util.concurrent.GlobalEventExecutor;
  29 import io.netty.util.concurrent.Promise;
  30 import java.io.Closeable;
  31 import java.net.InetSocketAddress;
  32 import java.net.SocketAddress;
  33 import org.opendaylight.netconf.api.NetconfSession;
  34 import org.opendaylight.netconf.api.NetconfSessionListener;
  35 import org.slf4j.Logger;
  36 import org.slf4j.LoggerFactory;
  37
  38 /**
  39  * Dispatcher class for creating servers and clients. The idea is to first create servers and clients and the run the
  40  * start method that will handle sockets in different thread.
  41  */
  42 @Deprecated
  43 public abstract class AbstractNetconfDispatcher<S extends NetconfSession, L extends NetconfSessionListener<? super S>>
  44         implements Closeable {
  45
  46     protected interface ChannelPipelineInitializer<C extends Channel, S extends NetconfSession> {
  47         /**
  48          * Initializes channel by specifying the handlers in its pipeline. Handlers are protocol specific, therefore
  49          * this method needs to be implemented in protocol specific Dispatchers.
  50          *
  51          * @param channel whose pipeline should be defined, also to be passed to {@link NetconfSessionNegotiatorFactory}
  52          * @param promise to be passed to {@link NetconfSessionNegotiatorFactory}
  53          */
  54         void initializeChannel(C channel, Promise<S> promise);
  55     }
  56
  57     protected interface PipelineInitializer<S extends NetconfSession>
  58         extends ChannelPipelineInitializer<SocketChannel, S> {
  59
  60     }
  61
  62
  63     private static final Logger LOG = LoggerFactory.getLogger(AbstractNetconfDispatcher.class);
  64
  65     private final EventLoopGroup bossGroup;
  66
  67     private final EventLoopGroup workerGroup;
  68
  69     private final EventExecutor executor;
  70
  71     protected AbstractNetconfDispatcher(final EventLoopGroup bossGroup, final EventLoopGroup workerGroup) {
  72         this(GlobalEventExecutor.INSTANCE, bossGroup, workerGroup);
  73     }
  74
  75     protected AbstractNetconfDispatcher(final EventExecutor executor, final EventLoopGroup bossGroup,
  76             final EventLoopGroup workerGroup) {
  77         this.bossGroup = requireNonNull(bossGroup);
  78         this.workerGroup = requireNonNull(workerGroup);
  79         this.executor = requireNonNull(executor);
  80     }
  81
  82
  83     /**
  84      * Creates server. Each server needs factories to pass their instances to client sessions.
  85      *
  86      * @param address address to which the server should be bound
  87      * @param initializer instance of PipelineInitializer used to initialize the channel pipeline
  88      *
  89      * @return ChannelFuture representing the binding process
  90      */
  91     protected ChannelFuture createServer(final InetSocketAddress address, final PipelineInitializer<S> initializer) {
  92         return createServer(address, NioServerSocketChannel.class, initializer);
  93     }
  94
  95     /**
  96      * Creates server. Each server needs factories to pass their instances to client sessions.
  97      *
  98      * @param address address to which the server should be bound
  99      * @param channelClass The {@link Class} which is used to create {@link Channel} instances from.
 100      * @param initializer instance of PipelineInitializer used to initialize the channel pipeline
 101      *
 102      * @return ChannelFuture representing the binding process
 103      */
 104     protected <C extends Channel> ChannelFuture createServer(final SocketAddress address,
 105             final Class<? extends ServerChannel> channelClass, final ChannelPipelineInitializer<C, S> initializer) {
 106         final ServerBootstrap b = new ServerBootstrap();
 107         b.childHandler(new ChannelInitializer<C>() {
 108
 109             @Override
 110             protected void initChannel(final C ch) {
 111                 initializer.initializeChannel(ch, new DefaultPromise<>(executor));
 112             }
 113         });
 114
 115         b.option(ChannelOption.SO_BACKLOG, 128);
 116         if (LocalServerChannel.class.equals(channelClass) == false) {
 117             // makes no sense for LocalServer and produces warning
 118             b.childOption(ChannelOption.SO_KEEPALIVE, true);
 119             b.childOption(ChannelOption.TCP_NODELAY , true);
 120         }
 121         b.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
 122         customizeBootstrap(b);
 123
 124         if (b.group() == null) {
 125             b.group(bossGroup, workerGroup);
 126         }
 127         try {
 128             b.channel(channelClass);
 129         } catch (final IllegalStateException e) {
 130             // FIXME: if this is ok, document why
 131             LOG.trace("Not overriding channelFactory on bootstrap {}", b, e);
 132         }
 133
 134         // Bind and start to accept incoming connections.
 135         final ChannelFuture f = b.bind(address);
 136         LOG.debug("Initiated server {} at {}.", f, address);
 137         return f;
 138     }
 139
 140     /**
 141      * Customize a server bootstrap before the server is created. This allows
 142      * subclasses to assign non-default server options before the server is
 143      * created.
 144      *
 145      * @param bootstrap Server bootstrap
 146      */
 147     protected void customizeBootstrap(final ServerBootstrap bootstrap) {
 148         // The default is a no-op
 149     }
 150
 151     /**
 152      * Customize a client bootstrap before the connection is attempted. This
 153      * allows subclasses to assign non-default options before the client is
 154      * created.
 155      *
 156      * @param bootstrap Client bootstrap
 157      */
 158     protected void customizeBootstrap(final Bootstrap bootstrap) {
 159         // The default is a no-op
 160     }
 161
 162     /**
 163      * Creates a client.
 164      *
 165      * @param address remote address
 166      * @param strategy Reconnection strategy to be used when initial connection fails
 167      *
 168      * @return Future representing the connection process. Its result represents the combined success of TCP connection
 169      *         as well as session negotiation.
 170      */
 171     protected Future<S> createClient(final InetSocketAddress address, final ReconnectStrategy strategy,
 172             final PipelineInitializer<S> initializer) {
 173         final Bootstrap b = new Bootstrap();
 174         final NetconfSessionPromise<S> p = new NetconfSessionPromise<>(executor, address, strategy, b);
 175         b.option(ChannelOption.SO_KEEPALIVE, true).handler(
 176                 new ChannelInitializer<SocketChannel>() {
 177                     @Override
 178                     protected void initChannel(final SocketChannel ch) {
 179                         initializer.initializeChannel(ch, p);
 180                     }
 181                 });
 182
 183         customizeBootstrap(b);
 184         setWorkerGroup(b);
 185         setChannelFactory(b);
 186
 187         p.connect();
 188         LOG.debug("Client created.");
 189         return p;
 190     }
 191
 192     /**
 193      * Create a client but use a pre-configured bootstrap.
 194      * This method however replaces the ChannelInitializer in the bootstrap. All other configuration is preserved.
 195      *
 196      * @param address remote address
 197      */
 198     protected Future<S> createClient(final InetSocketAddress address, final ReconnectStrategy strategy,
 199             final Bootstrap bootstrap, final PipelineInitializer<S> initializer) {
 200         final NetconfSessionPromise<S> p = new NetconfSessionPromise<>(executor, address, strategy, bootstrap);
 201
 202         bootstrap.handler(
 203                 new ChannelInitializer<SocketChannel>() {
 204                     @Override
 205                     protected void initChannel(final SocketChannel ch) {
 206                         initializer.initializeChannel(ch, p);
 207                     }
 208                 });
 209
 210         p.connect();
 211         LOG.debug("Client created.");
 212         return p;
 213     }
 214
 215     /**
 216      * Creates a client.
 217      *
 218      * @param address remote address
 219      * @param connectStrategyFactory Factory for creating reconnection strategy to be used when initial connection fails
 220      * @param reestablishStrategy Reconnection strategy to be used when the already-established session fails
 221      * @return Future representing the reconnection task. It will report completion based on reestablishStrategy, e.g.
 222      *         success if it indicates no further attempts should be made and failure if it reports an error
 223      * @deprecated Use
 224      *             {@link #createReconnectingClient(InetSocketAddress, ReconnectStrategyFactory, PipelineInitializer)}
 225      *             instead.
 226      */
 227     @Deprecated
 228     protected Future<Void> createReconnectingClient(final InetSocketAddress address,
 229             final ReconnectStrategyFactory connectStrategyFactory, final ReconnectStrategy reestablishStrategy,
 230             final PipelineInitializer<S> initializer) {
 231         return createReconnectingClient(address, connectStrategyFactory, initializer);
 232     }
 233
 234     /**
 235      * Creates a reconnecting client.
 236      *
 237      * @param address remote address
 238      * @param connectStrategyFactory Factory for creating reconnection strategy for every reconnect attempt
 239      * @return Future representing the reconnection task. It will report completion based on reestablishStrategy, e.g.
 240      *         success is never reported, only failure when it runs out of reconnection attempts.
 241      */
 242     protected Future<Void> createReconnectingClient(final InetSocketAddress address,
 243             final ReconnectStrategyFactory connectStrategyFactory, final PipelineInitializer<S> initializer) {
 244         final Bootstrap b = new Bootstrap();
 245
 246         final ReconnectPromise<S, L> p = new ReconnectPromise<>(GlobalEventExecutor.INSTANCE, this, address,
 247                 connectStrategyFactory, b, initializer);
 248
 249         b.option(ChannelOption.SO_KEEPALIVE, true);
 250
 251         customizeBootstrap(b);
 252         setWorkerGroup(b);
 253         setChannelFactory(b);
 254
 255         p.connect();
 256         return p;
 257     }
 258
 259     private static void setChannelFactory(final Bootstrap bootstrap) {
 260         // There is no way to detect if this was already set by
 261         // customizeBootstrap()
 262         try {
 263             bootstrap.channel(NioSocketChannel.class);
 264         } catch (final IllegalStateException e) {
 265             LOG.trace("Not overriding channelFactory on bootstrap {}", bootstrap, e);
 266         }
 267     }
 268
 269     private void setWorkerGroup(final Bootstrap bootstrap) {
 270         if (bootstrap.group() == null) {
 271             bootstrap.group(workerGroup);
 272         }
 273     }
 274
 275     @Deprecated
 276     @Override
 277     public void close() {
 278     }
 279 }