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