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