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