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

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