opendaylight/clustering/integrationtest/src/test/java/org/opendaylight/controller/clustering/services_implementation/internal/ClusteringServicesIT.java

   1 package org.opendaylight.controller.clustering.services_implementation.internal;\r
   2 \r
   3 import static org.junit.Assert.assertEquals;\r
   4 import static org.junit.Assert.assertFalse;\r
   5 import static org.junit.Assert.assertNotNull;\r
   6 import static org.junit.Assert.assertNull;\r
   7 import static org.junit.Assert.assertTrue;\r
   8 import static org.ops4j.pax.exam.CoreOptions.junitBundles;\r
   9 import static org.ops4j.pax.exam.CoreOptions.mavenBundle;\r
  10 import static org.ops4j.pax.exam.CoreOptions.options;\r
  11 import static org.ops4j.pax.exam.CoreOptions.systemPackages;\r
  12 import static org.ops4j.pax.exam.CoreOptions.systemProperty;\r
  13 \r
  14 import java.util.List;\r
  15 import java.util.concurrent.CopyOnWriteArrayList;\r
  16 import java.util.concurrent.TimeUnit;\r
  17 import java.net.InetAddress;\r
  18 import java.util.Dictionary;\r
  19 import java.util.HashSet;\r
  20 import java.util.Hashtable;\r
  21 import java.util.Set;\r
  22 import java.util.concurrent.ConcurrentMap;\r
  23 \r
  24 import javax.inject.Inject;\r
  25 \r
  26 import org.junit.Before;\r
  27 import org.junit.Test;\r
  28 import org.junit.runner.RunWith;\r
  29 import org.opendaylight.controller.clustering.services.CacheConfigException;\r
  30 import org.opendaylight.controller.clustering.services.CacheExistException;\r
  31 import org.opendaylight.controller.clustering.services.CacheListenerAddException;\r
  32 import org.opendaylight.controller.clustering.services.IClusterGlobalServices;\r
  33 import org.opendaylight.controller.clustering.services.IClusterServices;\r
  34 import org.opendaylight.controller.clustering.services.IClusterContainerServices;\r
  35 import org.opendaylight.controller.clustering.services.IClusterServices.cacheMode;\r
  36 import org.opendaylight.controller.clustering.services.IGetUpdates;\r
  37 import org.opendaylight.controller.clustering.services.ICacheUpdateAware;\r
  38 import org.opendaylight.controller.sal.utils.ServiceHelper;\r
  39 import org.opendaylight.controller.sal.core.UpdateType;\r
  40 import org.ops4j.pax.exam.Option;\r
  41 import org.ops4j.pax.exam.Configuration;\r
  42 import org.ops4j.pax.exam.junit.PaxExam;\r
  43 import org.ops4j.pax.exam.util.PathUtils;\r
  44 import org.osgi.framework.Bundle;\r
  45 import org.osgi.framework.BundleContext;\r
  46 import org.osgi.framework.ServiceRegistration;\r
  47 import org.slf4j.Logger;\r
  48 import org.slf4j.LoggerFactory;\r
  49 import java.util.concurrent.CountDownLatch;\r
  50 \r
  51 @RunWith(PaxExam.class)\r
  52 public class ClusteringServicesIT {\r
  53     private Logger log = LoggerFactory\r
  54         .getLogger(ClusteringServicesIT.class);\r
  55     // get the OSGI bundle context\r
  56     @Inject\r
  57     private BundleContext bc;\r
  58     private IClusterServices clusterServices = null;\r
  59     private IClusterContainerServices clusterDefaultServices = null;\r
  60     private IClusterGlobalServices clusterGlobalServices = null;\r
  61 \r
  62     // Configure the OSGi container\r
  63     @Configuration\r
  64     public Option[] config() {\r
  65         return options(\r
  66             //\r
  67             systemProperty("logback.configurationFile").value(\r
  68                 "file:" + PathUtils.getBaseDir()\r
  69                 + "/src/test/resources/logback.xml"),\r
  70             // To start OSGi console for inspection remotely\r
  71             systemProperty("osgi.console").value("2401"),\r
  72             // Set the systemPackages (used by clustering)\r
  73             systemPackages("sun.reflect", "sun.reflect.misc", "sun.misc"),\r
  74             // List framework bundles\r
  75             mavenBundle("equinoxSDK381",\r
  76                         "org.eclipse.equinox.console").versionAsInProject(),\r
  77             mavenBundle("equinoxSDK381",\r
  78                         "org.eclipse.equinox.util").versionAsInProject(),\r
  79             mavenBundle("equinoxSDK381",\r
  80                         "org.eclipse.osgi.services").versionAsInProject(),\r
  81             mavenBundle("equinoxSDK381",\r
  82                         "org.eclipse.equinox.ds").versionAsInProject(),\r
  83             mavenBundle("equinoxSDK381",\r
  84                         "org.apache.felix.gogo.command").versionAsInProject(),\r
  85             mavenBundle("equinoxSDK381",\r
  86                         "org.apache.felix.gogo.runtime").versionAsInProject(),\r
  87             mavenBundle("equinoxSDK381",\r
  88                         "org.apache.felix.gogo.shell").versionAsInProject(),\r
  89             // List logger bundles\r
  90             mavenBundle("org.slf4j", "slf4j-api").versionAsInProject(),\r
  91             mavenBundle("org.slf4j", "log4j-over-slf4j").versionAsInProject(),\r
  92             mavenBundle("ch.qos.logback", "logback-core").versionAsInProject(),\r
  93             mavenBundle("ch.qos.logback", "logback-classic").versionAsInProject(),\r
  94             // List all the bundles on which the test case depends\r
  95             mavenBundle("org.opendaylight.controller",\r
  96                         "clustering.services").versionAsInProject(),\r
  97             mavenBundle("org.opendaylight.controller",\r
  98                         "clustering.services-implementation").versionAsInProject(),\r
  99             mavenBundle("org.opendaylight.controller", "sal").versionAsInProject(),\r
 100             mavenBundle("org.opendaylight.controller",\r
 101                         "sal.implementation").versionAsInProject(),\r
 102             mavenBundle("org.opendaylight.controller", "configuration").versionAsInProject(),\r
 103             mavenBundle("org.opendaylight.controller", "containermanager").versionAsInProject(),\r
 104             mavenBundle("org.opendaylight.controller",\r
 105                         "containermanager.it.implementation").versionAsInProject(),\r
 106             mavenBundle("org.jboss.spec.javax.transaction",\r
 107                         "jboss-transaction-api_1.1_spec").versionAsInProject(),\r
 108             mavenBundle("org.apache.commons", "commons-lang3").versionAsInProject(),\r
 109             mavenBundle("org.apache.felix",\r
 110                         "org.apache.felix.dependencymanager").versionAsInProject(),\r
 111             mavenBundle("org.apache.felix",\r
 112                         "org.apache.felix.dependencymanager.shell").versionAsInProject(),\r
 113             mavenBundle("eclipselink", "javax.resource").versionAsInProject(),\r
 114             junitBundles());\r
 115     }\r
 116 \r
 117     private String stateToString(int state) {\r
 118         switch (state) {\r
 119         case Bundle.ACTIVE:\r
 120             return "ACTIVE";\r
 121         case Bundle.INSTALLED:\r
 122             return "INSTALLED";\r
 123         case Bundle.RESOLVED:\r
 124             return "RESOLVED";\r
 125         case Bundle.UNINSTALLED:\r
 126             return "UNINSTALLED";\r
 127         default:\r
 128             return "Not CONVERTED";\r
 129         }\r
 130     }\r
 131 \r
 132     @Before\r
 133     public void areWeReady() {\r
 134         assertNotNull(bc);\r
 135         boolean debugit = false;\r
 136         Bundle b[] = bc.getBundles();\r
 137         for (Bundle element : b) {\r
 138             int state = element.getState();\r
 139             if (state != Bundle.ACTIVE && state != Bundle.RESOLVED) {\r
 140                 log.debug("Bundle:" + element.getSymbolicName() + " state:"\r
 141                           + stateToString(state));\r
 142                 debugit = true;\r
 143             }\r
 144         }\r
 145         if (debugit) {\r
 146             log.debug("Do some debugging because some bundle is "\r
 147                       + "unresolved");\r
 148         }\r
 149 \r
 150         // Assert if true, if false we are good to go!\r
 151         assertFalse(debugit);\r
 152 \r
 153         this.clusterServices = (IClusterServices)ServiceHelper\r
 154             .getGlobalInstance(IClusterServices.class, this);\r
 155         assertNotNull(this.clusterServices);\r
 156 \r
 157         this.clusterDefaultServices = (IClusterContainerServices)ServiceHelper\r
 158             .getInstance(IClusterContainerServices.class, "default", this);\r
 159         assertNotNull(this.clusterDefaultServices);\r
 160 \r
 161         this.clusterGlobalServices = (IClusterGlobalServices)ServiceHelper\r
 162             .getGlobalInstance(IClusterGlobalServices.class, this);\r
 163         assertNotNull(this.clusterGlobalServices);\r
 164     }\r
 165 \r
 166     @Test\r
 167     public void clusterTest() throws CacheExistException, CacheConfigException,\r
 168         CacheListenerAddException {\r
 169 \r
 170         String container1 = "Container1";\r
 171         String container2 = "Container2";\r
 172         String cache1 = "Cache1";\r
 173         String cache2 = "Cache2";\r
 174         String cache3 = "Cache3";\r
 175 \r
 176         HashSet<cacheMode> cacheModeSet = new HashSet<cacheMode>();\r
 177         cacheModeSet.add(cacheMode.NON_TRANSACTIONAL);\r
 178         ConcurrentMap cm11 = this.clusterServices.createCache(container1,\r
 179                 cache1, cacheModeSet);\r
 180         assertNotNull(cm11);\r
 181 \r
 182         assertNull(this.clusterServices.getCache(container2, cache2));\r
 183         assertEquals(cm11, this.clusterServices.getCache(container1, cache1));\r
 184 \r
 185         assertFalse(this.clusterServices.existCache(container2, cache2));\r
 186         assertTrue(this.clusterServices.existCache(container1, cache1));\r
 187 \r
 188         ConcurrentMap cm12 = this.clusterServices.createCache(container1,\r
 189                 cache2, cacheModeSet);\r
 190         ConcurrentMap cm23 = this.clusterServices.createCache(container2,\r
 191                 cache3, cacheModeSet);\r
 192 \r
 193         HashSet<String> cacheList = (HashSet<String>) this.clusterServices\r
 194                 .getCacheList(container1);\r
 195         assertEquals(2, cacheList.size());\r
 196         assertTrue(cacheList.contains(cache1));\r
 197         assertTrue(cacheList.contains(cache2));\r
 198         assertFalse(cacheList.contains(cache3));\r
 199 \r
 200         assertNotNull(this.clusterServices.getCacheProperties(container1,\r
 201                 cache1));\r
 202 \r
 203         HashSet<IGetUpdates<?, ?>> listeners = (HashSet<IGetUpdates<?, ?>>) this.clusterServices\r
 204                 .getListeners(container1, cache1);\r
 205         assertEquals(0, listeners.size());\r
 206 \r
 207         IGetUpdates<?, ?> getUpdate1 = new GetUpdates();\r
 208         this.clusterServices.addListener(container1, cache1, getUpdate1);\r
 209         listeners = (HashSet<IGetUpdates<?, ?>>) this.clusterServices\r
 210                 .getListeners(container1, cache1);\r
 211         assertEquals(1, listeners.size());\r
 212         this.clusterServices.addListener(container1, cache1, new GetUpdates());\r
 213         listeners = (HashSet<IGetUpdates<?, ?>>) this.clusterServices\r
 214                 .getListeners(container1, cache1);\r
 215         assertEquals(2, listeners.size());\r
 216 \r
 217         listeners = (HashSet<IGetUpdates<?, ?>>) this.clusterServices\r
 218                 .getListeners(container2, cache3);\r
 219         assertEquals(0, listeners.size());\r
 220 \r
 221         this.clusterServices.removeListener(container1, cache1, getUpdate1);\r
 222         listeners = (HashSet<IGetUpdates<?, ?>>) this.clusterServices\r
 223                 .getListeners(container1, cache1);\r
 224         assertEquals(1, listeners.size());\r
 225 \r
 226         InetAddress addr = this.clusterServices.getMyAddress();\r
 227         assertNotNull(addr);\r
 228 \r
 229         List<InetAddress> addrList = this.clusterServices\r
 230                 .getClusteredControllers();\r
 231 \r
 232         this.clusterServices.destroyCache(container1, cache1);\r
 233         assertFalse(this.clusterServices.existCache(container1, cache1));\r
 234 \r
 235     }\r
 236 \r
 237     private class GetUpdates implements IGetUpdates<Integer, String> {\r
 238 \r
 239         @Override\r
 240         public void entryCreated(Integer key, String containerName,\r
 241                 String cacheName, boolean originLocal) {\r
 242             return;\r
 243         }\r
 244 \r
 245         @Override\r
 246         public void entryUpdated(Integer key, String newValue,\r
 247                 String containerName, String cacheName, boolean originLocal) {\r
 248             return;\r
 249         }\r
 250 \r
 251         @Override\r
 252         public void entryDeleted(Integer key, String containerName,\r
 253                 String cacheName, boolean originLocal) {\r
 254             return;\r
 255         }\r
 256     }\r
 257 \r
 258     @Test\r
 259     public void clusterContainerAndGlobalTest() throws CacheExistException, CacheConfigException,\r
 260         CacheListenerAddException, InterruptedException {\r
 261         String cache1 = "Cache1";\r
 262         String cache2 = "Cache2";\r
 263         // Lets test the case of caches with same name in different\r
 264         // containers (actually global an container case)\r
 265         String cache3 = "Cache2";\r
 266 \r
 267         HashSet<cacheMode> cacheModeSet = new HashSet<cacheMode>();\r
 268         cacheModeSet.add(cacheMode.NON_TRANSACTIONAL);\r
 269         ConcurrentMap cm11 = this.clusterDefaultServices.createCache(cache1, cacheModeSet);\r
 270         assertNotNull(cm11);\r
 271 \r
 272         assertTrue(this.clusterDefaultServices.existCache(cache1));\r
 273         assertEquals(cm11, this.clusterDefaultServices.getCache(cache1));\r
 274 \r
 275         ConcurrentMap cm12 = this.clusterDefaultServices.createCache(cache2, cacheModeSet);\r
 276         ConcurrentMap cm23 = this.clusterGlobalServices.createCache(cache3, cacheModeSet);\r
 277 \r
 278         // Now given cahe2 and cache3 have same name lets make sure\r
 279         // they don't return the same reference\r
 280         assertNotNull(this.clusterGlobalServices.getCache(cache2));\r
 281         // cm12 reference must be different than cm23\r
 282         assertTrue(cm12 != cm23);\r
 283 \r
 284         HashSet<String> cacheList = (HashSet<String>) this.clusterDefaultServices\r
 285             .getCacheList();\r
 286         assertEquals(2, cacheList.size());\r
 287         assertTrue(cacheList.contains(cache1));\r
 288         assertTrue(cacheList.contains(cache2));\r
 289 \r
 290         assertNotNull(this.clusterDefaultServices.getCacheProperties(cache1));\r
 291 \r
 292         {\r
 293             /***********************************/\r
 294             /* Testing cacheAware in Container */\r
 295             /***********************************/\r
 296             Dictionary<String, Object> props = new Hashtable<String, Object>();\r
 297             Set<String> propSet = new HashSet<String>();\r
 298             propSet.add(cache1);\r
 299             propSet.add(cache2);\r
 300             props.put("cachenames", propSet);\r
 301             CacheAware listener = new CacheAware();\r
 302             CacheAware listenerRepeated = new CacheAware();\r
 303             ServiceRegistration updateServiceReg = ServiceHelper.registerServiceWReg(ICacheUpdateAware.class, "default",\r
 304                                                                                      listener, props);\r
 305             assertNotNull(updateServiceReg);\r
 306 \r
 307             // Register another service for the same caches, this\r
 308             // should not get any update because we don't allow to\r
 309             // override the existing unless before unregistered\r
 310             ServiceRegistration updateServiceRegRepeated = ServiceHelper.registerServiceWReg(ICacheUpdateAware.class,\r
 311                                                                                              "default",\r
 312                                                                                              listenerRepeated, props);\r
 313             assertNotNull(updateServiceRegRepeated);\r
 314             CountDownLatch res = null;\r
 315             List<Update> ups = null;\r
 316             Update up = null;\r
 317             Integer k1 = new Integer(10);\r
 318             Long k2 = new Long(100L);\r
 319 \r
 320             /***********************/\r
 321             /* CREATE NEW KEY CASE */\r
 322             /***********************/\r
 323             // Start monitoring the updates\r
 324             res = listener.restart(2);\r
 325             // modify the cache\r
 326             cm11.put(k1, "foo");\r
 327             // Wait\r
 328             res.await(100L, TimeUnit.SECONDS);\r
 329             // Analyze the updates\r
 330             ups = listener.getUpdates();\r
 331             assertTrue(ups.size() == 2);\r
 332             // Validate that first we get an update (yes even in case of a\r
 333             // new value added)\r
 334             up = ups.get(0);\r
 335             assertTrue(up.t.equals(UpdateType.CHANGED));\r
 336             assertTrue(up.key.equals(k1));\r
 337             assertTrue(up.value.equals("foo"));\r
 338             assertTrue(up.cacheName.equals(cache1));\r
 339             // Validate that we then get a create\r
 340             up = ups.get(1);\r
 341             assertTrue(up.t.equals(UpdateType.ADDED));\r
 342             assertTrue(up.key.equals(k1));\r
 343             assertNull(up.value);\r
 344             assertTrue(up.cacheName.equals(cache1));\r
 345 \r
 346             /*******************************/\r
 347             /* UPDATE AN EXISTING KEY CASE */\r
 348             /*******************************/\r
 349             // Start monitoring the updates\r
 350             res = listener.restart(1);\r
 351             // modify the cache\r
 352             cm11.put(k1, "baz");\r
 353             // Wait\r
 354             res.await(100L, TimeUnit.SECONDS);\r
 355             // Analyze the updates\r
 356             ups = listener.getUpdates();\r
 357             assertTrue(ups.size() == 1);\r
 358             // Validate we get an update with expect fields\r
 359             up = ups.get(0);\r
 360             assertTrue(up.t.equals(UpdateType.CHANGED));\r
 361             assertTrue(up.key.equals(k1));\r
 362             assertTrue(up.value.equals("baz"));\r
 363             assertTrue(up.cacheName.equals(cache1));\r
 364 \r
 365             /**********************************/\r
 366             /* RE-UPDATE AN EXISTING KEY CASE */\r
 367             /**********************************/\r
 368             // Start monitoring the updates\r
 369             res = listener.restart(1);\r
 370             // modify the cache\r
 371             cm11.put(k1, "baz");\r
 372             // Wait\r
 373             res.await(100L, TimeUnit.SECONDS);\r
 374             // Analyze the updates\r
 375             ups = listener.getUpdates();\r
 376             assertTrue(ups.size() == 1);\r
 377             // Validate we get an update with expect fields\r
 378             up = ups.get(0);\r
 379             assertTrue(up.t.equals(UpdateType.CHANGED));\r
 380             assertTrue(up.key.equals(k1));\r
 381             assertTrue(up.value.equals("baz"));\r
 382             assertTrue(up.cacheName.equals(cache1));\r
 383 \r
 384             /********************************/\r
 385             /* REMOVAL OF EXISTING KEY CASE */\r
 386             /********************************/\r
 387             // Start monitoring the updates\r
 388             res = listener.restart(1);\r
 389             // modify the cache\r
 390             cm11.remove(k1);\r
 391             // Wait\r
 392             res.await(100L, TimeUnit.SECONDS);\r
 393             // Analyze the updates\r
 394             ups = listener.getUpdates();\r
 395             assertTrue(ups.size() == 1);\r
 396             // Validate we get a delete with expected fields\r
 397             up = ups.get(0);\r
 398             assertTrue(up.t.equals(UpdateType.REMOVED));\r
 399             assertTrue(up.key.equals(k1));\r
 400             assertNull(up.value);\r
 401             assertTrue(up.cacheName.equals(cache1));\r
 402 \r
 403             /***********************/\r
 404             /* CREATE NEW KEY CASE */\r
 405             /***********************/\r
 406             // Start monitoring the updates\r
 407             res = listener.restart(2);\r
 408             // modify the cache\r
 409             cm12.put(k2, new Short((short)15));\r
 410             // Wait\r
 411             res.await(100L, TimeUnit.SECONDS);\r
 412             // Analyze the updates\r
 413             ups = listener.getUpdates();\r
 414             assertTrue(ups.size() == 2);\r
 415             // Validate that first we get an update (yes even in case of a\r
 416             // new value added)\r
 417             up = ups.get(0);\r
 418             assertTrue(up.t.equals(UpdateType.CHANGED));\r
 419             assertTrue(up.key.equals(k2));\r
 420             assertTrue(up.value.equals(new Short((short)15)));\r
 421             assertTrue(up.cacheName.equals(cache2));\r
 422             // Validate that we then get a create\r
 423             up = ups.get(1);\r
 424             assertTrue(up.t.equals(UpdateType.ADDED));\r
 425             assertTrue(up.key.equals(k2));\r
 426             assertNull(up.value);\r
 427             assertTrue(up.cacheName.equals(cache2));\r
 428 \r
 429             /*******************************/\r
 430             /* UPDATE AN EXISTING KEY CASE */\r
 431             /*******************************/\r
 432             // Start monitoring the updates\r
 433             res = listener.restart(1);\r
 434             // modify the cache\r
 435             cm12.put(k2, "BAZ");\r
 436             // Wait\r
 437             res.await(100L, TimeUnit.SECONDS);\r
 438             // Analyze the updates\r
 439             ups = listener.getUpdates();\r
 440             assertTrue(ups.size() == 1);\r
 441             // Validate we get an update with expect fields\r
 442             up = ups.get(0);\r
 443             assertTrue(up.t.equals(UpdateType.CHANGED));\r
 444             assertTrue(up.key.equals(k2));\r
 445             assertTrue(up.value.equals("BAZ"));\r
 446             assertTrue(up.cacheName.equals(cache2));\r
 447 \r
 448             /********************************/\r
 449             /* REMOVAL OF EXISTING KEY CASE */\r
 450             /********************************/\r
 451             // Start monitoring the updates\r
 452             res = listener.restart(1);\r
 453             // modify the cache\r
 454             cm12.remove(k2);\r
 455             // Wait\r
 456             res.await(100L, TimeUnit.SECONDS);\r
 457             // Analyze the updates\r
 458             ups = listener.getUpdates();\r
 459             assertTrue(ups.size() == 1);\r
 460             // Validate we get a delete with expected fields\r
 461             up = ups.get(0);\r
 462             assertTrue(up.t.equals(UpdateType.REMOVED));\r
 463             assertTrue(up.key.equals(k2));\r
 464             assertNull(up.value);\r
 465             assertTrue(up.cacheName.equals(cache2));\r
 466 \r
 467             /******************************************************************/\r
 468             /* NOW LETS REMOVE THE REGISTRATION AND MAKE SURE NO UPDATS COMES */\r
 469             /******************************************************************/\r
 470             updateServiceReg.unregister();\r
 471             // Start monitoring the updates, noone should come in\r
 472             res = listener.restart(1);\r
 473 \r
 474             /***********************/\r
 475             /* CREATE NEW KEY CASE */\r
 476             /***********************/\r
 477             // modify the cache\r
 478             cm11.put(k1, "foo");\r
 479 \r
 480             /*******************************/\r
 481             /* UPDATE AN EXISTING KEY CASE */\r
 482             /*******************************/\r
 483             // modify the cache\r
 484             cm11.put(k1, "baz");\r
 485 \r
 486             /********************************/\r
 487             /* REMOVAL OF EXISTING KEY CASE */\r
 488             /********************************/\r
 489             // modify the cache\r
 490             cm11.remove(k1);\r
 491 \r
 492             /***********************/\r
 493             /* CREATE NEW KEY CASE */\r
 494             /***********************/\r
 495             // modify the cache\r
 496             cm12.put(k2, new Short((short)15));\r
 497 \r
 498             /*******************************/\r
 499             /* UPDATE AN EXISTING KEY CASE */\r
 500             /*******************************/\r
 501             // modify the cache\r
 502             cm12.put(k2, "BAZ");\r
 503 \r
 504             /********************************/\r
 505             /* REMOVAL OF EXISTING KEY CASE */\r
 506             /********************************/\r
 507             // modify the cache\r
 508             cm12.remove(k2);\r
 509 \r
 510 \r
 511             // Wait to make sure no updates came in, clearly this is\r
 512             // error prone as logic, but cannot find a better way than\r
 513             // this to make sure updates didn't get in\r
 514             res.await(1L, TimeUnit.SECONDS);\r
 515             // Analyze the updates\r
 516             ups = listener.getUpdates();\r
 517             assertTrue(ups.size() == 0);\r
 518         }\r
 519 \r
 520         {\r
 521             /***********************************/\r
 522             /* Testing cacheAware in Global */\r
 523             /***********************************/\r
 524             Dictionary<String, Object> props = new Hashtable<String, Object>();\r
 525             Set<String> propSet = new HashSet<String>();\r
 526             propSet.add(cache3);\r
 527             props.put("cachenames", propSet);\r
 528             CacheAware listener = new CacheAware();\r
 529             ServiceRegistration updateServiceReg = ServiceHelper.registerGlobalServiceWReg(ICacheUpdateAware.class,\r
 530                                                                                            listener, props);\r
 531             assertNotNull(updateServiceReg);\r
 532 \r
 533             CountDownLatch res = null;\r
 534             List<Update> ups = null;\r
 535             Update up = null;\r
 536             Integer k1 = new Integer(10);\r
 537 \r
 538             /***********************/\r
 539             /* CREATE NEW KEY CASE */\r
 540             /***********************/\r
 541             // Start monitoring the updates\r
 542             res = listener.restart(2);\r
 543             // modify the cache\r
 544             cm23.put(k1, "foo");\r
 545             // Wait\r
 546             res.await(100L, TimeUnit.SECONDS);\r
 547             // Analyze the updates\r
 548             ups = listener.getUpdates();\r
 549             assertTrue(ups.size() == 2);\r
 550             // Validate that first we get an update (yes even in case of a\r
 551             // new value added)\r
 552             up = ups.get(0);\r
 553             assertTrue(up.t.equals(UpdateType.CHANGED));\r
 554             assertTrue(up.key.equals(k1));\r
 555             assertTrue(up.value.equals("foo"));\r
 556             assertTrue(up.cacheName.equals(cache3));\r
 557             // Validate that we then get a create\r
 558             up = ups.get(1);\r
 559             assertTrue(up.t.equals(UpdateType.ADDED));\r
 560             assertTrue(up.key.equals(k1));\r
 561             assertNull(up.value);\r
 562             assertTrue(up.cacheName.equals(cache3));\r
 563 \r
 564             /*******************************/\r
 565             /* UPDATE AN EXISTING KEY CASE */\r
 566             /*******************************/\r
 567             // Start monitoring the updates\r
 568             res = listener.restart(1);\r
 569             // modify the cache\r
 570             cm23.put(k1, "baz");\r
 571             // Wait\r
 572             res.await(100L, TimeUnit.SECONDS);\r
 573             // Analyze the updates\r
 574             ups = listener.getUpdates();\r
 575             assertTrue(ups.size() == 1);\r
 576             // Validate we get an update with expect fields\r
 577             up = ups.get(0);\r
 578             assertTrue(up.t.equals(UpdateType.CHANGED));\r
 579             assertTrue(up.key.equals(k1));\r
 580             assertTrue(up.value.equals("baz"));\r
 581             assertTrue(up.cacheName.equals(cache3));\r
 582 \r
 583             /********************************/\r
 584             /* REMOVAL OF EXISTING KEY CASE */\r
 585             /********************************/\r
 586             // Start monitoring the updates\r
 587             res = listener.restart(1);\r
 588             // modify the cache\r
 589             cm23.remove(k1);\r
 590             // Wait\r
 591             res.await(100L, TimeUnit.SECONDS);\r
 592             // Analyze the updates\r
 593             ups = listener.getUpdates();\r
 594             assertTrue(ups.size() == 1);\r
 595             // Validate we get a delete with expected fields\r
 596             up = ups.get(0);\r
 597             assertTrue(up.t.equals(UpdateType.REMOVED));\r
 598             assertTrue(up.key.equals(k1));\r
 599             assertNull(up.value);\r
 600             assertTrue(up.cacheName.equals(cache3));\r
 601 \r
 602             /******************************************************************/\r
 603             /* NOW LETS REMOVE THE REGISTRATION AND MAKE SURE NO UPDATS COMES */\r
 604             /******************************************************************/\r
 605             updateServiceReg.unregister();\r
 606             // Start monitoring the updates, noone should come in\r
 607             res = listener.restart(1);\r
 608 \r
 609             /***********************/\r
 610             /* CREATE NEW KEY CASE */\r
 611             /***********************/\r
 612             // modify the cache\r
 613             cm23.put(k1, "foo");\r
 614 \r
 615             /*******************************/\r
 616             /* UPDATE AN EXISTING KEY CASE */\r
 617             /*******************************/\r
 618             // modify the cache\r
 619             cm23.put(k1, "baz");\r
 620 \r
 621             /********************************/\r
 622             /* REMOVAL OF EXISTING KEY CASE */\r
 623             /********************************/\r
 624             // modify the cache\r
 625             cm23.remove(k1);\r
 626 \r
 627             // Wait to make sure no updates came in, clearly this is\r
 628             // error prone as logic, but cannot find a better way than\r
 629             // this to make sure updates didn't get in\r
 630             res.await(1L, TimeUnit.SECONDS);\r
 631             // Analyze the updates\r
 632             ups = listener.getUpdates();\r
 633             assertTrue(ups.size() == 0);\r
 634         }\r
 635 \r
 636         InetAddress addr = this.clusterDefaultServices.getMyAddress();\r
 637         assertNotNull(addr);\r
 638 \r
 639         List<InetAddress> addrList = this.clusterDefaultServices\r
 640             .getClusteredControllers();\r
 641 \r
 642         this.clusterDefaultServices.destroyCache(cache1);\r
 643         assertFalse(this.clusterDefaultServices.existCache(cache1));\r
 644     }\r
 645 \r
 646     private class Update {\r
 647         Object key;\r
 648         Object value;\r
 649         String cacheName;\r
 650         UpdateType t;\r
 651 \r
 652         Update (UpdateType t, Object key, Object value, String cacheName) {\r
 653             this.t = t;\r
 654             this.key = key;\r
 655             this.value = value;\r
 656             this.cacheName = cacheName;\r
 657         }\r
 658     }\r
 659 \r
 660     private class CacheAware implements ICacheUpdateAware {\r
 661         private CopyOnWriteArrayList<Update> gotUpdates;\r
 662         private CountDownLatch latch = null;\r
 663 \r
 664         CacheAware() {\r
 665             this.gotUpdates = new CopyOnWriteArrayList<Update>();\r
 666         }\r
 667 \r
 668 \r
 669         /**\r
 670          * Restart the monitor of the updates on the CacheAware object\r
 671          *\r
 672          * @param expectedOperations Number of expected updates\r
 673          *\r
 674          * @return a countdown latch which will be used to wait till the updates are done\r
 675          */\r
 676         CountDownLatch restart(int expectedOperations) {\r
 677             this.gotUpdates.clear();\r
 678             this.latch = new CountDownLatch(expectedOperations);\r
 679             return this.latch;\r
 680         }\r
 681 \r
 682         List<Update> getUpdates() {\r
 683             return this.gotUpdates;\r
 684         }\r
 685 \r
 686         @Override\r
 687         public void entryCreated(Object key, String cacheName, boolean originLocal) {\r
 688             log.debug("CACHE[{}] Got an entry created for key:{}", cacheName, key);\r
 689             Update u = new Update(UpdateType.ADDED, key, null, cacheName);\r
 690             this.gotUpdates.add(u);\r
 691             this.latch.countDown();\r
 692         }\r
 693 \r
 694         @Override\r
 695         public void entryUpdated(Object key, Object newValue, String cacheName, boolean originLocal) {\r
 696             log.debug("CACHE[{}] Got an entry updated for key:{} newValue:{}", cacheName, key, newValue);\r
 697             Update u = new Update(UpdateType.CHANGED, key, newValue, cacheName);\r
 698             this.gotUpdates.add(u);\r
 699             this.latch.countDown();\r
 700         }\r
 701 \r
 702         @Override\r
 703         public void entryDeleted(Object key, String cacheName, boolean originLocal) {\r
 704             log.debug("CACHE[{}] Got an entry delete for key:{}", cacheName, key);\r
 705             Update u = new Update(UpdateType.REMOVED, key, null, cacheName);\r
 706             this.gotUpdates.add(u);\r
 707             this.latch.countDown();\r
 708         }\r
 709     }\r
 710 }\r