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