framework/src/main/java/org/opendaylight/protocol/framework/DispatcherImpl.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.ChannelFutureListener;
  14 import io.netty.channel.ChannelHandler;
  15 import io.netty.channel.ChannelInitializer;
  16 import io.netty.channel.ChannelOption;
  17 import io.netty.channel.EventLoopGroup;
  18 import io.netty.channel.nio.NioEventLoopGroup;
  19 import io.netty.channel.socket.SocketChannel;
  20 import io.netty.channel.socket.nio.NioServerSocketChannel;
  21 import io.netty.channel.socket.nio.NioSocketChannel;
  22 import io.netty.util.concurrent.DefaultPromise;
  23 import io.netty.util.concurrent.Future;
  24
  25 import java.io.IOException;
  26 import java.net.InetSocketAddress;
  27 import java.util.Timer;
  28
  29 import org.slf4j.Logger;
  30 import org.slf4j.LoggerFactory;
  31
  32 /**
  33  * Dispatcher class for creating servers and clients. The idea is to first create servers and clients and the run the
  34  * start method that will handle sockets in different thread.
  35  */
  36 public final class DispatcherImpl implements Dispatcher, SessionParent {
  37
  38         final class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
  39
  40                 private final ProtocolServer server;
  41
  42                 private ProtocolSession session;
  43
  44                 public ServerChannelInitializer(final ProtocolServer server) {
  45                         this.server = server;
  46                 }
  47
  48                 @Override
  49                 protected void initChannel(final SocketChannel ch) throws Exception {
  50                         final ProtocolHandlerFactory factory = new ProtocolHandlerFactory(DispatcherImpl.this.messageFactory);
  51                         final ChannelHandler handler = factory.getSessionOutboundHandler();
  52                         ch.pipeline().addFirst("outbound", handler);
  53                         ch.pipeline().addFirst("decoder", factory.getDecoder());
  54                         this.session = this.server.createSession(DispatcherImpl.this.stateTimer, ch);
  55
  56                         ch.pipeline().addAfter("decoder", "inbound", factory.getSessionInboundHandler(this.session));
  57                         ch.pipeline().addAfter("inbound", "encoder", factory.getEncoder());
  58                 }
  59
  60                 public ProtocolSession getSession() {
  61                         return this.session;
  62                 }
  63
  64         }
  65
  66         final class ClientChannelInitializer extends ChannelInitializer<SocketChannel> {
  67
  68                 private final ProtocolSessionFactory sfactory;
  69
  70                 private final ProtocolConnection connection;
  71
  72                 private ProtocolSession session;
  73
  74                 public ClientChannelInitializer(final ProtocolConnection connection, final ProtocolSessionFactory sfactory) {
  75                         this.connection = connection;
  76                         this.sfactory = sfactory;
  77                 }
  78
  79                 @Override
  80                 protected void initChannel(final SocketChannel ch) throws Exception {
  81                         final ProtocolHandlerFactory factory = new ProtocolHandlerFactory(DispatcherImpl.this.messageFactory);
  82                         final ChannelHandler handler = factory.getSessionOutboundHandler();
  83                         ch.pipeline().addFirst("outbound", handler);
  84                         ch.pipeline().addFirst("decoder", factory.getDecoder());
  85                         this.session = this.sfactory.getProtocolSession(DispatcherImpl.this, DispatcherImpl.this.stateTimer, this.connection, 0,
  86                                         ch.pipeline().context(ProtocolSessionOutboundHandler.class));
  87                         ch.pipeline().addAfter("decoder", "inbound", factory.getSessionInboundHandler(this.session));
  88                         ch.pipeline().addAfter("inbound", "encoder", factory.getEncoder());
  89                 }
  90
  91                 public ProtocolSession getSession() {
  92                         return this.session;
  93                 }
  94
  95         }
  96
  97         final class ProtocolSessionPromise extends DefaultPromise<ProtocolSession> {
  98                 private final ChannelFuture cf;
  99
 100                 ProtocolSessionPromise(final ChannelFuture cf) {
 101                         super();
 102                         this.cf = cf;
 103                 }
 104
 105                 @Override
 106                 public synchronized boolean cancel(final boolean mayInterruptIfRunning) {
 107                         this.cf.cancel(mayInterruptIfRunning);
 108                         return super.cancel(mayInterruptIfRunning);
 109                 }
 110         }
 111
 112         private static final Logger logger = LoggerFactory.getLogger(DispatcherImpl.class);
 113
 114         private final EventLoopGroup bossGroup;
 115
 116         private final EventLoopGroup workerGroup;
 117
 118         /**
 119          * Timer object grouping FSM Timers
 120          */
 121         private final Timer stateTimer;
 122
 123         private final ProtocolMessageFactory messageFactory;
 124
 125         public DispatcherImpl(final ProtocolMessageFactory factory) {
 126                 this.bossGroup = new NioEventLoopGroup();
 127                 this.workerGroup = new NioEventLoopGroup();
 128                 this.stateTimer = new Timer();
 129                 this.messageFactory = factory;
 130         }
 131
 132         @Override
 133         public ProtocolServer createServer(final InetSocketAddress address, final ProtocolConnectionFactory connectionFactory,
 134                         final ProtocolSessionFactory sessionFactory) {
 135                 final ProtocolServer server = new ProtocolServer(address, connectionFactory, sessionFactory);
 136                 final ServerBootstrap b = new ServerBootstrap();
 137                 b.group(this.bossGroup, this.workerGroup);
 138                 b.channel(NioServerSocketChannel.class);
 139                 b.option(ChannelOption.SO_BACKLOG, 128);
 140                 b.childHandler(new ServerChannelInitializer(server));
 141                 b.childOption(ChannelOption.SO_KEEPALIVE, true);
 142
 143                 // Bind and start to accept incoming connections.
 144                 b.bind(address);
 145                 logger.debug("Server {} created.", server);
 146                 return server;
 147         }
 148
 149         @Override
 150         public Future<ProtocolSession> createClient(final ProtocolConnection connection, final ProtocolSessionFactory sfactory) {
 151                 final Bootstrap b = new Bootstrap();
 152                 b.group(this.workerGroup);
 153                 b.channel(NioSocketChannel.class);
 154                 b.option(ChannelOption.SO_KEEPALIVE, true);
 155                 final ClientChannelInitializer init = new ClientChannelInitializer(connection, sfactory);
 156                 b.handler(init);
 157                 final ChannelFuture f = b.connect(connection.getPeerAddress());
 158                 final ProtocolSessionPromise p = new ProtocolSessionPromise(f);
 159
 160                 f.addListener(new ChannelFutureListener() {
 161                         @Override
 162                         public void operationComplete(final ChannelFuture cf) {
 163                                 if (cf.isSuccess()) {
 164                                         p.setSuccess(init.getSession());
 165                                         return;
 166                                 } else if (cf.isCancelled()) {
 167                                         p.cancel(false);
 168                                 } else
 169                                         p.setFailure(cf.cause());
 170                         }
 171                 });
 172                 logger.debug("Client created.");
 173                 return p;
 174         }
 175
 176         @Override
 177         public void close() throws IOException {
 178                 this.workerGroup.shutdownGracefully();
 179                 this.bossGroup.shutdownGracefully();
 180         }
 181
 182         @Override
 183         public void onSessionClosed(final ProtocolSession session) {
 184                 // TODO Auto-generated method stub
 185         }
 186 }