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<?, ?, ?>> {
  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         protected AbstractDispatcher(EventLoopGroup bossGroup, EventLoopGroup workerGroup) {
  55                 this.bossGroup = bossGroup;
  56                 this.workerGroup = workerGroup;
  57         }
  58
  59         /**
  60          * Creates server. Each server needs factories to pass their instances to client sessions.
  61          *
  62          * @param address address to which the server should be bound
  63          * @param initializer instance of PipelineInitializer used to initialize the channel pipeline
  64          *
  65          * @return ChannelFuture representing the binding process
  66          */
  67         protected ChannelFuture createServer(final InetSocketAddress address, final PipelineInitializer<S> initializer) {
  68                 final ServerBootstrap b = new ServerBootstrap();
  69                 b.group(this.bossGroup, this.workerGroup);
  70                 b.channel(NioServerSocketChannel.class);
  71                 b.option(ChannelOption.SO_BACKLOG, 128);
  72                 b.childHandler(new ChannelInitializer<SocketChannel>() {
  73
  74                         @Override
  75                         protected void initChannel(final SocketChannel ch) throws Exception {
  76                                 initializer.initializeChannel(ch, new DefaultPromise<S>(GlobalEventExecutor.INSTANCE));
  77                         }
  78                 });
  79                 b.childOption(ChannelOption.SO_KEEPALIVE, true);
  80
  81                 // Bind and start to accept incoming connections.
  82                 final ChannelFuture f = b.bind(address);
  83                 logger.debug("Initiated server {} at {}.", f, address);
  84                 return f;
  85
  86         }
  87
  88         /**
  89          * Creates a client.
  90          *
  91          * @param address remote address
  92          * @param connectStrategy Reconnection strategy to be used when initial connection fails
  93          *
  94          * @return Future representing the connection process. Its result represents the combined success of TCP connection
  95          *               as well as session negotiation.
  96          */
  97         protected Future<S> createClient(final InetSocketAddress address, final ReconnectStrategy strategy, final PipelineInitializer<S> initializer) {
  98                 final Bootstrap b = new Bootstrap();
  99                 final ProtocolSessionPromise<S> p = new ProtocolSessionPromise<S>(address, strategy, b);
 100                 b.group(this.workerGroup).channel(NioSocketChannel.class).option(ChannelOption.SO_KEEPALIVE, true).handler(
 101                                 new ChannelInitializer<SocketChannel>() {
 102
 103                                         @Override
 104                                         protected void initChannel(final SocketChannel ch) throws Exception {
 105                                                 initializer.initializeChannel(ch, p);
 106                                         }
 107                                 });
 108                 p.connect();
 109                 logger.debug("Client created.");
 110                 return p;
 111         }
 112
 113         /**
 114          * Creates a client.
 115          *
 116          * @param address remote address
 117          * @param connectStrategyFactory Factory for creating reconnection strategy to be used when initial connection fails
 118          * @param reestablishStrategy Reconnection strategy to be used when the already-established session fails
 119          *
 120          * @return Future representing the reconnection task. It will report completion based on reestablishStrategy, e.g.
 121          *               success if it indicates no further attempts should be made and failure if it reports an error
 122          */
 123         protected Future<Void> createReconnectingClient(final InetSocketAddress address, final ReconnectStrategyFactory connectStrategyFactory,
 124                         final ReconnectStrategy reestablishStrategy, final PipelineInitializer<S> initializer) {
 125
 126                 final ReconnectPromise<S, L> p = new ReconnectPromise<S, L>(this, address, connectStrategyFactory, reestablishStrategy, initializer);
 127                 p.connect();
 128
 129                 return p;
 130
 131         }
 132
 133 }