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