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