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

   1 /*
   2  * Copyright © 2014, 2017 Red Hat, 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
   9 package org.opendaylight.ovsdb.lib.impl;
  10
  11 import com.fasterxml.jackson.annotation.JsonInclude.Include;
  12 import com.fasterxml.jackson.databind.DeserializationFeature;
  13 import com.fasterxml.jackson.databind.ObjectMapper;
  14 import com.google.common.util.concurrent.FutureCallback;
  15 import com.google.common.util.concurrent.Futures;
  16 import com.google.common.util.concurrent.ListenableFuture;
  17 import com.google.common.util.concurrent.ThreadFactoryBuilder;
  18 import io.netty.bootstrap.Bootstrap;
  19 import io.netty.bootstrap.ServerBootstrap;
  20 import io.netty.channel.AdaptiveRecvByteBufAllocator;
  21 import io.netty.channel.Channel;
  22 import io.netty.channel.ChannelFuture;
  23 import io.netty.channel.ChannelInitializer;
  24 import io.netty.channel.ChannelOption;
  25 import io.netty.channel.EventLoopGroup;
  26 import io.netty.channel.nio.NioEventLoopGroup;
  27 import io.netty.channel.socket.SocketChannel;
  28 import io.netty.channel.socket.nio.NioServerSocketChannel;
  29 import io.netty.channel.socket.nio.NioSocketChannel;
  30 import io.netty.handler.codec.string.StringEncoder;
  31 import io.netty.handler.logging.LogLevel;
  32 import io.netty.handler.logging.LoggingHandler;
  33 import io.netty.handler.ssl.SslHandler;
  34 import io.netty.handler.timeout.IdleStateHandler;
  35 import io.netty.handler.timeout.ReadTimeoutHandler;
  36 import io.netty.util.CharsetUtil;
  37 import java.net.InetAddress;
  38 import java.util.ArrayList;
  39 import java.util.Arrays;
  40 import java.util.Collection;
  41 import java.util.List;
  42 import java.util.Map;
  43 import java.util.Map.Entry;
  44 import java.util.Set;
  45 import java.util.concurrent.ConcurrentHashMap;
  46 import java.util.concurrent.ExecutorService;
  47 import java.util.concurrent.Executors;
  48 import java.util.concurrent.ScheduledExecutorService;
  49 import java.util.concurrent.TimeUnit;
  50 import java.util.concurrent.atomic.AtomicBoolean;
  51 import javax.annotation.Nullable;
  52 import javax.inject.Inject;
  53 import javax.inject.Singleton;
  54 import javax.net.ssl.SSLContext;
  55 import javax.net.ssl.SSLEngine;
  56 import javax.net.ssl.SSLEngineResult.HandshakeStatus;
  57 import javax.net.ssl.SSLPeerUnverifiedException;
  58 import org.apache.aries.blueprint.annotation.service.Reference;
  59 import org.apache.aries.blueprint.annotation.service.Service;
  60 import org.opendaylight.aaa.cert.api.ICertificateManager;
  61 import org.opendaylight.ovsdb.lib.OvsdbClient;
  62 import org.opendaylight.ovsdb.lib.OvsdbConnection;
  63 import org.opendaylight.ovsdb.lib.OvsdbConnectionInfo.ConnectionType;
  64 import org.opendaylight.ovsdb.lib.OvsdbConnectionInfo.SocketConnectionType;
  65 import org.opendaylight.ovsdb.lib.OvsdbConnectionListener;
  66 import org.opendaylight.ovsdb.lib.jsonrpc.ExceptionHandler;
  67 import org.opendaylight.ovsdb.lib.jsonrpc.JsonRpcDecoder;
  68 import org.opendaylight.ovsdb.lib.jsonrpc.JsonRpcEndpoint;
  69 import org.opendaylight.ovsdb.lib.jsonrpc.JsonRpcServiceBinderHandler;
  70 import org.opendaylight.ovsdb.lib.message.OvsdbRPC;
  71 import org.slf4j.Logger;
  72 import org.slf4j.LoggerFactory;
  73
  74 /**
  75  * OvsDBConnectionService provides OVSDB connection management functionality which includes
  76  * both Active and Passive connections.
  77  * From the Library perspective, Active OVSDB connections are those that are initiated from
  78  * the Controller towards the ovsdb-manager.
  79  * While Passive OVSDB connections are those that are initiated from the ovs towards
  80  * the controller.
  81  *
  82  * <p>Applications that use OvsDBConnectionService can use the OvsDBConnection class' connect APIs
  83  * to initiate Active connections and can listen to the asynchronous Passive connections via
  84  * registerConnectionListener listener API.
  85  *
  86  * <p>The library is designed as Java modular component that can work in both OSGi and non-OSGi
  87  * environment. Hence a single instance of the service will be active (via Service Registry in OSGi)
  88  * and a Singleton object in a non-OSGi environment.
  89  */
  90 @Singleton
  91 @Service(classes = OvsdbConnection.class)
  92 public class OvsdbConnectionService implements AutoCloseable, OvsdbConnection {
  93     private static final Logger LOG = LoggerFactory.getLogger(OvsdbConnectionService.class);
  94     private static final int IDLE_READER_TIMEOUT = 30;
  95     private static final int READ_TIMEOUT = 180;
  96     private static final String OVSDB_RPC_TASK_TIMEOUT_PARAM = "ovsdb-rpc-task-timeout";
  97     private static final String USE_SSL = "use-ssl";
  98     private static final int RETRY_PERIOD = 100; // retry after 100 milliseconds
  99
 100     private static final ScheduledExecutorService EXECUTOR_SERVICE = Executors.newScheduledThreadPool(10,
 101             new ThreadFactoryBuilder().setNameFormat("OVSDBPassiveConnServ-%d").build());
 102
 103     private static final ExecutorService CONNECTION_NOTIFIER_SERVICE = Executors
 104             .newCachedThreadPool(new ThreadFactoryBuilder().setNameFormat("OVSDBConnNotifSer-%d").build());
 105
 106     private static final StalePassiveConnectionService STALE_PASSIVE_CONNECTION_SERVICE =
 107             new StalePassiveConnectionService((client) -> {
 108                 notifyListenerForPassiveConnection(client);
 109                 return null;
 110             });
 111
 112     private static final Set<OvsdbConnectionListener> CONNECTION_LISTENERS = ConcurrentHashMap.newKeySet();
 113     private static final Map<OvsdbClient, Channel> CONNECTIONS = new ConcurrentHashMap<>();
 114     private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper()
 115             .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
 116             .setSerializationInclusion(Include.NON_NULL);
 117
 118     private volatile boolean useSSL = false;
 119     private final ICertificateManager certManagerSrv;
 120
 121     private volatile int jsonRpcDecoderMaxFrameLength = 100000;
 122     private volatile Channel serverChannel;
 123
 124     private final AtomicBoolean singletonCreated = new AtomicBoolean(false);
 125     private volatile String listenerIp = "0.0.0.0";
 126     private volatile int listenerPort = 6640;
 127
 128     @Inject
 129     public OvsdbConnectionService(@Reference(filter = "type=default-certificate-manager")
 130             final ICertificateManager certManagerSrv) {
 131         this.certManagerSrv = certManagerSrv;
 132     }
 133
 134     /**
 135      * If the SSL flag is enabled, the method internally will establish TLS communication using the default
 136      * ODL certificateManager SSLContext and attributes.
 137      */
 138     @Override
 139     public OvsdbClient connect(final InetAddress address, final int port) {
 140         if (useSSL) {
 141             if (certManagerSrv == null) {
 142                 LOG.error("Certificate Manager service is not available cannot establish the SSL communication.");
 143                 return null;
 144             }
 145             return connectWithSsl(address, port, certManagerSrv);
 146         } else {
 147             return connectWithSsl(address, port, null /* SslContext */);
 148         }
 149     }
 150
 151     @Override
 152     @SuppressWarnings("checkstyle:IllegalCatch")
 153     public OvsdbClient connectWithSsl(final InetAddress address, final int port,
 154             final ICertificateManager certificateManagerSrv) {
 155         try {
 156             Bootstrap bootstrap = new Bootstrap();
 157             bootstrap.group(new NioEventLoopGroup());
 158             bootstrap.channel(NioSocketChannel.class);
 159             bootstrap.option(ChannelOption.TCP_NODELAY, true);
 160             bootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(65535, 65535, 65535));
 161
 162             bootstrap.handler(new ChannelInitializer<SocketChannel>() {
 163                 @Override
 164                 public void initChannel(final SocketChannel channel) throws Exception {
 165                     if (certificateManagerSrv != null && certificateManagerSrv.getServerContext() != null) {
 166                         SSLContext sslContext = certificateManagerSrv.getServerContext();
 167                         /* First add ssl handler if ssl context is given */
 168                         SSLEngine engine =
 169                             sslContext.createSSLEngine(address.toString(), port);
 170                         engine.setUseClientMode(true);
 171                         channel.pipeline().addLast("ssl", new SslHandler(engine));
 172                     }
 173                     channel.pipeline().addLast(
 174                             //new LoggingHandler(LogLevel.INFO),
 175                             new JsonRpcDecoder(jsonRpcDecoderMaxFrameLength),
 176                             new StringEncoder(CharsetUtil.UTF_8),
 177                             new IdleStateHandler(IDLE_READER_TIMEOUT, 0, 0),
 178                             new ReadTimeoutHandler(READ_TIMEOUT),
 179                             new ExceptionHandler(OvsdbConnectionService.this));
 180                 }
 181             });
 182
 183             ChannelFuture future = bootstrap.connect(address, port).sync();
 184             Channel channel = future.channel();
 185             return getChannelClient(channel, ConnectionType.ACTIVE, SocketConnectionType.SSL);
 186         } catch (InterruptedException e) {
 187             LOG.warn("Failed to connect {}:{}", address, port, e);
 188         } catch (Throwable throwable) {
 189             // sync() re-throws exceptions declared as Throwable, so the compiler doesn't see them
 190             LOG.error("Error while binding to address {}, port {}", address, port, throwable);
 191             throw throwable;
 192         }
 193         return null;
 194     }
 195
 196     @Override
 197     public void disconnect(final OvsdbClient client) {
 198         if (client == null) {
 199             return;
 200         }
 201         Channel channel = CONNECTIONS.get(client);
 202         if (channel != null) {
 203             //It's an explicit disconnect from user, so no need to notify back
 204             //to user about the disconnect.
 205             client.setConnectionPublished(false);
 206             channel.disconnect();
 207         }
 208         CONNECTIONS.remove(client);
 209     }
 210
 211     @Override
 212     public void registerConnectionListener(final OvsdbConnectionListener listener) {
 213         LOG.info("registerConnectionListener: registering {}", listener.getClass().getSimpleName());
 214         CONNECTION_LISTENERS.add(listener);
 215         notifyAlreadyExistingConnectionsToListener(listener);
 216     }
 217
 218     private void notifyAlreadyExistingConnectionsToListener(final OvsdbConnectionListener listener) {
 219         for (final OvsdbClient client : getConnections()) {
 220             CONNECTION_NOTIFIER_SERVICE.execute(() -> {
 221                 LOG.trace("Connection {} notified to listener {}", client.getConnectionInfo(), listener);
 222                 listener.connected(client);
 223             });
 224         }
 225     }
 226
 227     @Override
 228     public void unregisterConnectionListener(final OvsdbConnectionListener listener) {
 229         CONNECTION_LISTENERS.remove(listener);
 230     }
 231
 232     private static OvsdbClient getChannelClient(final Channel channel, final ConnectionType type,
 233             final SocketConnectionType socketConnType) {
 234
 235         JsonRpcEndpoint factory = new JsonRpcEndpoint(OBJECT_MAPPER, channel);
 236         JsonRpcServiceBinderHandler binderHandler = new JsonRpcServiceBinderHandler(factory);
 237         binderHandler.setContext(channel);
 238         channel.pipeline().addLast(binderHandler);
 239
 240         OvsdbRPC rpc = factory.getClient(channel, OvsdbRPC.class);
 241         OvsdbClientImpl client = new OvsdbClientImpl(rpc, channel, type, socketConnType);
 242         client.setConnectionPublished(true);
 243         CONNECTIONS.put(client, channel);
 244         ChannelFuture closeFuture = channel.closeFuture();
 245         closeFuture.addListener(new ChannelConnectionHandler(client));
 246         return client;
 247     }
 248
 249     /**
 250      * Method that initiates the Passive OVSDB channel listening functionality.
 251      * By default the ovsdb passive connection will listen in port 6640 which can
 252      * be overridden using the ovsdb.listenPort system property.
 253      */
 254     @Override
 255     public synchronized boolean startOvsdbManager() {
 256         final int ovsdbListenerPort = this.listenerPort;
 257         final String ovsdbListenerIp = this.listenerIp;
 258         if (!singletonCreated.getAndSet(true)) {
 259             LOG.info("startOvsdbManager: Starting");
 260             new Thread(() -> ovsdbManager(ovsdbListenerIp, ovsdbListenerPort)).start();
 261             return true;
 262         } else {
 263             return false;
 264         }
 265     }
 266
 267     /**
 268      * Method that initiates the Passive OVSDB channel listening functionality
 269      * with ssl.By default the ovsdb passive connection will listen in port
 270      * 6640 which can be overridden using the ovsdb.listenPort system property.
 271      */
 272     @Override
 273     public synchronized boolean startOvsdbManagerWithSsl(final String ovsdbListenIp, final int ovsdbListenPort,
 274                                                          final ICertificateManager certificateManagerSrv,
 275                                                          final String[] protocols, final String[] cipherSuites) {
 276         if (!singletonCreated.getAndSet(true)) {
 277             new Thread(() -> ovsdbManagerWithSsl(ovsdbListenIp, ovsdbListenPort,
 278                     certificateManagerSrv, protocols, cipherSuites)).start();
 279             return true;
 280         } else {
 281             return false;
 282         }
 283     }
 284
 285     @Override
 286     public synchronized boolean restartOvsdbManagerWithSsl(final String ovsdbListenIp,
 287         final int ovsdbListenPort,
 288         final ICertificateManager certificateManagerSrv,
 289         final String[] protocols,
 290         final String[] cipherSuites) {
 291         if (singletonCreated.getAndSet(false) && serverChannel != null) {
 292             serverChannel.close();
 293             LOG.info("Server channel closed");
 294         }
 295         serverChannel = null;
 296         return startOvsdbManagerWithSsl(ovsdbListenIp, ovsdbListenPort,
 297             certificateManagerSrv, protocols, cipherSuites);
 298     }
 299
 300     /**
 301      * OVSDB Passive listening thread that uses Netty ServerBootstrap to open
 302      * passive connection handle channel callbacks.
 303      * If the SSL flag is enabled, the method internally will establish TLS communication using the default
 304      * ODL certificateManager SSLContext and attributes.
 305      */
 306     private void ovsdbManager(final String ip, final int port) {
 307         if (useSSL) {
 308             if (certManagerSrv == null) {
 309                 LOG.error("Certificate Manager service is not available cannot establish the SSL communication.");
 310                 return;
 311             }
 312             ovsdbManagerWithSsl(ip, port, certManagerSrv, certManagerSrv.getTlsProtocols(),
 313                     certManagerSrv.getCipherSuites());
 314         } else {
 315             ovsdbManagerWithSsl(ip, port, null /* SslContext */, null, null);
 316         }
 317     }
 318
 319     /**
 320      * OVSDB Passive listening thread that uses Netty ServerBootstrap to open
 321      * passive connection with Ssl and handle channel callbacks.
 322      */
 323     @SuppressWarnings("checkstyle:IllegalCatch")
 324     private void ovsdbManagerWithSsl(final String ip, final int port, final ICertificateManager certificateManagerSrv,
 325                                             final String[] protocols, final String[] cipherSuites) {
 326         EventLoopGroup bossGroup = new NioEventLoopGroup();
 327         EventLoopGroup workerGroup = new NioEventLoopGroup();
 328         try {
 329             ServerBootstrap serverBootstrap = new ServerBootstrap();
 330             serverBootstrap.group(bossGroup, workerGroup)
 331                     .channel(NioServerSocketChannel.class)
 332                     .option(ChannelOption.SO_BACKLOG, 100)
 333                     .handler(new LoggingHandler(LogLevel.INFO))
 334                     .childHandler(new ChannelInitializer<SocketChannel>() {
 335                         @Override
 336                         public void initChannel(final SocketChannel channel) throws Exception {
 337                             LOG.debug("New Passive channel created : {}", channel);
 338                             if (certificateManagerSrv != null && certificateManagerSrv.getServerContext() != null) {
 339                                 /* Add SSL handler first if SSL context is provided */
 340                                 SSLContext sslContext = certificateManagerSrv.getServerContext();
 341                                 SSLEngine engine = sslContext.createSSLEngine();
 342                                 engine.setUseClientMode(false); // work in a server mode
 343                                 engine.setNeedClientAuth(true); // need client authentication
 344                                 if (protocols != null && protocols.length > 0) {
 345                                     //Set supported protocols
 346                                     engine.setEnabledProtocols(protocols);
 347                                     LOG.debug("Supported ssl protocols {}",
 348                                             Arrays.toString(engine.getSupportedProtocols()));
 349                                     LOG.debug("Enabled ssl protocols {}",
 350                                             Arrays.toString(engine.getEnabledProtocols()));
 351                                 }
 352                                 if (cipherSuites != null && cipherSuites.length > 0) {
 353                                     //Set supported cipher suites
 354                                     engine.setEnabledCipherSuites(cipherSuites);
 355                                     LOG.debug("Enabled cipher suites {}",
 356                                             Arrays.toString(engine.getEnabledCipherSuites()));
 357                                 }
 358                                 channel.pipeline().addLast("ssl", new SslHandler(engine));
 359                             }
 360
 361                             channel.pipeline().addLast(
 362                                  new JsonRpcDecoder(jsonRpcDecoderMaxFrameLength),
 363                                  new StringEncoder(CharsetUtil.UTF_8),
 364                                  new IdleStateHandler(IDLE_READER_TIMEOUT, 0, 0),
 365                                  new ReadTimeoutHandler(READ_TIMEOUT),
 366                                  new ExceptionHandler(OvsdbConnectionService.this));
 367
 368                             handleNewPassiveConnection(channel);
 369                         }
 370                     });
 371             serverBootstrap.option(ChannelOption.TCP_NODELAY, true);
 372             serverBootstrap.option(ChannelOption.RCVBUF_ALLOCATOR,
 373                     new AdaptiveRecvByteBufAllocator(65535, 65535, 65535));
 374             // Start the server.
 375             ChannelFuture channelFuture = serverBootstrap.bind(ip, port).sync();
 376             Channel serverListenChannel = channelFuture.channel();
 377             serverChannel = serverListenChannel;
 378             // Wait until the server socket is closed.
 379             serverListenChannel.closeFuture().sync();
 380         } catch (InterruptedException e) {
 381             LOG.error("Thread interrupted", e);
 382         } catch (Throwable throwable) {
 383             // sync() re-throws exceptions declared as Throwable, so the compiler doesn't see them
 384             LOG.error("Error while binding to address {}, port {}", ip, port, throwable);
 385             throw throwable;
 386         } finally {
 387             // Shut down all event loops to terminate all threads.
 388             bossGroup.shutdownGracefully();
 389             workerGroup.shutdownGracefully();
 390         }
 391     }
 392
 393     private static void handleNewPassiveConnection(final OvsdbClient client) {
 394         ListenableFuture<List<String>> echoFuture = client.echo();
 395         LOG.debug("Send echo message to probe the OVSDB switch {}",client.getConnectionInfo());
 396         Futures.addCallback(echoFuture, new FutureCallback<List<String>>() {
 397             @Override
 398             public void onSuccess(@Nullable final List<String> result) {
 399                 LOG.debug("Probe was successful to OVSDB switch {}",client.getConnectionInfo());
 400                 List<OvsdbClient> clientsFromSameNode = getPassiveClientsFromSameNode(client);
 401                 if (clientsFromSameNode.size() == 0) {
 402                     notifyListenerForPassiveConnection(client);
 403                 } else {
 404                     STALE_PASSIVE_CONNECTION_SERVICE.handleNewPassiveConnection(client, clientsFromSameNode);
 405                 }
 406             }
 407
 408             @Override
 409             public void onFailure(final Throwable failureException) {
 410                 LOG.error("Probe failed to OVSDB switch. Disconnecting the channel {}", client.getConnectionInfo());
 411                 client.disconnect();
 412             }
 413         }, CONNECTION_NOTIFIER_SERVICE);
 414     }
 415
 416     private static void handleNewPassiveConnection(final Channel channel) {
 417         if (!channel.isOpen()) {
 418             LOG.warn("Channel {} is not open, skipped further processing of the connection.",channel);
 419             return;
 420         }
 421         SslHandler sslHandler = (SslHandler) channel.pipeline().get("ssl");
 422         if (sslHandler != null) {
 423             class HandleNewPassiveSslRunner implements Runnable {
 424                 private int retryTimes = 3;
 425
 426                 private void retry() {
 427                     if (retryTimes > 0) {
 428                         EXECUTOR_SERVICE.schedule(this,  RETRY_PERIOD, TimeUnit.MILLISECONDS);
 429                     } else {
 430                         LOG.debug("channel closed {}", channel);
 431                         channel.disconnect();
 432                     }
 433                     retryTimes--;
 434                 }
 435
 436                 @Override
 437                 public void run() {
 438                     HandshakeStatus status = sslHandler.engine().getHandshakeStatus();
 439                     LOG.debug("Handshake status {}", status);
 440                     switch (status) {
 441                         case FINISHED:
 442                         case NOT_HANDSHAKING:
 443                             if (sslHandler.engine().getSession().getCipherSuite()
 444                                     .equals("SSL_NULL_WITH_NULL_NULL")) {
 445                                 // Not begin handshake yet. Retry later.
 446                                 LOG.debug("handshake not begin yet {}", status);
 447                                 retry();
 448                             } else {
 449                               //Check if peer is trusted before notifying listeners
 450                                 try {
 451                                     sslHandler.engine().getSession().getPeerCertificates();
 452                                     //Handshake done. Notify listener.
 453                                     OvsdbClient client = getChannelClient(channel, ConnectionType.PASSIVE,
 454                                         SocketConnectionType.SSL);
 455                                     handleNewPassiveConnection(client);
 456                                 } catch (SSLPeerUnverifiedException e) {
 457                                     //Trust manager is still checking peer certificate. Retry later
 458                                     LOG.debug("Peer certifiacte is not verified yet {}", status);
 459                                     retry();
 460                                 }
 461                             }
 462                             break;
 463
 464                         case NEED_UNWRAP:
 465                         case NEED_TASK:
 466                             //Handshake still ongoing. Retry later.
 467                             LOG.debug("handshake not done yet {}", status);
 468                             retry();
 469                             break;
 470
 471                         case NEED_WRAP:
 472                             if (sslHandler.engine().getSession().getCipherSuite()
 473                                     .equals("SSL_NULL_WITH_NULL_NULL")) {
 474                                 /* peer not authenticated. No need to notify listener in this case. */
 475                                 LOG.error("Ssl handshake fail. channel {}", channel);
 476                                 channel.disconnect();
 477                             } else {
 478                                 /*
 479                                  * peer is authenticated. Give some time to wait for completion.
 480                                  * If status is still NEED_WRAP, client might already disconnect.
 481                                  * This happens when the first time client connects to controller in two-way handshake.
 482                                  * After obtaining controller certificate, client will disconnect and start
 483                                  * new connection with controller certificate it obtained.
 484                                  * In this case no need to do anything for the first connection attempt. Just skip
 485                                  * since client will reconnect later.
 486                                  */
 487                                 LOG.debug("handshake not done yet {}", status);
 488                                 retry();
 489                             }
 490                             break;
 491
 492                         default:
 493                             LOG.error("unknown hadshake status {}", status);
 494                     }
 495                 }
 496             }
 497
 498             EXECUTOR_SERVICE.schedule(new HandleNewPassiveSslRunner(),
 499                     RETRY_PERIOD, TimeUnit.MILLISECONDS);
 500         } else {
 501             EXECUTOR_SERVICE.execute(() -> {
 502                 OvsdbClient client = getChannelClient(channel, ConnectionType.PASSIVE,
 503                     SocketConnectionType.NON_SSL);
 504                 handleNewPassiveConnection(client);
 505             });
 506         }
 507     }
 508
 509     public static void channelClosed(final OvsdbClient client) {
 510         LOG.info("Connection closed {}", client.getConnectionInfo());
 511         CONNECTIONS.remove(client);
 512         if (client.isConnectionPublished()) {
 513             for (OvsdbConnectionListener listener : CONNECTION_LISTENERS) {
 514                 listener.disconnected(client);
 515             }
 516         }
 517         STALE_PASSIVE_CONNECTION_SERVICE.clientDisconnected(client);
 518     }
 519
 520     @Override
 521     public Collection<OvsdbClient> getConnections() {
 522         return CONNECTIONS.keySet();
 523     }
 524
 525     @Override
 526     public void close() throws Exception {
 527         LOG.info("OvsdbConnectionService closed");
 528         JsonRpcEndpoint.close();
 529     }
 530
 531     @Override
 532     public OvsdbClient getClient(final Channel channel) {
 533         for (Entry<OvsdbClient, Channel> entry : CONNECTIONS.entrySet()) {
 534             OvsdbClient client = entry.getKey();
 535             Channel ctx = entry.getValue();
 536             if (ctx.equals(channel)) {
 537                 return client;
 538             }
 539         }
 540         return null;
 541     }
 542
 543     private static List<OvsdbClient> getPassiveClientsFromSameNode(final OvsdbClient ovsdbClient) {
 544         List<OvsdbClient> passiveClients = new ArrayList<>();
 545         for (OvsdbClient client : CONNECTIONS.keySet()) {
 546             if (!client.equals(ovsdbClient)
 547                     && client.getConnectionInfo().getRemoteAddress()
 548                             .equals(ovsdbClient.getConnectionInfo().getRemoteAddress())
 549                     && client.getConnectionInfo().getType() == ConnectionType.PASSIVE) {
 550                 passiveClients.add(client);
 551             }
 552         }
 553         return passiveClients;
 554     }
 555
 556     public static void notifyListenerForPassiveConnection(final OvsdbClient client) {
 557         client.setConnectionPublished(true);
 558         for (final OvsdbConnectionListener listener : CONNECTION_LISTENERS) {
 559             CONNECTION_NOTIFIER_SERVICE.execute(() -> {
 560                 LOG.trace("Connection {} notified to listener {}", client.getConnectionInfo(), listener);
 561                 listener.connected(client);
 562             });
 563         }
 564     }
 565
 566     public void setOvsdbRpcTaskTimeout(final int timeout) {
 567         JsonRpcEndpoint.setReaperInterval(timeout);
 568     }
 569
 570     /**
 571      * Set useSSL flag.
 572      *
 573      * @param flag boolean for using ssl
 574      */
 575     public void setUseSsl(final boolean flag) {
 576         useSSL = flag;
 577     }
 578
 579     /**
 580      * Blueprint property setter method. Blueprint call this method and set the value of json rpc decoder
 581      * max frame length to the value configured for config option (json-rpc-decoder-max-frame-length) in
 582      * the configuration file. This option is only configured at the  boot time of the controller. Any
 583      * change at the run time will have no impact.
 584      * @param maxFrameLength Max frame length (default : 100000)
 585      */
 586     public void setJsonRpcDecoderMaxFrameLength(final int maxFrameLength) {
 587         jsonRpcDecoderMaxFrameLength = maxFrameLength;
 588         LOG.info("Json Rpc Decoder Max Frame Length set to : {}", jsonRpcDecoderMaxFrameLength);
 589     }
 590
 591     public void setOvsdbListenerIp(final String ip) {
 592         LOG.info("OVSDB IP for listening connection is set to : {}", ip);
 593         listenerIp = ip;
 594     }
 595
 596     public void setOvsdbListenerPort(final int portNumber) {
 597         LOG.info("OVSDB port for listening connection is set to : {}", portNumber);
 598         listenerPort = portNumber;
 599     }
 600
 601     public void updateConfigParameter(final Map<String, Object> configParameters) {
 602         if (configParameters != null && !configParameters.isEmpty()) {
 603             LOG.debug("Config parameters received : {}", configParameters.entrySet());
 604             for (Map.Entry<String, Object> paramEntry : configParameters.entrySet()) {
 605                 if (paramEntry.getKey().equalsIgnoreCase(OVSDB_RPC_TASK_TIMEOUT_PARAM)) {
 606                     setOvsdbRpcTaskTimeout(Integer.parseInt((String)paramEntry.getValue()));
 607                 } else if (paramEntry.getKey().equalsIgnoreCase(USE_SSL)) {
 608                     useSSL = Boolean.parseBoolean(paramEntry.getValue().toString());
 609                 }
 610             }
 611         }
 612     }
 613 }