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