| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more contributor license |
| * agreements. See the NOTICE file distributed with this work for additional information regarding |
| * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance with the License. You may obtain a |
| * copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software distributed under the License |
| * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express |
| * or implied. See the License for the specific language governing permissions and limitations under |
| * the License. |
| */ |
| package org.apache.geode.pdx; |
| |
| import static org.apache.geode.internal.Assert.assertTrue; |
| import static org.apache.geode.internal.Assert.fail; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertNotNull; |
| |
| import java.util.List; |
| import java.util.Properties; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.TimeoutException; |
| |
| import org.junit.Test; |
| import org.junit.experimental.categories.Category; |
| |
| import org.apache.geode.cache.AttributesFactory; |
| import org.apache.geode.cache.Cache; |
| import org.apache.geode.cache.CacheEvent; |
| import org.apache.geode.cache.CacheFactory; |
| import org.apache.geode.cache.DataPolicy; |
| import org.apache.geode.cache.PartitionAttributesFactory; |
| import org.apache.geode.cache.Region; |
| import org.apache.geode.cache.RegionShortcut; |
| import org.apache.geode.cache.Scope; |
| import org.apache.geode.cache.TransactionEvent; |
| import org.apache.geode.cache.TransactionListener; |
| import org.apache.geode.cache.TransactionWriter; |
| import org.apache.geode.cache.TransactionWriterException; |
| import org.apache.geode.distributed.internal.ClusterDistributionManager; |
| import org.apache.geode.distributed.internal.DistributionMessage; |
| import org.apache.geode.distributed.internal.DistributionMessageObserver; |
| import org.apache.geode.internal.cache.DistributedCacheOperation; |
| import org.apache.geode.internal.cache.GemFireCacheImpl; |
| import org.apache.geode.internal.cache.LocalRegion; |
| import org.apache.geode.pdx.internal.EnumInfo; |
| import org.apache.geode.pdx.internal.PeerTypeRegistration; |
| import org.apache.geode.test.dunit.AsyncInvocation; |
| import org.apache.geode.test.dunit.DUnitBlackboard; |
| import org.apache.geode.test.dunit.Host; |
| import org.apache.geode.test.dunit.SerializableCallable; |
| import org.apache.geode.test.dunit.VM; |
| import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase; |
| import org.apache.geode.test.junit.categories.SerializationTest; |
| |
| @Category({SerializationTest.class}) |
| public class PdxSerializableDUnitTest extends JUnit4CacheTestCase { |
| |
| private static final SimpleClass ANOBJECT = new SimpleClass(57, (byte) 3); |
| private static final String TEST_REGION_NAME = "testSimplePdx"; |
| |
| @Test |
| public void testSimplePut() { |
| Host host = Host.getHost(0); |
| VM vm1 = host.getVM(0); |
| VM vm2 = host.getVM(1); |
| VM vm3 = host.getVM(2); |
| |
| SerializableCallable createRegion = new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| AttributesFactory af = new AttributesFactory(); |
| af.setScope(Scope.DISTRIBUTED_ACK); |
| af.setDataPolicy(DataPolicy.REPLICATE); |
| createRootRegion(TEST_REGION_NAME, af.create()); |
| return null; |
| } |
| }; |
| |
| vm1.invoke(createRegion); |
| vm2.invoke(createRegion); |
| vm3.invoke(createRegion); |
| |
| vm1.invoke(() -> doSimplePut(false)); |
| vm2.invoke(() -> verifySimplePut()); |
| |
| vm3.invoke(new SerializableCallable("check for PDX") { |
| |
| @Override |
| public Object call() throws Exception { |
| assertNotNull(getRootRegion(PeerTypeRegistration.REGION_NAME)); |
| return null; |
| } |
| }); |
| } |
| |
| private void createPR() { |
| getCache().createRegionFactory(RegionShortcut.PARTITION) |
| .setPartitionAttributes(new PartitionAttributesFactory<Integer, Object>().create()) |
| .create(TEST_REGION_NAME); |
| } |
| |
| @Test |
| public void testSimplePutOnPRWithTx() { |
| Host host = Host.getHost(0); |
| VM vm1 = host.getVM(0); |
| VM vm2 = host.getVM(1); |
| VM vm3 = host.getVM(2); |
| |
| vm1.invoke(() -> createPR()); |
| vm2.invoke(() -> createPR()); |
| vm3.invoke(() -> createPR()); |
| vm1.invoke(() -> doSimplePut(true)); |
| |
| vm2.invoke(() -> verifySimplePut()); |
| |
| vm3.invoke(() -> { |
| assertNotNull(getRootRegion(PeerTypeRegistration.REGION_NAME)); |
| }); |
| } |
| |
| @Test |
| public void testTransactionCallbacksNotInvoked() { |
| Host host = Host.getHost(0); |
| VM vm1 = host.getVM(0); |
| VM vm2 = host.getVM(1); |
| |
| SerializableCallable createRegionAndAddPoisonedListener = new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| AttributesFactory af = new AttributesFactory(); |
| af.setScope(Scope.DISTRIBUTED_ACK); |
| af.setDataPolicy(DataPolicy.REPLICATE); |
| createRootRegion(TEST_REGION_NAME, af.create()); |
| addPoisonedTransactionListeners(); |
| return null; |
| } |
| }; |
| |
| vm1.invoke(createRegionAndAddPoisonedListener); |
| vm2.invoke(createRegionAndAddPoisonedListener); |
| vm1.invoke(() -> doSimplePut(true)); |
| vm2.invoke(() -> verifySimplePut()); |
| } |
| |
| @Test |
| public void testPersistenceDefaultDiskStore() throws Throwable { |
| |
| SerializableCallable createRegion = new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| // Make sure the type registry is persistent |
| CacheFactory cf = new CacheFactory(); |
| cf.setPdxPersistent(true); |
| getCache(cf); |
| AttributesFactory af = new AttributesFactory(); |
| af.setScope(Scope.DISTRIBUTED_ACK); |
| af.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE); |
| createRootRegion(TEST_REGION_NAME, af.create()); |
| return null; |
| } |
| }; |
| |
| persistenceTest(createRegion); |
| } |
| |
| @Test |
| public void testPersistenceExplicitDiskStore() throws Throwable { |
| SerializableCallable createRegion = new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| // Make sure the type registry is persistent |
| CacheFactory cf = new CacheFactory(); |
| cf.setPdxPersistent(true); |
| cf.setPdxDiskStore("store1"); |
| Cache cache = getCache(cf); |
| cache.createDiskStoreFactory().setMaxOplogSize(1).setDiskDirs(getDiskDirs()) |
| .create("store1"); |
| AttributesFactory af = new AttributesFactory(); |
| af.setScope(Scope.DISTRIBUTED_ACK); |
| af.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE); |
| af.setDiskStoreName("store1"); |
| createRootRegion(TEST_REGION_NAME, af.create()); |
| return null; |
| } |
| }; |
| persistenceTest(createRegion); |
| } |
| |
| |
| private void persistenceTest(SerializableCallable createRegion) throws Throwable { |
| Host host = Host.getHost(0); |
| VM vm1 = host.getVM(0); |
| VM vm2 = host.getVM(1); |
| VM vm3 = host.getVM(2); |
| vm1.invoke(createRegion); |
| vm2.invoke(createRegion); |
| |
| vm1.invoke(() -> doSimplePut(false)); |
| |
| final SerializableCallable checkForObject = new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| Region r = getRootRegion(TEST_REGION_NAME); |
| assertEquals(ANOBJECT, r.get(1)); |
| // Ok, now the type registry should exist |
| assertNotNull(getRootRegion(PeerTypeRegistration.REGION_NAME)); |
| return null; |
| } |
| }; |
| |
| vm2.invoke(checkForObject); |
| |
| SerializableCallable closeCache = new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| closeCache(); |
| return null; |
| } |
| }; |
| |
| |
| // Close the cache in both VMs |
| vm1.invoke(closeCache); |
| vm2.invoke(closeCache); |
| |
| |
| // Now recreate the region, recoverying from disk |
| AsyncInvocation future1 = vm1.invokeAsync(createRegion); |
| AsyncInvocation future2 = vm2.invokeAsync(createRegion); |
| |
| future1.getResult(); |
| future2.getResult(); |
| |
| // Make sure we can still find and deserialize the result. |
| vm1.invoke(checkForObject); |
| vm2.invoke(checkForObject); |
| |
| // Make sure a late comer can still create the type registry. |
| vm3.invoke(createRegion); |
| |
| vm3.invoke(checkForObject); |
| } |
| |
| |
| @Test |
| public void testVmWaitsForPdxType() throws Throwable { |
| VM vm0 = Host.getHost(0).getVM(0); |
| VM vm1 = Host.getHost(0).getVM(1); |
| getBlackboard().initBlackboard(); |
| final Properties properties = getDistributedSystemProperties(); |
| properties.put("conserve-sockets", "false"); |
| |
| // steps: |
| // 1 create two caches and define a PdxType |
| // 2 install a block in VM1 that delays receipt of new PDX types |
| // 3 update the value of the PdxInstance in VM0 using a new Enum type |
| // 4 get the value in VM0 |
| // The result should be that step 4 hangs unless the bug is fixed |
| |
| vm0.invoke("create cache", () -> { |
| Cache cache = getCache(properties); |
| Region region = cache.createRegionFactory(RegionShortcut.REPLICATE).create("testRegion"); |
| region.put("TestObject", new TestPdxObject("aString", 1, 1.0, TestPdxObject.AnEnum.ONE)); |
| }); |
| |
| vm1.invoke("create cache and region", () -> { |
| Cache cache = getCache(properties); |
| // note that initial image transfer in testRegion will cause the object to be serialized in |
| // vm0 |
| // and populate the PdxRegion in this vm |
| cache.createRegionFactory(RegionShortcut.REPLICATE).create("testRegion"); |
| |
| // this message observer will ensure that a new PDX registration doesn't occur |
| final DUnitBlackboard bb = getBlackboard(); |
| DistributionMessageObserver.setInstance(new DistributionMessageObserver() { |
| @Override |
| public void beforeProcessMessage(ClusterDistributionManager dm, DistributionMessage msg) { |
| if (msg instanceof DistributedCacheOperation.CacheOperationMessage) { |
| try { |
| DistributedCacheOperation.CacheOperationMessage cmsg = |
| (DistributedCacheOperation.CacheOperationMessage) msg; |
| String path = cmsg.getRegionPath(); |
| if (path.equals(PeerTypeRegistration.REGION_FULL_PATH)) { |
| System.out |
| .println("message observer found a PDX update message and is stalling: " + msg); |
| try { |
| bb.signalGate("listenerWasInvoked"); |
| bb.waitForGate("pdxObjectGetStarting", 3, TimeUnit.SECONDS); |
| // let the get() commence and block |
| Thread.sleep(30000); |
| System.out.println("message observer after sleep "); |
| } catch (InterruptedException e) { |
| System.out.println("message observer is done stalling1 "); |
| bb.setMailbox("listenerProblem", e); |
| } catch (TimeoutException e) { |
| System.out.println("message observer is done stalling2"); |
| bb.setMailbox("listenerProblem", e); |
| } finally { |
| System.out.println("message observer is done stalling"); |
| } |
| } |
| } catch (Exception e) { |
| e.printStackTrace(); |
| } |
| } |
| } |
| }); |
| }); |
| |
| AsyncInvocation async0 = vm0.invokeAsync("propagate value with new pdx enum type", () -> { |
| Cache cache = getCache(properties); |
| final Region pdxRegion = cache.getRegion(PeerTypeRegistration.REGION_FULL_PATH); |
| final DUnitBlackboard bb = getBlackboard(); |
| // now we register a new Id for our enum in a different thread. This will |
| // block in vm1 due to its message observer |
| Thread t = new Thread("PdxSerializableDUnitTest async thread") { |
| @Override |
| public void run() { |
| bb.signalGate("asyncThreadReady"); |
| // pdxRegion.put(new EnumId(0x3010101), new EnumInfo(TestPdxObject.AnEnum.TWO)); |
| ((GemFireCacheImpl) cache).getPdxRegistry().addRemoteEnum(0x3010101, |
| new EnumInfo(TestPdxObject.AnEnum.TWO)); |
| } |
| }; |
| t.setDaemon(true); |
| t.start(); |
| try { |
| bb.waitForGate("asyncThreadReady", 20, TimeUnit.SECONDS); |
| } catch (TimeoutException e) { |
| fail("timed out"); |
| } |
| // reserialization will use the new Enumeration PDX type |
| Region region = cache.getRegion("testRegion"); |
| bb.waitForGate("listenerWasInvoked", 20, TimeUnit.SECONDS); |
| region.put("TestObject", new TestPdxObject("TestObject'", 2, 2.0, TestPdxObject.AnEnum.TWO)); |
| System.out.println("TestObject added put"); |
| bb.signalGate("pdxObjectPut"); |
| }); |
| |
| // vm0 has sent a new TestObject but vm1 does not have the enum type needed to |
| // deserialize it. |
| AsyncInvocation async1 = vm1.invokeAsync("try to read object w/o enum type", () -> { |
| DUnitBlackboard bb = getBlackboard(); |
| bb.waitForGate("pdxObjectPut", 10, TimeUnit.SECONDS); |
| Region region = getCache(properties).getRegion("testRegion"); |
| bb.signalGate("pdxObjectGetStarting"); |
| Object testObject = region.get("TestObject"); |
| System.out.println("found " + testObject); |
| }); |
| |
| |
| DUnitBlackboard bb = getBlackboard(); |
| |
| try { |
| async0.join(20000); |
| async1.join(10000); |
| if (async0.exceptionOccurred()) { |
| throw async0.getException(); |
| } |
| if (async1.exceptionOccurred()) { |
| throw async1.getException(); |
| } |
| assertTrue(bb.isGateSignaled("listenerWasInvoked")); |
| /* |
| * Throwable throwable = (Throwable)bb.getMailbox("listenerProblem"); if (throwable != null) { |
| * RuntimeException rte = new RuntimeException("message observer had a problem", throwable); |
| * throw rte; } |
| */ |
| } finally { |
| bb.signalGate("pdxObjectGetStarting"); |
| bb.signalGate("pdxObjectPut"); |
| bb.initBlackboard(); |
| } |
| } |
| |
| public static class TestPdxObject implements org.apache.geode.pdx.PdxSerializable { |
| |
| public String stringVar; |
| public int intVar; |
| public double floatVar; |
| |
| public static enum AnEnum { |
| ONE, TWO |
| } |
| |
| ; |
| |
| public AnEnum enumVar; |
| |
| public TestPdxObject() {} |
| |
| public TestPdxObject(String stringVar, int intVar, double floatVar, AnEnum enumVar) { |
| this.stringVar = stringVar; |
| this.intVar = intVar; |
| this.floatVar = floatVar; |
| this.enumVar = enumVar; |
| } |
| |
| @Override |
| public void toData(final PdxWriter writer) { |
| writer.writeString("stringVar", stringVar).writeInt("intVar", intVar) |
| .writeDouble("floatVar", floatVar).writeObject("enumVar", this.enumVar); |
| } |
| |
| @Override |
| public void fromData(final PdxReader reader) { |
| stringVar = reader.readString("stringVar"); |
| intVar = reader.readInt("intVar"); |
| floatVar = reader.readDouble("floatVar"); |
| enumVar = (AnEnum) reader.readObject("enumVar"); |
| } |
| |
| @Override |
| public String toString() { |
| return "TestPdxObject [stringVar=" + stringVar + ", intVar=" + intVar + ", floatVar=" |
| + floatVar + ", enumVar=" + enumVar + "]"; |
| } |
| } |
| |
| /** |
| * add a listener and writer that will throw an exception when invoked if events are for internal |
| * regions |
| */ |
| public void addPoisonedTransactionListeners() { |
| MyTestTransactionListener listener = new MyTestTransactionListener(); |
| getCache().getCacheTransactionManager().addListener(listener); |
| getCache().getCacheTransactionManager().setWriter(listener); |
| } |
| |
| |
| private void doSimplePut(boolean useTransaction) { |
| // Check to make sure the type region is not yet created |
| Region r = getRootRegion(TEST_REGION_NAME); |
| if (useTransaction) { |
| getCache().getCacheTransactionManager().begin(); |
| } |
| try { |
| r.put(1, ANOBJECT); |
| } finally { |
| if (useTransaction) { |
| getCache().getCacheTransactionManager().commit(); |
| } |
| } |
| // Ok, now the type registry should exist |
| assertNotNull(getRootRegion(PeerTypeRegistration.REGION_NAME)); |
| } |
| |
| private void verifySimplePut() { |
| // Ok, now the type registry should exist |
| assertNotNull(getRootRegion(PeerTypeRegistration.REGION_NAME)); |
| Region r = getRootRegion(TEST_REGION_NAME); |
| assertEquals(ANOBJECT, r.get(1)); |
| } |
| |
| private static class MyTestTransactionListener implements TransactionWriter, TransactionListener { |
| private MyTestTransactionListener() { |
| |
| } |
| |
| private void checkEvent(TransactionEvent event) { |
| List<CacheEvent<?, ?>> events = event.getEvents(); |
| System.out.println("MyTestTransactionListener.checkEvent: events are " + events); |
| for (CacheEvent<?, ?> cacheEvent : events) { |
| if (((LocalRegion) cacheEvent.getRegion()).isPdxTypesRegion()) { |
| throw new UnsupportedOperationException("found internal event: " + cacheEvent + " region=" |
| + cacheEvent.getRegion().getName()); |
| } |
| } |
| } |
| |
| @Override |
| public void beforeCommit(TransactionEvent event) throws TransactionWriterException { |
| checkEvent(event); |
| } |
| |
| @Override |
| public void close() {} |
| |
| @Override |
| public void afterCommit(TransactionEvent event) { |
| checkEvent(event); |
| } |
| |
| @Override |
| public void afterFailedCommit(TransactionEvent event) { |
| checkEvent(event); |
| } |
| |
| @Override |
| public void afterRollback(TransactionEvent event) { |
| checkEvent(event); |
| } |
| } |
| } |
