diff --git a/src/main/java/org/springframework/data/gemfire/transaction/GemfireTransactionManager.java b/src/main/java/org/springframework/data/gemfire/transaction/GemfireTransactionManager.java index 98568daa..2d5fd188 100644 --- a/src/main/java/org/springframework/data/gemfire/transaction/GemfireTransactionManager.java +++ b/src/main/java/org/springframework/data/gemfire/transaction/GemfireTransactionManager.java @@ -17,9 +17,15 @@ package org.springframework.data.gemfire.transaction; -import org.apache.geode.cache.Cache; +import static org.springframework.data.gemfire.transaction.GemfireTransactionManager.CacheHolder.newCacheHolder; +import static org.springframework.data.gemfire.transaction.GemfireTransactionManager.CacheTransactionObject.newCacheTransactionObject; + +import java.util.concurrent.TimeUnit; + import org.apache.geode.cache.CacheTransactionManager; +import org.apache.geode.cache.GemFireCache; import org.apache.geode.cache.Region; +import org.apache.geode.cache.TransactionId; import org.springframework.beans.factory.InitializingBean; import org.springframework.transaction.CannotCreateTransactionException; import org.springframework.transaction.NoTransactionException; @@ -33,252 +39,380 @@ import org.springframework.transaction.support.TransactionSynchronizationManager import org.springframework.util.Assert; /** - * Local transaction manager for GemFire Enterprise Fabric (GEF). Provides a - * {@link PlatformTransactionManager} implementation for a single GemFire - * {@link CacheTransactionManager}. + * Local Transaction Management for Pivotal GemFire. Provides a Spring {@link PlatformTransactionManager} implementation + * for the Pivotal GemFire {@link CacheTransactionManager}. * - * Binds one or multiple GemFire regions for the specified {@link Cache} to the - * thread, potentially allowing for one region per cache model. + * Binds one or multiple GemFire {@link Region Regions} for the specified {@link GemFireCache} to the thread, + * potentially allowing for one {@link Region} per {@link GemFireCache} model. * *

- * This local strategy is an alternative to executing cache operations within - * JTA transactions. Its advantage is that is able to work in any environment, - * for example a stand-alone application or a test suite. It is not able - * to provide XA transactions, for example to share transactions with data - * access. + * This local strategy is an alternative to executing cache operations within JTA transactions. + * Its advantage is that is able to work in any environment, for example a stand-alone application + * or a test suite. It is not able to provide XA transactions, for example to share transactions + * with data access. * *

- * To prevent dirty reads, by default, the cache is configured to return copies - * rather then direct references for get operations. As a - * workaround, one could use explicitly deep copy objects before making changes - * to them to avoid unnecessary copying on every fetch. + * By default, to prevent dirty reads, the {@link GemFireCache} is configured to return copies rather then direct references + * for get data access operations. As a workaround, one could use explicitly deep copy objects before + * making changes to them to avoid unnecessary copying on every fetch. * * @author Costin Leau * @author John Blum - * @see org.apache.geode.cache.Cache#setCopyOnRead(boolean) + * @see org.apache.geode.cache.GemFireCache#setCopyOnRead(boolean) * @see org.apache.geode.cache.CacheTransactionManager * @see org.apache.geode.cache.Region#get(Object) * @see org.apache.geode.CopyHelper#copy(Object) * @see org.springframework.transaction.support.AbstractPlatformTransactionManager * @see #setCopyOnRead(boolean) */ -// TODO add lenient behavior if a transaction is already started on the current -// thread (what should happen then) -@SuppressWarnings("serial") +@SuppressWarnings("unused") public class GemfireTransactionManager extends AbstractPlatformTransactionManager implements InitializingBean, ResourceTransactionManager { - private Cache cache; + protected static final TimeUnit DEFAULT_RESUME_WAIT_TIME_UNIT = TimeUnit.SECONDS; + + private GemFireCache cache; private boolean copyOnRead = true; + private Long resumeWaitTime; + + private TimeUnit resumeWaitTimeUnit = DEFAULT_RESUME_WAIT_TIME_UNIT; + /** - * Creates a new GemfireTransactionManager instance. + * Constructs an instance of the {@link GemfireTransactionManager}. */ public GemfireTransactionManager() { } /** - * Creates a new GemfireTransactionManager instance. + * Constructs an instance of the {@link GemfireTransactionManager} initialized with + * the given {@link GemFireCache} reference. * - * @param cache a reference to the GemFire Cache associated with Cache transactions. + * @param cache reference to the {@link GemFireCache} associated with cache transactions. + * @see org.apache.geode.cache.GemFireCache + * @see #afterPropertiesSet() */ - public GemfireTransactionManager(Cache cache) { + public GemfireTransactionManager(GemFireCache cache) { this.cache = cache; afterPropertiesSet(); } + /** + * @inheritDoc + */ @Override public void afterPropertiesSet() { - Assert.notNull(cache, "Cache property is required"); - cache.setCopyOnRead(copyOnRead); + Assert.notNull(this.cache, "Cache is required"); + this.cache.setCopyOnRead(isCopyOnRead()); } + /** + * @inheritDoc + */ @Override protected Object doGetTransaction() throws TransactionException { - CacheTransactionObject txObject = new CacheTransactionObject(); - CacheHolder cacheHolder = (CacheHolder) TransactionSynchronizationManager.getResource(getCache()); - txObject.setHolder(cacheHolder); - return txObject; + return newCacheTransactionObject((CacheHolder) TransactionSynchronizationManager.getResource(getCache())); } + /** + * @inheritDoc + */ @Override protected boolean isExistingTransaction(Object transaction) throws TransactionException { - CacheTransactionObject txObject = (CacheTransactionObject) transaction; - // Consider a pre-bound cache as transaction. - return (txObject.getHolder() != null); + // consider a pre-bound cache as a transaction + return ((CacheTransactionObject) transaction).isHolding(); } + /** + * @inheritDoc + */ @Override protected void doBegin(Object transaction, TransactionDefinition definition) throws TransactionException { - CacheTransactionObject txObject = (CacheTransactionObject) transaction; - - Cache cache; - try { - cache = getCache(); + CacheTransactionObject cacheTransaction = (CacheTransactionObject) transaction; + GemFireCache cache = getCache(); + if (logger.isDebugEnabled()) { - logger.debug("Acquired Cache [" + cache + "] for local Cache transaction"); + logger.debug(String.format("Acquired GemFire Cache [%s] for local cache transaction", cache)); } - txObject.setHolder(new CacheHolder()); - cache.getCacheTransactionManager().begin(); - TransactionSynchronizationManager.bindResource(cache, txObject.getHolder()); - } + CacheTransactionManager cacheTransactionManager = getCacheTransactionManager(); - catch (IllegalStateException ex) { - throw new CannotCreateTransactionException( - "An ongoing transaction already is already associated with the current thread; are there multiple transaction managers ?", - ex); + // begin GemFire local cache transaction + cacheTransactionManager.begin(); + + TransactionId transactionId = cacheTransactionManager.getTransactionId(); + + if (transactionId != null) { + TransactionSynchronizationManager.bindResource(cache, + cacheTransaction.setAndGetHolder(newCacheHolder(transactionId))); + } + } + catch (IllegalStateException e) { + throw new CannotCreateTransactionException(String.format("%1$s; %2$s", + "An existing, ongoing transaction is already associated with the current thread", + "are multiple transaction managers present?"), e); } } + /** + * @inheritDoc + */ @Override protected void doCommit(DefaultTransactionStatus status) throws TransactionException { - // CacheTransactionObject txObject = (CacheTransactionObject) - // status.getTransaction(); - if (status.isDebug()) { - logger.debug("Committing Gemfire local transaction on Cache [" + cache + "]"); - } try { - cache.getCacheTransactionManager().commit(); - } - catch (IllegalStateException ex) { - throw new NoTransactionException( - "No transaction associated with the current thread; are there multiple transaction managers ?", ex); - } + if (status.isDebug()) { + logger.debug("Committing local cache transaction"); + } - catch (org.apache.geode.cache.TransactionException ex) { - throw new GemfireTransactionCommitException("Unexpected failure on commit of Cache local transaction", ex); + getCacheTransactionManager().commit(); + } + catch (IllegalStateException e) { + throw new NoTransactionException( + "No transaction is associated with the current thread; are multiple transaction managers present?", e); + } + catch (org.apache.geode.cache.TransactionException e) { + throw new GemfireTransactionCommitException( + "Unexpected failure occurred on commit of local cache transaction", e); } } + /** + * @inheritDoc + */ + @Override + protected Object doSuspend(Object transaction) throws TransactionException { + if (getCacheTransactionManager().suspend() != null) { + TransactionSynchronizationManager.unbindResource(getCache()); + return ((CacheTransactionObject) transaction).setAndReturnExistingHolder(null); + } + + return null; + } + + /** + * @inheritDoc + */ + @Override + protected void doResume(Object transaction, Object suspendedResources) throws TransactionException { + if (suspendedResources instanceof CacheHolder) { + CacheHolder holder = (CacheHolder) suspendedResources; + + boolean resumeSuccessful = (isResumeWaitTimeSet() + ? getCacheTransactionManager().tryResume(holder.getTransactionId(), + getResumeWaitTime(), getResumeWaitTimeUnit()) + : getCacheTransactionManager().tryResume(holder.getTransactionId())); + + if (resumeSuccessful) { + TransactionSynchronizationManager.bindResource(getCache(), + ((CacheTransactionObject) transaction).setAndGetHolder(holder)); + } + } + } + + /** + * @inheritDoc + */ @Override protected void doRollback(DefaultTransactionStatus status) throws TransactionException { - if (status.isDebug()) { - logger.debug("Rolling back Cache local transaction for [" + cache + "]"); - } try { - cache.getCacheTransactionManager().rollback(); + if (status.isDebug()) { + logger.debug("Rolling back local cache transaction"); + } + + getCacheTransactionManager().rollback(); } - catch (IllegalStateException ex) { + catch (IllegalStateException e) { throw new NoTransactionException( - "No transaction associated with the current thread; are there multiple transaction managers ?", ex); + "No transaction is associated with the current thread; are multiple transaction managers present?", + e); } } - @Override - protected void doSetRollbackOnly(DefaultTransactionStatus status) { - CacheTransactionObject txObject = (CacheTransactionObject) status.getTransaction(); - if (status.isDebug()) { - logger.debug("Setting Gemfire local transaction [" + txObject.getHolder() + "] rollback-only"); - } - txObject.getHolder().setRollbackOnly(); - } - + /** + * @inheritDoc + */ @Override protected void doCleanupAfterCompletion(Object transaction) { - // Remove the cache holder from the thread. - TransactionSynchronizationManager.unbindResource(cache); + TransactionSynchronizationManager.unbindResource(getCache()); } + /** + * @inheritDoc + */ + @Override + protected void doSetRollbackOnly(DefaultTransactionStatus status) { + ((CacheTransactionObject) status.getTransaction()).getHolder().setRollbackOnly(); + } + + /** + * @inheritDoc + */ @Override protected final boolean useSavepointForNestedTransaction() { return false; } /** - * Returns the Cache that this instance manages local transactions for. - * - * @return Gemfire cache + * Sets a reference to the {@link GemFireCache} for which this transaction manager + * will manage local cache transactions. + * @param cache reference to the {@link GemFireCache}. + * @see org.apache.geode.cache.GemFireCache */ - public Cache getCache() { - return cache; - } - - /** - * Sets the Cache that this instance manages local transactions for. - * - * @param cache Gemfire cache - */ - public void setCache(Cache cache) { + public void setCache(GemFireCache cache) { this.cache = cache; } - @Override - public Object getResourceFactory() { - return getCache(); - } - /** - * Sets the GemFire Cache {@link Region} (as an alternative in setting in the Cache directly). + * Returns a reference to the {@link GemFireCache} for which this transaction manager + * will manage local cache transactions. * - * @param the Region key class type. - * @param the Region value class type. - * @param region the Gemfire Cache Region directly involved in the Cache transaction. + * @return a reference to the {@link GemFireCache}. + * @see org.apache.geode.cache.GemFireCache */ - public void setRegion(Region region) { - Assert.notNull(region, "non-null arguments are required"); - this.cache = (Cache)region.getRegionService(); + public GemFireCache getCache() { + return this.cache; + } + + protected CacheTransactionManager getCacheTransactionManager() { + return getCache().getCacheTransactionManager(); } /** - * Indicates whether the cache returns direct references or copies of the - * objects (default) it manages. While copies imply additional work for - * every fetch operation, direct references can cause dirty reads across - * concurrent threads in the same VM, whether or not transactions are used. + * Set whether the cache returns direct object references or copies of the objects it manages. + * While copies imply additional work for every fetch operation, direct object references can + * cause dirty reads across concurrent threads in the same VM, whether or not transactions are used. * - * One could explicitly deep copy objects before making changes (for example - * by using {@link org.apache.geode.CopyHelper#copy(Object)} in which - * case this setting can be set to false. However, unless there - * is a measurable performance penalty, the recommendation is to keep this - * setting to true + * One could explicitly deep copy objects before making changes (for example by using + * {@link org.apache.geode.CopyHelper#copy(Object)} in which case this setting + * can be set to false * - * @param copyOnRead whether copies (default) rather then direct references - * will be returned on fetch operations + * However, unless there is a measurable performance penalty, the recommendation is + * to keep this setting to true. + * + * @param copyOnRead boolean value indicating whether copies (default) rather then direct object references + * will be returned on fetch operations. */ public void setCopyOnRead(boolean copyOnRead) { this.copyOnRead = copyOnRead; } /** - * Indicates whether copy on read is set or not on the transaction manager. + * Indicates whether copy on read is set and used for fetch data access operations. * + * @return the setting for copy-on-read. * @see #setCopyOnRead(boolean) - * @return the copyOnRead */ public boolean isCopyOnRead() { return copyOnRead; } + /** + * Sets the GemFire cache {@link Region} as an alternative in setting in the {@link GemFireCache} directly. + * + * @param {@link Class} type of the {@link Region} key. + * @param {@link Class} type of the {@link Region} value. + * @param region GemFire cache {@link Region} directly involved in the local cache transaction. + * @throws IllegalArgumentException if {@link Region} is {@literal null}. + * @see org.apache.geode.cache.Region + */ + public void setRegion(Region region) { + Assert.notNull(region, "Region must not be null"); + this.cache = (GemFireCache) region.getRegionService(); + } + + /** + * @inheritDoc + */ + @Override + public Object getResourceFactory() { + return getCache(); + } + + public void setResumeWaitTime(Long resumeWaitTime) { + this.resumeWaitTime = resumeWaitTime; + } + + protected Long getResumeWaitTime() { + return this.resumeWaitTime; + } + + protected boolean isResumeWaitTimeSet() { + Long resumeWaitTime = getResumeWaitTime(); + return (resumeWaitTime != null && resumeWaitTime > 0); + } + + public void setResumeWaitTimeUnit(TimeUnit resumeWaitTimeUnit) { + this.resumeWaitTimeUnit = resumeWaitTimeUnit; + } + + protected TimeUnit getResumeWaitTimeUnit() { + TimeUnit localResumeWaitTimeUnit = this.resumeWaitTimeUnit; + return (localResumeWaitTimeUnit != null ? localResumeWaitTimeUnit : DEFAULT_RESUME_WAIT_TIME_UNIT); + } + /** * GemfireTM local transaction object. * * @author Costin Leau */ - private static class CacheTransactionObject { + protected static class CacheTransactionObject { + private CacheHolder cacheHolder; - public CacheHolder getHolder() { - return cacheHolder; + static CacheTransactionObject newCacheTransactionObject(CacheHolder cacheHolder) { + CacheTransactionObject transactionObject = new CacheTransactionObject(); + transactionObject.setHolder(cacheHolder); + return transactionObject; } - public void setHolder(CacheHolder cacheHolder) { - this.cacheHolder = cacheHolder; + boolean isHolding() { + return (getHolder() != null); + } + + CacheHolder getHolder() { + return this.cacheHolder; + } + + void setHolder(CacheHolder holder) { + this.cacheHolder = holder; + } + + CacheHolder setAndGetHolder(CacheHolder holder) { + setHolder(holder); + return getHolder(); + } + + CacheHolder setAndReturnExistingHolder(CacheHolder cacheHolder) { + CacheHolder existingHolder = getHolder(); + setHolder(cacheHolder); + return existingHolder; } } - private static class CacheHolder { + protected static class CacheHolder { private boolean rollbackOnly = false; - @SuppressWarnings("unused") - public boolean isRollbackOnly() { - return rollbackOnly; + private TransactionId transactionId; + + static CacheHolder newCacheHolder(TransactionId transactionId) { + CacheHolder cacheHolder = new CacheHolder(); + cacheHolder.transactionId = transactionId; + return cacheHolder; } - public void setRollbackOnly() { - rollbackOnly = true; + public boolean isRollbackOnly() { + return this.rollbackOnly; + } + + void setRollbackOnly() { + this.rollbackOnly = true; + } + + TransactionId getTransactionId() { + return this.transactionId; } } } diff --git a/src/test/java/org/springframework/data/gemfire/transaction/GemfireTransactionManagerIntegrationTests.java b/src/test/java/org/springframework/data/gemfire/transaction/GemfireTransactionManagerIntegrationTests.java new file mode 100644 index 00000000..d13b8885 --- /dev/null +++ b/src/test/java/org/springframework/data/gemfire/transaction/GemfireTransactionManagerIntegrationTests.java @@ -0,0 +1,175 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed 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.springframework.data.gemfire.transaction; + +import static org.assertj.core.api.Assertions.assertThat; + +import javax.annotation.Resource; + +import org.apache.geode.cache.GemFireCache; +import org.apache.geode.cache.Region; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.data.gemfire.LocalRegionFactoryBean; +import org.springframework.data.gemfire.config.annotation.PeerCacheApplication; +import org.springframework.stereotype.Repository; +import org.springframework.stereotype.Service; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.transaction.annotation.EnableTransactionManagement; +import org.springframework.transaction.annotation.Propagation; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.util.Assert; + +/** + * Integration tests for the {@link GemfireTransactionManager}. + * + * @author John Blum + * @see org.junit.Test + * @see org.junit.runner.RunWith + * @see org.springframework.data.gemfire.transaction.GemfireTransactionManager + * @see org.springframework.test.context.ContextConfiguration + * @see org.springframework.test.context.junit4.SpringRunner + * @see org.springframework.transaction.annotation.EnableTransactionManagement + * @see org.springframework.transaction.annotation.Propagation + * @since 1.9.1 + */ +@RunWith(SpringRunner.class) +@ContextConfiguration(classes = + GemfireTransactionManagerIntegrationTests.GemfireTransactionManagerIntegrationTestsConfiguration.class) +public class GemfireTransactionManagerIntegrationTests { + + protected static final String GEMFIRE_LOG_LEVEL = "warning"; + + @Resource(name = "Example") + @SuppressWarnings("unused") + private Region example; + + @Autowired + @SuppressWarnings("all") + private SuspendAndResumeCacheTransactionsService service; + + @Test(expected = IllegalArgumentException.class) + public void suspendAndResumeIsSuccessful() { + try { + assertThat(example).isEmpty(); + + service.doCacheTransactions(); + } + catch (IllegalArgumentException e) { + assertThat(e).hasMessage("boom!"); + assertThat(e).hasNoCause(); + + throw e; + } + finally { + assertThat(example).hasSize(1); + assertThat(example.containsKey("tx-1-op-1")).isFalse(); + assertThat(example.containsKey("tx-2-op-1")).isTrue(); + } + } + + @Configuration + @EnableTransactionManagement + @Import(GemFireConfiguration.class) + @SuppressWarnings("unused") + static class GemfireTransactionManagerIntegrationTestsConfiguration { + + @Bean + GemfireTransactionManager transactionManager(GemFireCache gemfireCache) { + return new GemfireTransactionManager(gemfireCache); + } + + @Bean + SuspendAndResumeCacheTransactionsRepository suspendAndResumeCacheTransactionsRepository(GemFireCache gemFireCache) { + return new SuspendAndResumeCacheTransactionsRepository(gemFireCache.getRegion("Example")); + } + + @Bean + SuspendAndResumeCacheTransactionsService suspendAndResumeCacheTransactionsService( + SuspendAndResumeCacheTransactionsRepository repository) { + + return new SuspendAndResumeCacheTransactionsService(repository); + } + } + + @SuppressWarnings("unused") + @PeerCacheApplication(name = "GemfireTransactionManagerIntegrationTests", logLevel = GEMFIRE_LOG_LEVEL) + static class GemFireConfiguration { + + @Bean(name = "Example") + LocalRegionFactoryBean exampleRegion(GemFireCache gemfireCache) { + LocalRegionFactoryBean example = new LocalRegionFactoryBean(); + + example.setCache(gemfireCache); + example.setClose(false); + example.setPersistent(false); + + return example; + } + } + + @Service + static class SuspendAndResumeCacheTransactionsService { + + SuspendAndResumeCacheTransactionsRepository repository; + + SuspendAndResumeCacheTransactionsService(SuspendAndResumeCacheTransactionsRepository repository) { + Assert.notNull(repository, "Repository must not be null"); + this.repository = repository; + } + + @Transactional + public void doCacheTransactions() { + repository.doOperationOneInTransactionOne(); + repository.doOperationOneInTransactionTwo(); + repository.doOperationTwoInTransactionOne(); + } + } + + @Repository + static class SuspendAndResumeCacheTransactionsRepository { + + @SuppressWarnings("all") + private Region example; + + SuspendAndResumeCacheTransactionsRepository(Region example) { + Assert.notNull(example, "Region must not be null"); + this.example = example; + } + + @Transactional(propagation = Propagation.REQUIRED) + public void doOperationOneInTransactionOne() { + example.put("tx-1-op-1", "test"); + } + + @Transactional(propagation = Propagation.REQUIRES_NEW) + public void doOperationOneInTransactionTwo() { + example.put("tx-2-op-1", "test"); + } + + @Transactional(propagation = Propagation.REQUIRED) + public void doOperationTwoInTransactionOne() { + throw new IllegalArgumentException("boom!"); + } + } +} diff --git a/src/test/resources/org/springframework/data/gemfire/AutoRegionLookupWithComponentScanningIntegrationTests-context.xml b/src/test/resources/org/springframework/data/gemfire/AutoRegionLookupWithComponentScanningIntegrationTests-context.xml index e2a5e2c8..5fa415d2 100644 --- a/src/test/resources/org/springframework/data/gemfire/AutoRegionLookupWithComponentScanningIntegrationTests-context.xml +++ b/src/test/resources/org/springframework/data/gemfire/AutoRegionLookupWithComponentScanningIntegrationTests-context.xml @@ -17,6 +17,7 @@ +