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