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