framework/src/main/java/org/opendaylight/protocol/framework/AbstractDispatcher.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.protocol.framework;
   9
  10 import io.netty.bootstrap.Bootstrap;
  11 import io.netty.bootstrap.ServerBootstrap;
  12 import io.netty.channel.ChannelFuture;
  13 import io.netty.channel.ChannelInitializer;
  14 import io.netty.channel.ChannelOption;
  15 import io.netty.channel.EventLoopGroup;
  16 import io.netty.channel.nio.NioEventLoopGroup;
  17 import io.netty.channel.socket.SocketChannel;
  18 import io.netty.channel.socket.nio.NioServerSocketChannel;
  19 import io.netty.channel.socket.nio.NioSocketChannel;
  20 import io.netty.util.concurrent.DefaultPromise;
  21 import io.netty.util.concurrent.Future;
  22 import io.netty.util.concurrent.GlobalEventExecutor;
  23 import io.netty.util.concurrent.Promise;
  24 import org.slf4j.Logger;
  25 import org.slf4j.LoggerFactory;
  26
  27 import java.io.Closeable;
  28 import java.net.InetSocketAddress;
  29
  30 /**
  31  * Dispatcher class for creating servers and clients. The idea is to first create servers and clients and the run the
  32  * start method that will handle sockets in different thread.
  33  */
  34 public abstract class AbstractDispatcher<S extends ProtocolSession<?>, L extends SessionListener<?, ?, ?>> implements Closeable {
  35
  36         protected interface PipelineInitializer<S extends ProtocolSession<?>> {
  37                 /**
  38                  * Initializes channel by specifying the handlers in its pipeline. Handlers are protocol specific, therefore this
  39                  * method needs to be implemented in protocol specific Dispatchers.
  40                  *
  41                  * @param channel whose pipeline should be defined, also to be passed to {@link SessionNegotiatorFactory}
  42                  * @param promise to be passed to {@link SessionNegotiatorFactory}
  43                  */
  44                 public void initializeChannel(SocketChannel channel, Promise<S> promise);
  45         }
  46
  47
  48         private static final Logger logger = LoggerFactory.getLogger(AbstractDispatcher.class);
  49
  50         private final EventLoopGroup bossGroup;
  51
  52         private final EventLoopGroup workerGroup;
  53
  54         /**
  55          * Internally creates new instances of NioEventLoopGroup, might deplete system resources and result in Too many open files exception.
  56          *
  57          * @deprecated use {@link AbstractDispatcher#AbstractDispatcher(io.netty.channel.EventLoopGroup, io.netty.channel.EventLoopGroup)} instead.
  58          */
  59         @Deprecated
  60         protected AbstractDispatcher() {
  61                 this(new NioEventLoopGroup(),new NioEventLoopGroup());
  62         }
  63
  64         protected AbstractDispatcher(EventLoopGroup bossGroup, EventLoopGroup workerGroup) {
  65                 this.bossGroup = bossGroup;
  66                 this.workerGroup = workerGroup;
  67         }
  68
  69         /**
  70          * Creates server. Each server needs factories to pass their instances to client sessions.
  71          *
  72          * @param address address to which the server should be bound
  73          * @param initializer instance of PipelineInitializer used to initialize the channel pipeline
  74          *
  75          * @return ChannelFuture representing the binding process
  76          */
  77         protected ChannelFuture createServer(final InetSocketAddress address, final PipelineInitializer<S> initializer) {
  78                 final ServerBootstrap b = new ServerBootstrap();
  79                 b.group(this.bossGroup, this.workerGroup);
  80                 b.channel(NioServerSocketChannel.class);
  81                 b.option(ChannelOption.SO_BACKLOG, 128);
  82                 b.childHandler(new ChannelInitializer<SocketChannel>() {
  83
  84                         @Override
  85                         protected void initChannel(final SocketChannel ch) throws Exception {
  86                                 initializer.initializeChannel(ch, new DefaultPromise<S>(GlobalEventExecutor.INSTANCE));
  87                         }
  88                 });
  89                 b.childOption(ChannelOption.SO_KEEPALIVE, true);
  90
  91                 // Bind and start to accept incoming connections.
  92                 final ChannelFuture f = b.bind(address);
  93                 logger.debug("Initiated server {} at {}.", f, address);
  94                 return f;
  95
  96         }
  97
  98         /**
  99          * Creates a client.
 100          *
 101          * @param address remote address
 102          * @param connectStrategy Reconnection strategy to be used when initial connection fails
 103          *
 104          * @return Future representing the connection process. Its result represents the combined success of TCP connection
 105          *               as well as session negotiation.
 106          */
 107         protected Future<S> createClient(final InetSocketAddress address, final ReconnectStrategy strategy, final PipelineInitializer<S> initializer) {
 108                 final Bootstrap b = new Bootstrap();
 109                 final ProtocolSessionPromise<S> p = new ProtocolSessionPromise<S>(address, strategy, b);
 110                 b.group(this.workerGroup).channel(NioSocketChannel.class).option(ChannelOption.SO_KEEPALIVE, true).handler(
 111                                 new ChannelInitializer<SocketChannel>() {
 112
 113                                         @Override
 114                                         protected void initChannel(final SocketChannel ch) throws Exception {
 115                                                 initializer.initializeChannel(ch, p);
 116                                         }
 117                                 });
 118                 p.connect();
 119                 logger.debug("Client created.");
 120                 return p;
 121         }
 122
 123         /**
 124          * Creates a client.
 125          *
 126          * @param address remote address
 127          * @param connectStrategyFactory Factory for creating reconnection strategy to be used when initial connection fails
 128          * @param reestablishStrategy Reconnection strategy to be used when the already-established session fails
 129          *
 130          * @return Future representing the reconnection task. It will report completion based on reestablishStrategy, e.g.
 131          *               success if it indicates no further attempts should be made and failure if it reports an error
 132          */
 133         protected Future<Void> createReconnectingClient(final InetSocketAddress address, final ReconnectStrategyFactory connectStrategyFactory,
 134                         final ReconnectStrategy reestablishStrategy, final PipelineInitializer<S> initializer) {
 135
 136                 final ReconnectPromise<S, L> p = new ReconnectPromise<S, L>(this, address, connectStrategyFactory, reestablishStrategy, initializer);
 137                 p.connect();
 138
 139                 return p;
 140
 141         }
 142
 143     /**
 144      * @deprecated Should only be used with {@link AbstractDispatcher#AbstractDispatcher()}
 145      */
 146     @Deprecated
 147     @Override
 148     public void close() {
 149         try {
 150             this.workerGroup.shutdownGracefully();
 151         } finally {
 152             this.bossGroup.shutdownGracefully();
 153         }
 154     }
 155
 156 }