opendaylight/adsal/clustering/services_implementation/src/main/java/org/opendaylight/controller/clustering/services_implementation/internal/ClusterManager.java

   1
   2 /*
   3  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
   4  *
   5  * This program and the accompanying materials are made available under the
   6  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   7  * and is available at http://www.eclipse.org/legal/epl-v10.html
   8  */
   9
  10 package org.opendaylight.controller.clustering.services_implementation.internal;
  11
  12 import java.net.InetAddress;
  13 import java.net.NetworkInterface;
  14 import java.net.SocketException;
  15 import java.net.UnknownHostException;
  16 import java.net.URI;
  17 import java.net.URISyntaxException;
  18 import java.util.ArrayList;
  19 import java.util.EnumSet;
  20 import java.util.Enumeration;
  21 import java.util.HashSet;
  22 import java.util.List;
  23 import java.util.Properties;
  24 import java.util.Set;
  25 import java.util.StringTokenizer;
  26 import java.util.concurrent.ConcurrentMap;
  27 import java.util.concurrent.TimeUnit;
  28
  29 import javax.transaction.HeuristicMixedException;
  30 import javax.transaction.HeuristicRollbackException;
  31 import javax.transaction.NotSupportedException;
  32 import javax.transaction.RollbackException;
  33 import javax.transaction.SystemException;
  34 import javax.transaction.Transaction;
  35 import javax.transaction.TransactionManager;
  36
  37 import org.infinispan.Cache;
  38 import org.infinispan.configuration.cache.Configuration;
  39 import org.infinispan.configuration.cache.ConfigurationBuilder;
  40 import org.infinispan.configuration.global.GlobalConfigurationBuilder;
  41 import org.infinispan.configuration.parsing.ConfigurationBuilderHolder;
  42 import org.infinispan.configuration.parsing.ParserRegistry;
  43 import org.infinispan.manager.DefaultCacheManager;
  44 import org.infinispan.manager.EmbeddedCacheManager;
  45 import org.infinispan.notifications.Listener;
  46 import org.infinispan.notifications.cachemanagerlistener.annotation.ViewChanged;
  47 import org.infinispan.notifications.cachemanagerlistener.event.ViewChangedEvent;
  48 import org.infinispan.remoting.transport.Address;
  49 import org.infinispan.remoting.transport.Transport;
  50 import org.infinispan.remoting.transport.jgroups.JGroupsAddress;
  51 import org.infinispan.remoting.transport.jgroups.JGroupsTransport;
  52 import org.jgroups.Channel;
  53 import org.jgroups.Event;
  54 import org.jgroups.stack.GossipRouter;
  55 import org.opendaylight.controller.clustering.services.CacheConfigException;
  56 import org.opendaylight.controller.clustering.services.CacheExistException;
  57 import org.opendaylight.controller.clustering.services.CacheListenerAddException;
  58 import org.opendaylight.controller.clustering.services.IClusterServices;
  59 import org.opendaylight.controller.clustering.services.IGetUpdates;
  60 import org.opendaylight.controller.clustering.services.IListenRoleChange;
  61 import org.opendaylight.controller.clustering.services.ListenRoleChangeAddException;
  62 import org.slf4j.Logger;
  63 import org.slf4j.LoggerFactory;
  64
  65 public class ClusterManager implements IClusterServices {
  66     protected static final Logger logger = LoggerFactory
  67             .getLogger(ClusterManager.class);
  68     private DefaultCacheManager cm;
  69     GossipRouter gossiper;
  70     private HashSet<IListenRoleChange> roleChangeListeners;
  71     private ViewChangedListener cacheManagerListener;
  72
  73     private static String loopbackAddress = InetAddress.getLoopbackAddress().getHostAddress();
  74     private static final int gossipRouterPortDefault = 12001;
  75     // defaultTransactionTimeout is 60 seconds
  76     private static int DEFAULT_TRANSACTION_TIMEOUT = 60;
  77
  78     /**
  79      * Start a JGroups GossipRouter if we are a supernode. The
  80      * GosispRouter is nothing more than a simple
  81      * rendevouz-pointer. All the nodes that wants to join the cluster
  82      * will come to any of the rendevouz point and they introduce the
  83      * nodes to all the others. Once the meet and greet phase if over,
  84      * the nodes will open a full-mesh with the remaining n-1 nodes,
  85      * so even if the GossipRouter goes down nothing is lost.
  86      * NOTE: This function has the side effect to set some of the
  87      * JGROUPS configurations, this because in this function already
  88      * we try to retrieve some of the network capabilities of the
  89      * device and so it's better not to do that again
  90      *
  91      *
  92      * @return GossipRouter
  93      */
  94     private GossipRouter startGossiper() {
  95         boolean amIGossipRouter = false;
  96         Integer gossipRouterPort = gossipRouterPortDefault;
  97         InetAddress gossipRouterAddress = null;
  98         String supernodes_list = System.getProperty("supernodes",
  99                 loopbackAddress);
 100         /*
 101          * Check the environment for the "container" variable, if this is set
 102          * and is equal to "lxc", then ODL is running inside an lxc
 103          * container, and address resolution of supernodes will be modified
 104          * accordingly.
 105          */
 106         boolean inContainer = "lxc".equals(System.getenv("container"));
 107         StringBuffer sanitized_supernodes_list = new StringBuffer();
 108         List<InetAddress> myAddresses = new ArrayList<InetAddress>();
 109
 110         if (inContainer) {
 111             logger.trace("DOCKER: Resolving supernode host names using docker container semantics");
 112         }
 113
 114         StringTokenizer supernodes = new StringTokenizer(supernodes_list, ":");
 115         if (supernodes.hasMoreTokens()) {
 116             // Populate the list of my addresses
 117             try {
 118                 Enumeration<NetworkInterface> e = NetworkInterface.getNetworkInterfaces();
 119                 while (e.hasMoreElements()) {
 120                     NetworkInterface n = e.nextElement();
 121                     Enumeration<InetAddress> ee = n.getInetAddresses();
 122                     while (ee.hasMoreElements()) {
 123                         InetAddress i = ee.nextElement();
 124                         myAddresses.add(i);
 125                     }
 126                 }
 127             } catch (SocketException se) {
 128                 logger.error("Cannot get the list of network interfaces");
 129                 return null;
 130             }
 131         }
 132         while (supernodes.hasMoreTokens()) {
 133             String curr_supernode = supernodes.nextToken();
 134             logger.debug("Examining supernode {}", curr_supernode);
 135             StringTokenizer host_port = new StringTokenizer(curr_supernode,
 136                     "[]");
 137             String host;
 138             String port;
 139             Integer port_num = gossipRouterPortDefault;
 140             if (host_port.countTokens() > 2) {
 141                 logger.error("Error parsing supernode {} proceed to the next one",
 142                         curr_supernode);
 143                 continue;
 144             }
 145             host = host_port.nextToken();
 146             InetAddress hostAddr;
 147             /*
 148              * If we are in a container and the hostname begins with a '+', this is
 149              * an indication that we should resolve this host name in the context
 150              * of a docker container.
 151              *
 152              * Specifically this means:
 153              * '+self'   : self reference and the host will be mapped to the value of
 154              *             HOSTNAME in the environment
 155              * '+<name>' : references another container by its name. The docker established
 156              *             environment variables will be used to resolve the host to an
 157              *             IP address.
 158              */
 159             if (inContainer && host != null && host.charAt(0) == '+') {
 160                 if ("+self".equals(host)) {
 161                     host = System.getenv("HOSTNAME");
 162                 } else {
 163                     String link = System.getenv(host.substring(1).toUpperCase() + "_PORT");
 164                     if (link != null) {
 165                         try {
 166                             host = new URI(link).getHost();
 167                         } catch (URISyntaxException e) {
 168                             logger.error("DOCKER: Unable to translate container reference ({}) to host IP Address, will attempt using normal host name",
 169                                 host.substring(1));
 170                         }
 171                     }
 172                 }
 173             }
 174
 175             try {
 176                 hostAddr = InetAddress.getByName(host);
 177             } catch (UnknownHostException ue) {
 178                 logger.error("Host {} is not known", host);
 179                 continue;
 180             }
 181             if (host_port.hasMoreTokens()) {
 182                 port = host_port.nextToken();
 183                 try {
 184                     port_num = Integer.valueOf(port);
 185                 } catch (NumberFormatException ne) {
 186                     logger.error("Supplied supernode gossip port is not recognized, using default gossip port {}",
 187                                  gossipRouterPortDefault);
 188                     port_num = gossipRouterPortDefault;
 189                 }
 190                 if ((port_num > 65535) || (port_num < 0)) {
 191                     logger.error("Supplied supernode gossip port is outside a valid TCP port range");
 192                     port_num = gossipRouterPortDefault;
 193                 }
 194             }
 195             if (!amIGossipRouter) {
 196                 if (host != null) {
 197                     for (InetAddress myAddr : myAddresses) {
 198                         if (myAddr.equals(hostAddr)) {
 199                             amIGossipRouter = true;
 200                             gossipRouterAddress = hostAddr;
 201                             gossipRouterPort = port_num;
 202                             break;
 203                         }
 204                     }
 205                 }
 206             }
 207             if (!sanitized_supernodes_list.toString().equals("")) {
 208                 sanitized_supernodes_list.append(",");
 209             }
 210             sanitized_supernodes_list.append(hostAddr.getHostAddress()).append("[").append(port_num).append("]");
 211         }
 212
 213         if (amIGossipRouter) {
 214             // Set the Jgroups binding interface to the one we got
 215             // from the supernodes attribute
 216             if (gossipRouterAddress != null) {
 217                 System.setProperty("jgroups.tcp.address", gossipRouterAddress
 218                         .getHostAddress());
 219             }
 220         } else {
 221             // Set the Jgroup binding interface to the one we are well
 222             // known outside or else to the first with non-local
 223             // scope.
 224             try {
 225                 String myBind = InetAddress.getLocalHost().getHostAddress();
 226                 if (myBind == null
 227                         || InetAddress.getLocalHost().isLoopbackAddress()) {
 228                     for (InetAddress myAddr : myAddresses) {
 229                         if (myAddr.isLoopbackAddress()
 230                                 || myAddr.isLinkLocalAddress()) {
 231                             logger.debug("Skipping local address {}",
 232                                          myAddr.getHostAddress());
 233                             continue;
 234                         } else {
 235                             // First non-local address
 236                             myBind = myAddr.getHostAddress();
 237                             logger.debug("First non-local address {}", myBind);
 238                             break;
 239                         }
 240                     }
 241                 }
 242                 String jgroupAddress = System
 243                         .getProperty("jgroups.tcp.address");
 244                 if (jgroupAddress == null) {
 245                     if (myBind != null) {
 246                         logger.debug("Set bind address to be {}", myBind);
 247                         System.setProperty("jgroups.tcp.address", myBind);
 248                     } else {
 249                         logger
 250                                 .debug("Set bind address to be LOCALHOST=127.0.0.1");
 251                         System.setProperty("jgroups.tcp.address", "127.0.0.1");
 252                     }
 253                 } else {
 254                     logger.debug("jgroup.tcp.address already set to be {}",
 255                             jgroupAddress);
 256                 }
 257             } catch (UnknownHostException uhe) {
 258                 logger
 259                         .error("Met UnknownHostException while trying to get binding address for jgroups");
 260             }
 261         }
 262
 263         // The supernodes list constitute also the tcpgossip initial
 264         // host list
 265         System.setProperty("jgroups.tcpgossip.initial_hosts",
 266                 sanitized_supernodes_list.toString());
 267         logger.debug("jgroups.tcp.address set to {}",
 268                 System.getProperty("jgroups.tcp.address"));
 269         logger.debug("jgroups.tcpgossip.initial_hosts set to {}",
 270                 System.getProperty("jgroups.tcpgossip.initial_hosts"));
 271         GossipRouter res = null;
 272         if (amIGossipRouter) {
 273             logger.info("I'm a GossipRouter will listen on port {}",
 274                     gossipRouterPort);
 275             // Start a GossipRouter with JMX support
 276             res = new GossipRouter(gossipRouterPort, null, true);
 277         }
 278         return res;
 279     }
 280
 281     private void exitOnSecurityException(Exception ioe) {
 282         Throwable cause = ioe.getCause();
 283         while (cause != null) {
 284             if (cause instanceof java.lang.SecurityException) {
 285                 logger.error("Failed Cluster authentication. Stopping Controller...");
 286                 System.exit(0);
 287             }
 288             cause = cause.getCause();
 289         }
 290     }
 291
 292     public void start() {
 293         this.gossiper = startGossiper();
 294         if (this.gossiper != null) {
 295             logger.debug("Trying to start Gossiper");
 296             try {
 297                 this.gossiper.start();
 298                 logger.info("Started GossipRouter");
 299             } catch (Exception e) {
 300                 logger.error("GossipRouter didn't start. Exception Stack Trace",
 301                              e);
 302             }
 303         }
 304         logger.info("Starting the ClusterManager");
 305         try {
 306             ParserRegistry parser = new ParserRegistry(this.getClass()
 307                     .getClassLoader());
 308             String infinispanConfigFile =
 309                     System.getProperty("org.infinispan.config.file", "config/infinispan-config.xml");
 310             logger.debug("Using configuration file:{}", infinispanConfigFile);
 311             ConfigurationBuilderHolder holder = parser.parseFile(infinispanConfigFile);
 312             GlobalConfigurationBuilder globalBuilder = holder.getGlobalConfigurationBuilder();
 313             globalBuilder.serialization()
 314                     .classResolver(new ClassResolver())
 315                     .build();
 316             this.cm = new DefaultCacheManager(holder, false);
 317             logger.debug("Allocated ClusterManager");
 318             if (this.cm != null) {
 319                 this.cm.start();
 320                 this.cm.startCache();
 321                 logger.debug("Started the ClusterManager");
 322             }
 323         } catch (Exception ioe) {
 324             logger.error("Cannot configure infinispan .. bailing out ");
 325             logger.error("Stack Trace that raised th exception");
 326             logger.error("",ioe);
 327             this.cm = null;
 328             exitOnSecurityException(ioe);
 329             this.stop();
 330         }
 331         logger.debug("Cache Manager has value {}", this.cm);
 332     }
 333
 334     public void stop() {
 335         logger.info("Stopping the ClusterManager");
 336         if (this.cm != null) {
 337             logger.info("Found a valid ClusterManager, now let it be stopped");
 338             this.cm.stop();
 339             this.cm = null;
 340         }
 341         if (this.gossiper != null) {
 342             this.gossiper.stop();
 343             this.gossiper = null;
 344         }
 345     }
 346
 347     @Override
 348     public ConcurrentMap<?, ?> createCache(String containerName,
 349             String cacheName, Set<cacheMode> cMode) throws CacheExistException,
 350             CacheConfigException {
 351         EmbeddedCacheManager manager = this.cm;
 352         Cache<Object,Object> c;
 353         String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
 354         if (manager == null) {
 355             return null;
 356         }
 357
 358         if (manager.cacheExists(realCacheName)) {
 359             throw new CacheExistException();
 360         }
 361
 362         // Sanity check to avoid contrasting parameters between transactional
 363         // and not
 364         if (cMode.containsAll(EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL,
 365                 IClusterServices.cacheMode.TRANSACTIONAL))) {
 366             throw new CacheConfigException();
 367         }
 368
 369         // Sanity check to avoid contrasting parameters between sync and async
 370         if (cMode.containsAll(EnumSet.of(IClusterServices.cacheMode.SYNC, IClusterServices.cacheMode.ASYNC))) {
 371             throw new CacheConfigException();
 372         }
 373
 374         Configuration fromTemplateConfig = null;
 375         /*
 376          * Fetch transactional/non-transactional templates
 377          */
 378         // Check if transactional
 379         if (cMode.contains(IClusterServices.cacheMode.TRANSACTIONAL)) {
 380             fromTemplateConfig = manager.getCacheConfiguration("transactional-type");
 381         } else if (cMode.contains(IClusterServices.cacheMode.NON_TRANSACTIONAL)) {
 382             fromTemplateConfig = manager.getDefaultCacheConfiguration();
 383         }
 384
 385         // If none set the transactional property then just return null
 386         if (fromTemplateConfig == null) {
 387             return null;
 388         }
 389
 390         ConfigurationBuilder builder = new ConfigurationBuilder();
 391         builder.read(fromTemplateConfig);
 392         /*
 393          * Now evaluate async/sync
 394          */
 395         if (cMode.contains(IClusterServices.cacheMode.ASYNC)) {
 396             builder.clustering()
 397                     .cacheMode(fromTemplateConfig.clustering()
 398                             .cacheMode()
 399                             .toAsync());
 400         } else if (cMode.contains(IClusterServices.cacheMode.SYNC)) {
 401             builder.clustering()
 402                     .cacheMode(fromTemplateConfig.clustering()
 403                             .cacheMode()
 404                             .toSync());
 405         }
 406
 407         manager.defineConfiguration(realCacheName, builder.build());
 408         c = manager.getCache(realCacheName);
 409         return c;
 410     }
 411
 412     @Override
 413     public ConcurrentMap<?, ?> getCache(String containerName, String cacheName) {
 414         EmbeddedCacheManager manager = this.cm;
 415         Cache<Object,Object> c;
 416         String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
 417         if (manager == null) {
 418             return null;
 419         }
 420
 421         if (manager.cacheExists(realCacheName)) {
 422             c = manager.getCache(realCacheName);
 423             return c;
 424         }
 425         return null;
 426     }
 427
 428     @Override
 429     public void destroyCache(String containerName, String cacheName) {
 430         EmbeddedCacheManager manager = this.cm;
 431         String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
 432         if (manager == null) {
 433             return;
 434         }
 435         if (manager.cacheExists(realCacheName)) {
 436             manager.removeCache(realCacheName);
 437         }
 438     }
 439
 440     @Override
 441     public boolean existCache(String containerName, String cacheName) {
 442         EmbeddedCacheManager manager = this.cm;
 443
 444         if (manager == null) {
 445             return false;
 446         }
 447
 448         String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
 449         return manager.cacheExists(realCacheName);
 450     }
 451
 452     @Override
 453     public Set<String> getCacheList(String containerName) {
 454         Set<String> perContainerCaches = new HashSet<String>();
 455         EmbeddedCacheManager manager = this.cm;
 456         if (manager == null) {
 457             return null;
 458         }
 459         for (String cacheName : manager.getCacheNames()) {
 460             if (!manager.isRunning(cacheName)) {
 461                 continue;
 462             }
 463             if (cacheName.startsWith("{" + containerName + "}_")) {
 464                 String[] res = cacheName.split("[{}]");
 465                 if (res.length >= 4 && res[1].equals(containerName)
 466                         && res[2].equals("_")) {
 467                     perContainerCaches.add(res[3]);
 468                 }
 469             }
 470         }
 471
 472         return (perContainerCaches);
 473     }
 474
 475     @Override
 476     public Properties getCacheProperties(String containerName, String cacheName) {
 477         EmbeddedCacheManager manager = this.cm;
 478         if (manager == null) {
 479             return null;
 480         }
 481         String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
 482         if (!manager.cacheExists(realCacheName)) {
 483             return null;
 484         }
 485         Configuration conf = manager.getCache(realCacheName).getAdvancedCache()
 486                 .getCacheConfiguration();
 487         Properties p = new Properties();
 488         p.setProperty(IClusterServices.cacheProps.TRANSACTION_PROP.toString(),
 489                 conf.transaction().toString());
 490         p.setProperty(IClusterServices.cacheProps.CLUSTERING_PROP.toString(),
 491                 conf.clustering().toString());
 492         p.setProperty(IClusterServices.cacheProps.LOCKING_PROP.toString(), conf
 493                 .locking().toString());
 494         return p;
 495     }
 496
 497     @Override
 498     public void addListener(String containerName, String cacheName,
 499             IGetUpdates<?, ?> u) throws CacheListenerAddException {
 500         EmbeddedCacheManager manager = this.cm;
 501         Cache<Object,Object> c;
 502         String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
 503         if (manager == null) {
 504             return;
 505         }
 506
 507         if (!manager.cacheExists(realCacheName)) {
 508             throw new CacheListenerAddException();
 509         }
 510         c = manager.getCache(realCacheName);
 511         CacheListenerContainer cl = new CacheListenerContainer(u,
 512                 containerName, cacheName);
 513         c.addListener(cl);
 514     }
 515
 516     @Override
 517     public Set<IGetUpdates<?, ?>> getListeners(String containerName,
 518             String cacheName) {
 519         EmbeddedCacheManager manager = this.cm;
 520         Cache<Object,Object> c;
 521         String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
 522         if (manager == null) {
 523             return null;
 524         }
 525
 526         if (!manager.cacheExists(realCacheName)) {
 527             return null;
 528         }
 529         c = manager.getCache(realCacheName);
 530
 531         Set<IGetUpdates<?, ?>> res = new HashSet<IGetUpdates<?, ?>>();
 532         Set<Object> listeners = c.getListeners();
 533         for (Object listener : listeners) {
 534             if (listener instanceof CacheListenerContainer) {
 535                 CacheListenerContainer cl = (CacheListenerContainer) listener;
 536                 res.add(cl.whichListener());
 537             }
 538         }
 539
 540         return res;
 541     }
 542
 543     @Override
 544     public void removeListener(String containerName, String cacheName,
 545             IGetUpdates<?, ?> u) {
 546         EmbeddedCacheManager manager = this.cm;
 547         Cache<Object,Object> c;
 548         String realCacheName = "{" + containerName + "}_{" + cacheName + "}";
 549         if (manager == null) {
 550             return;
 551         }
 552
 553         if (!manager.cacheExists(realCacheName)) {
 554             return;
 555         }
 556         c = manager.getCache(realCacheName);
 557
 558         Set<Object> listeners = c.getListeners();
 559         for (Object listener : listeners) {
 560             if (listener instanceof CacheListenerContainer) {
 561                 CacheListenerContainer cl = (CacheListenerContainer) listener;
 562                 if (cl.whichListener() == u) {
 563                     c.removeListener(listener);
 564                     return;
 565                 }
 566             }
 567         }
 568     }
 569
 570     @Override
 571     public void tbegin() throws NotSupportedException, SystemException {
 572         // call tbegin with the default timeout
 573         tbegin(DEFAULT_TRANSACTION_TIMEOUT, TimeUnit.SECONDS);
 574     }
 575
 576     @Override
 577     public void tbegin(long timeout, TimeUnit unit) throws NotSupportedException, SystemException {
 578         EmbeddedCacheManager manager = this.cm;
 579         if (manager == null) {
 580             throw new IllegalStateException();
 581         }
 582         TransactionManager tm = manager.getCache("transactional-type")
 583                 .getAdvancedCache().getTransactionManager();
 584         if (tm == null) {
 585             throw new IllegalStateException();
 586         }
 587         long timeoutSec = unit.toSeconds(timeout);
 588         if((timeoutSec > Integer.MAX_VALUE) || (timeoutSec <= 0)) {
 589             // fall back to the default timeout
 590             tm.setTransactionTimeout(DEFAULT_TRANSACTION_TIMEOUT);
 591         } else {
 592             // cast is ok here
 593             // as here we are sure that timeoutSec < = Integer.MAX_VALUE.
 594             tm.setTransactionTimeout((int) timeoutSec);
 595         }
 596         tm.begin();
 597     }
 598
 599     @Override
 600     public void tcommit() throws RollbackException, HeuristicMixedException,
 601             HeuristicRollbackException, java.lang.SecurityException,
 602             java.lang.IllegalStateException, SystemException {
 603         EmbeddedCacheManager manager = this.cm;
 604         if (manager == null) {
 605             throw new IllegalStateException();
 606         }
 607         TransactionManager tm = manager.getCache("transactional-type")
 608                 .getAdvancedCache().getTransactionManager();
 609         if (tm == null) {
 610             throw new IllegalStateException();
 611         }
 612         tm.commit();
 613     }
 614
 615     @Override
 616     public void trollback() throws java.lang.IllegalStateException,
 617             java.lang.SecurityException, SystemException {
 618         EmbeddedCacheManager manager = this.cm;
 619         if (manager == null) {
 620             throw new IllegalStateException();
 621         }
 622         TransactionManager tm = manager.getCache("transactional-type")
 623                 .getAdvancedCache().getTransactionManager();
 624         if (tm == null) {
 625             throw new IllegalStateException();
 626         }
 627         tm.rollback();
 628     }
 629
 630     @Override
 631     public Transaction tgetTransaction() throws SystemException {
 632         EmbeddedCacheManager manager = this.cm;
 633         if (manager == null) {
 634             throw new IllegalStateException();
 635         }
 636         TransactionManager tm = manager.getCache("transactional-type")
 637                 .getAdvancedCache().getTransactionManager();
 638         if (tm == null) {
 639             return null;
 640         }
 641         return tm.getTransaction();
 642     }
 643
 644     @Override
 645     public boolean amIStandby() {
 646         EmbeddedCacheManager manager = this.cm;
 647         if (manager == null) {
 648             // In case we cannot fetch the information, lets assume we
 649             // are standby, so to have less responsibility.
 650             return true;
 651         }
 652         return (!manager.isCoordinator());
 653     }
 654
 655     private InetAddress addressToInetAddress(Address a) {
 656         EmbeddedCacheManager manager = this.cm;
 657         if ((manager == null) || (a == null)) {
 658             // In case we cannot fetch the information, lets assume we
 659             // are standby, so to have less responsibility.
 660             return null;
 661         }
 662         Transport t = manager.getTransport();
 663         if (t instanceof JGroupsTransport) {
 664             JGroupsTransport jt = (JGroupsTransport) t;
 665             Channel c = jt.getChannel();
 666             if (a instanceof JGroupsAddress) {
 667                 JGroupsAddress ja = (JGroupsAddress) a;
 668                 org.jgroups.Address phys = (org.jgroups.Address) c
 669                         .down(new Event(Event.GET_PHYSICAL_ADDRESS, ja
 670                                 .getJGroupsAddress()));
 671                 if (phys instanceof org.jgroups.stack.IpAddress) {
 672                     InetAddress bindAddress = ((org.jgroups.stack.IpAddress) phys)
 673                             .getIpAddress();
 674                     return bindAddress;
 675                 }
 676             }
 677         }
 678         return null;
 679     }
 680
 681     @Override
 682     public List<InetAddress> getClusteredControllers() {
 683         EmbeddedCacheManager manager = this.cm;
 684         if (manager == null) {
 685             return null;
 686         }
 687         List<Address> controllers = manager.getMembers();
 688         if ((controllers == null) || controllers.size() == 0) {
 689             return null;
 690         }
 691
 692         List<InetAddress> clusteredControllers = new ArrayList<InetAddress>();
 693         for (Address a : controllers) {
 694             InetAddress inetAddress = addressToInetAddress(a);
 695             if (inetAddress != null
 696                     && !inetAddress.getHostAddress().equals(loopbackAddress)) {
 697                 clusteredControllers.add(inetAddress);
 698             }
 699         }
 700         return clusteredControllers;
 701     }
 702
 703     @Override
 704     public InetAddress getMyAddress() {
 705         EmbeddedCacheManager manager = this.cm;
 706         if (manager == null) {
 707             return null;
 708         }
 709         return addressToInetAddress(manager.getAddress());
 710     }
 711
 712     @Override
 713     public InetAddress getActiveAddress() {
 714         EmbeddedCacheManager manager = this.cm;
 715         if (manager == null) {
 716             // In case we cannot fetch the information, lets assume we
 717             // are standby, so to have less responsibility.
 718             return null;
 719         }
 720
 721         return addressToInetAddress(manager.getCoordinator());
 722     }
 723
 724     @Override
 725     public void listenRoleChange(IListenRoleChange i)
 726             throws ListenRoleChangeAddException {
 727         EmbeddedCacheManager manager = this.cm;
 728         if (manager == null) {
 729             // In case we cannot fetch the information, lets assume we
 730             // are standby, so to have less responsibility.
 731             throw new ListenRoleChangeAddException();
 732         }
 733
 734         if (this.roleChangeListeners == null) {
 735             this.roleChangeListeners = new HashSet<IListenRoleChange>();
 736             this.cacheManagerListener = new ViewChangedListener(
 737                     this.roleChangeListeners);
 738             manager.addListener(this.cacheManagerListener);
 739         }
 740
 741         if (this.roleChangeListeners != null) {
 742             this.roleChangeListeners.add(i);
 743         }
 744     }
 745
 746     @Override
 747     public void unlistenRoleChange(IListenRoleChange i) {
 748         EmbeddedCacheManager manager = this.cm;
 749         if (manager == null) {
 750             // In case we cannot fetch the information, lets assume we
 751             // are standby, so to have less responsibility.
 752             return;
 753         }
 754
 755         if (this.roleChangeListeners != null) {
 756             this.roleChangeListeners.remove(i);
 757         }
 758
 759         if ((this.roleChangeListeners != null && this.roleChangeListeners
 760                 .isEmpty())
 761                 && (this.cacheManagerListener != null)) {
 762             manager.removeListener(this.cacheManagerListener);
 763             this.cacheManagerListener = null;
 764             this.roleChangeListeners = null;
 765         }
 766     }
 767
 768     @Listener
 769     public class ViewChangedListener {
 770         Set<IListenRoleChange> roleListeners;
 771
 772         public ViewChangedListener(Set<IListenRoleChange> s) {
 773             this.roleListeners = s;
 774         }
 775
 776         @ViewChanged
 777         public void viewChanged(ViewChangedEvent e) {
 778             for (IListenRoleChange i : this.roleListeners) {
 779                 i.newActiveAvailable();
 780             }
 781         }
 782     }
 783 }