library/src/main/java/org/opendaylight/ovsdb/lib/impl/OvsdbConnectionService.java

   1 /*
   2  * Copyright (C) 2014 Red Hat, Inc.
   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  * Authors : Madhu Venugopal
   9  */
  10
  11 package org.opendaylight.ovsdb.lib.impl;
  12
  13 import io.netty.bootstrap.Bootstrap;
  14 import io.netty.bootstrap.ServerBootstrap;
  15 import io.netty.channel.AdaptiveRecvByteBufAllocator;
  16 import io.netty.channel.Channel;
  17 import io.netty.channel.ChannelFuture;
  18 import io.netty.channel.ChannelInitializer;
  19 import io.netty.channel.ChannelOption;
  20 import io.netty.channel.EventLoopGroup;
  21 import io.netty.channel.nio.NioEventLoopGroup;
  22 import io.netty.channel.socket.SocketChannel;
  23 import io.netty.channel.socket.nio.NioServerSocketChannel;
  24 import io.netty.channel.socket.nio.NioSocketChannel;
  25 import io.netty.handler.codec.string.StringEncoder;
  26 import io.netty.handler.logging.LogLevel;
  27 import io.netty.handler.logging.LoggingHandler;
  28 import io.netty.util.CharsetUtil;
  29 import io.netty.handler.ssl.SslHandler;
  30
  31 import javax.net.ssl.SSLContext;
  32 import javax.net.ssl.SSLEngine;
  33
  34 import java.net.InetAddress;
  35 import java.util.Collection;
  36 import java.util.Map;
  37 import java.util.Set;
  38 import java.util.concurrent.ExecutorService;
  39 import java.util.concurrent.Executors;
  40
  41 import org.opendaylight.ovsdb.lib.OvsdbClient;
  42 import org.opendaylight.ovsdb.lib.OvsdbConnection;
  43 import org.opendaylight.ovsdb.lib.OvsdbConnectionInfo.ConnectionType;
  44 import org.opendaylight.ovsdb.lib.OvsdbConnectionListener;
  45 import org.opendaylight.ovsdb.lib.jsonrpc.ExceptionHandler;
  46 import org.opendaylight.ovsdb.lib.jsonrpc.JsonRpcDecoder;
  47 import org.opendaylight.ovsdb.lib.jsonrpc.JsonRpcEndpoint;
  48 import org.opendaylight.ovsdb.lib.jsonrpc.JsonRpcServiceBinderHandler;
  49 import org.opendaylight.ovsdb.lib.message.OvsdbRPC;
  50 import org.slf4j.Logger;
  51 import org.slf4j.LoggerFactory;
  52
  53 import com.fasterxml.jackson.annotation.JsonInclude.Include;
  54 import com.fasterxml.jackson.databind.DeserializationFeature;
  55 import com.fasterxml.jackson.databind.ObjectMapper;
  56 import com.google.common.collect.Maps;
  57 import com.google.common.collect.Sets;
  58
  59 /**
  60  * OvsDBConnectionService provides OVSDB connection management functionality which includes
  61  * both Active and Passive connections.
  62  * From the Library perspective, Active OVSDB connections are those that are initiated from
  63  * the Controller towards the ovsdb-manager.
  64  * While Passive OVSDB connections are those that are initiated from the ovs towards
  65  * the controller.
  66  *
  67  * Applications that use OvsDBConnectionService can use the OvsDBConnection class' connect APIs
  68  * to initiate Active connections and can listen to the asynchronous Passive connections via
  69  * registerConnectionListener listener API.
  70  *
  71  * The library is designed as Java modular component that can work in both OSGi and non-OSGi
  72  * environment. Hence a single instance of the service will be active (via Service Registry in OSGi)
  73  * and a Singleton object in a non-OSGi environment.
  74  */
  75 public class OvsdbConnectionService implements OvsdbConnection {
  76     private static final Logger logger = LoggerFactory.getLogger(OvsdbConnectionService.class);
  77     private final static int NUM_THREADS = 3;
  78
  79     // Singleton Service object that can be used in Non-OSGi environment
  80     private static Set<OvsdbConnectionListener> connectionListeners = Sets.newHashSet();
  81     private static Map<OvsdbClient, Channel> connections = Maps.newHashMap();
  82     private static OvsdbConnection connectionService;
  83     private volatile static boolean singletonCreated = false;
  84
  85     public static OvsdbConnection getService() {
  86         if (connectionService == null) {
  87             connectionService = new OvsdbConnectionService();
  88         }
  89         return connectionService;
  90     }
  91     @Override
  92     public OvsdbClient connect(final InetAddress address, final int port) {
  93         return connectWithSsl(address, port, null /* SslContext */);
  94     }
  95     @Override
  96     public OvsdbClient connectWithSsl(final InetAddress address, final int port,
  97                                final SSLContext sslContext) {
  98         try {
  99             Bootstrap bootstrap = new Bootstrap();
 100             bootstrap.group(new NioEventLoopGroup());
 101             bootstrap.channel(NioSocketChannel.class);
 102             bootstrap.option(ChannelOption.TCP_NODELAY, true);
 103             bootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(65535, 65535, 65535));
 104
 105             bootstrap.handler(new ChannelInitializer<SocketChannel>() {
 106                 @Override
 107                 public void initChannel(SocketChannel channel) throws Exception {
 108                     if (sslContext != null) {
 109                         /* First add ssl handler if ssl context is given */
 110                         SSLEngine engine =
 111                            sslContext.createSSLEngine(address.toString(), port);
 112                         engine.setUseClientMode(true);
 113                         channel.pipeline().addLast("ssl", new SslHandler(engine));
 114                     }
 115                     channel.pipeline().addLast(
 116                             //new LoggingHandler(LogLevel.INFO),
 117                             new JsonRpcDecoder(100000),
 118                             new StringEncoder(CharsetUtil.UTF_8),
 119                             new ExceptionHandler());
 120                 }
 121             });
 122
 123             ChannelFuture future = bootstrap.connect(address, port).sync();
 124             Channel channel = future.channel();
 125             OvsdbClient client = getChannelClient(channel, ConnectionType.ACTIVE, Executors.newFixedThreadPool(NUM_THREADS));
 126             return client;
 127         } catch (InterruptedException e) {
 128             System.out.println("Thread was interrupted during connect");
 129         }
 130         return null;
 131     }
 132
 133     @Override
 134     public void disconnect(OvsdbClient client) {
 135         if (client == null) return;
 136         Channel channel = connections.get(client);
 137         if (channel != null) {
 138             channel.disconnect();
 139         }
 140         connections.remove(client);
 141     }
 142
 143     @Override
 144     public void registerConnectionListener(OvsdbConnectionListener listener) {
 145         connectionListeners.add(listener);
 146     }
 147
 148     @Override
 149     public void unregisterConnectionListener(OvsdbConnectionListener listener) {
 150         connectionListeners.remove(listener);
 151     }
 152
 153     private static OvsdbClient getChannelClient(Channel channel, ConnectionType type,
 154                                                 ExecutorService executorService) {
 155         ObjectMapper objectMapper = new ObjectMapper();
 156         objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
 157         objectMapper.setSerializationInclusion(Include.NON_NULL);
 158
 159         JsonRpcEndpoint factory = new JsonRpcEndpoint(objectMapper, channel);
 160         JsonRpcServiceBinderHandler binderHandler = new JsonRpcServiceBinderHandler(factory);
 161         binderHandler.setContext(channel);
 162         channel.pipeline().addLast(binderHandler);
 163
 164         OvsdbRPC rpc = factory.getClient(channel, OvsdbRPC.class);
 165         OvsdbClientImpl client = new OvsdbClientImpl(rpc, channel, type, executorService);
 166         connections.put(client, channel);
 167         ChannelFuture closeFuture = channel.closeFuture();
 168         closeFuture.addListener(new ChannelConnectionHandler(client));
 169         return client;
 170     }
 171
 172     /**
 173      * Method that initiates the Passive OVSDB channel listening functionality.
 174      * By default the ovsdb passive connection will listen in port 6640 which can
 175      * be overridden using the ovsdb.listenPort system property.
 176      */
 177     @Override
 178     synchronized
 179     public boolean startOvsdbManager(final int ovsdbListenPort) {
 180         if (!singletonCreated) {
 181             new Thread() {
 182                 @Override
 183                 public void run() {
 184                     ovsdbManager(ovsdbListenPort);
 185                 }
 186             }.start();
 187             singletonCreated = true;
 188             return true;
 189         } else {
 190             return false;
 191         }
 192     }
 193
 194     /**
 195      * Method that initiates the Passive OVSDB channel listening functionality
 196      * with ssl.By default the ovsdb passive connection will listen in port
 197      * 6640 which can be overridden using the ovsdb.listenPort system property.
 198      */
 199     @Override
 200     synchronized
 201     public boolean startOvsdbManagerWithSsl(final int ovsdbListenPort,
 202                                      final SSLContext sslContext) {
 203         if (!singletonCreated) {
 204             new Thread() {
 205                 @Override
 206                 public void run() {
 207                     ovsdbManagerWithSsl(ovsdbListenPort, sslContext);
 208                 }
 209             }.start();
 210             singletonCreated = true;
 211             return true;
 212         } else {
 213             return false;
 214         }
 215     }
 216
 217     /**
 218      * OVSDB Passive listening thread that uses Netty ServerBootstrap to open
 219      * passive connection handle channel callbacks.
 220      */
 221     private static void ovsdbManager(int port) {
 222         ovsdbManagerWithSsl(port, null /* SslContext */);
 223     }
 224
 225     /**
 226      * OVSDB Passive listening thread that uses Netty ServerBootstrap to open
 227      * passive connection with Ssl and handle channel callbacks.
 228      */
 229     private static void ovsdbManagerWithSsl(int port, final SSLContext sslContext) {
 230         EventLoopGroup bossGroup = new NioEventLoopGroup();
 231         EventLoopGroup workerGroup = new NioEventLoopGroup();
 232         try {
 233             ServerBootstrap b = new ServerBootstrap();
 234             b.group(bossGroup, workerGroup)
 235              .channel(NioServerSocketChannel.class)
 236              .option(ChannelOption.SO_BACKLOG, 100)
 237              .handler(new LoggingHandler(LogLevel.INFO))
 238              .childHandler(new ChannelInitializer<SocketChannel>() {
 239                  @Override
 240                  public void initChannel(SocketChannel channel) throws Exception {
 241                      logger.debug("New Passive channel created : "+ channel.toString());
 242                      if (sslContext != null) {
 243                          /* Add SSL handler first if SSL context is provided */
 244                          SSLEngine engine = sslContext.createSSLEngine();
 245                          engine.setUseClientMode(false); // work in a server mode
 246                          engine.setNeedClientAuth(true); // need client authentication
 247                          channel.pipeline().addLast("ssl", new SslHandler(engine));
 248                      }
 249
 250                      channel.pipeline().addLast(
 251                              new JsonRpcDecoder(100000),
 252                              new StringEncoder(CharsetUtil.UTF_8),
 253                              new ExceptionHandler());
 254
 255                      handleNewPassiveConnection(channel);
 256                  }
 257              });
 258             b.option(ChannelOption.TCP_NODELAY, true);
 259             b.option(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(65535, 65535, 65535));
 260             // Start the server.
 261             ChannelFuture f = b.bind(port).sync();
 262             Channel serverListenChannel =  f.channel();
 263             // Wait until the server socket is closed.
 264             serverListenChannel.closeFuture().sync();
 265         } catch (InterruptedException e) {
 266             logger.error("Thread interrupted", e);
 267         } finally {
 268             // Shut down all event loops to terminate all threads.
 269             bossGroup.shutdownGracefully();
 270             workerGroup.shutdownGracefully();
 271         }
 272     }
 273
 274     private static ExecutorService executorService = Executors.newFixedThreadPool(10);
 275     private static void handleNewPassiveConnection(final Channel channel) {
 276         executorService.execute(new Runnable() {
 277             @Override
 278             public void run() {
 279                 OvsdbClient client = getChannelClient(channel, ConnectionType.PASSIVE, Executors.newFixedThreadPool(NUM_THREADS));
 280                 for (OvsdbConnectionListener listener : connectionListeners) {
 281                     listener.connected(client);
 282                 }
 283             }
 284         });
 285     }
 286
 287     public static void channelClosed(final OvsdbClient client) {
 288         logger.info("Connection closed {}", client.getConnectionInfo().toString());
 289         connections.remove(client);
 290         for (OvsdbConnectionListener listener : connectionListeners) {
 291             listener.disconnected(client);
 292         }
 293     }
 294     @Override
 295     public Collection<OvsdbClient> getConnections() {
 296         return connections.keySet();
 297     }
 298 }