SGF-598 - Add support for suspend and resume in GemfireTransactionManager.
This commit is contained in:
John Blum
@@ -16,6 +16,16 @@
|
||||
|
||||
package org.springframework.data.gemfire;
|
||||
|
||||
import static org.springframework.data.gemfire.GemfireTransactionManager.CacheHolder.newCacheHolder;
|
||||
import static org.springframework.data.gemfire.GemfireTransactionManager.CacheTransactionObject.newCacheTransactionObject;
|
||||
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import com.gemstone.gemfire.cache.CacheTransactionManager;
|
||||
import com.gemstone.gemfire.cache.GemFireCache;
|
||||
import com.gemstone.gemfire.cache.Region;
|
||||
import com.gemstone.gemfire.cache.TransactionId;
|
||||
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.transaction.CannotCreateTransactionException;
|
||||
import org.springframework.transaction.NoTransactionException;
|
||||
@@ -28,257 +38,381 @@ import org.springframework.transaction.support.ResourceTransactionManager;
|
||||
import org.springframework.transaction.support.TransactionSynchronizationManager;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
import com.gemstone.gemfire.cache.Cache;
|
||||
import com.gemstone.gemfire.cache.CacheTransactionManager;
|
||||
import com.gemstone.gemfire.cache.Region;
|
||||
|
||||
/**
|
||||
* Local transaction manager for GemFire Enterprise Fabric (GEF). Provides a
|
||||
* {@link PlatformTransactionManager} implementation for a single 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.
|
||||
*
|
||||
* Local Transaction Management for Pivotal GemFire. Provides a Spring {@link PlatformTransactionManager} implementation
|
||||
* for the Pivotal GemFire {@link CacheTransactionManager}.
|
||||
*
|
||||
* 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.
|
||||
*
|
||||
* <p>
|
||||
* 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 <i>not</i> 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 <i>not</i> able to provide XA transactions, for example to share transactions
|
||||
* with data access.
|
||||
*
|
||||
* <p>
|
||||
* To prevent dirty reads, by default, the cache is configured to return copies
|
||||
* rather then direct references for <code>get</code> 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 <code>get</code> 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 com.gemstone.gemfire.cache.GemFireCache#setCopyOnRead(boolean)
|
||||
* @see com.gemstone.gemfire.cache.CacheTransactionManager
|
||||
* @see com.gemstone.gemfire.cache.Cache#setCopyOnRead(boolean)
|
||||
* @see com.gemstone.gemfire.cache.Region#get(Object)
|
||||
* @see com.gemstone.gemfire.CopyHelper#copy(Object)
|
||||
* @see #setCopyOnRead(boolean)
|
||||
* @see org.springframework.transaction.support.AbstractPlatformTransactionManager
|
||||
*
|
||||
* @author Costin Leau
|
||||
* @see #setCopyOnRead(boolean)
|
||||
*/
|
||||
// TODO add lenient behavior if a transaction is already started on the current
|
||||
// thread (what should happen then)
|
||||
@SuppressWarnings("serial")
|
||||
public class GemfireTransactionManager extends AbstractPlatformTransactionManager implements InitializingBean,
|
||||
ResourceTransactionManager {
|
||||
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.
|
||||
*
|
||||
* @param cache a reference to the GemFire Cache associated with Cache transactions.
|
||||
* Constructs an instance of the {@link GemfireTransactionManager} initialized with
|
||||
* the given {@link GemFireCache} reference.
|
||||
*
|
||||
* @param cache reference to the {@link GemFireCache} associated with cache transactions.
|
||||
* @see com.gemstone.gemfire.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 = null;
|
||||
|
||||
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();
|
||||
if (status.isDebug()) {
|
||||
logger.debug("Committing local cache transaction");
|
||||
}
|
||||
|
||||
getCacheTransactionManager().commit();
|
||||
}
|
||||
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);
|
||||
}
|
||||
|
||||
catch (com.gemstone.gemfire.cache.TransactionException ex) {
|
||||
throw new GemfireTransactionCommitException("Unexpected failure on commit of Cache local transaction", ex);
|
||||
catch (com.gemstone.gemfire.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 com.gemstone.gemfire.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 <K> the Region key class type.
|
||||
* @param <V> 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 com.gemstone.gemfire.cache.GemFireCache
|
||||
*/
|
||||
public <K, V> void setRegion(Region<K, V> 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 com.gemstone.gemfire.CopyHelper#copy(Object)} in which
|
||||
* case this setting can be set to <code>false</code>. However, unless there
|
||||
* is a measurable performance penalty, the recommendation is to keep this
|
||||
* setting to <code>true</code>
|
||||
*
|
||||
* @param copyOnRead whether copies (default) rather then direct references
|
||||
* will be returned on fetch operations
|
||||
* One could explicitly deep copy objects before making changes (for example by using
|
||||
* {@link com.gemstone.gemfire.CopyHelper#copy(Object)} in which case this setting
|
||||
* can be set to <code>false</code>
|
||||
*
|
||||
* However, unless there is a measurable performance penalty, the recommendation is
|
||||
* to keep this setting to <code>true</code>.
|
||||
*
|
||||
* @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 <K> {@link Class} type of the {@link Region} key.
|
||||
* @param <V> {@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 com.gemstone.gemfire.cache.Region
|
||||
*/
|
||||
public <K, V> void setRegion(Region<K, V> 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;
|
||||
public static CacheTransactionObject newCacheTransactionObject(CacheHolder cacheHolder) {
|
||||
CacheTransactionObject transactionObject = new CacheTransactionObject();
|
||||
transactionObject.setHolder(cacheHolder);
|
||||
return transactionObject;
|
||||
}
|
||||
|
||||
public void setHolder(CacheHolder cacheHolder) {
|
||||
this.cacheHolder = cacheHolder;
|
||||
public boolean isHolding() {
|
||||
return (getHolder() != null);
|
||||
}
|
||||
|
||||
public CacheHolder getHolder() {
|
||||
return this.cacheHolder;
|
||||
}
|
||||
|
||||
public void setHolder(CacheHolder holder) {
|
||||
this.cacheHolder = holder;
|
||||
}
|
||||
|
||||
public CacheHolder setAndGetHolder(CacheHolder holder) {
|
||||
setHolder(holder);
|
||||
return getHolder();
|
||||
}
|
||||
|
||||
public 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")
|
||||
private TransactionId transactionId;
|
||||
|
||||
public static CacheHolder newCacheHolder(TransactionId transactionId) {
|
||||
CacheHolder cacheHolder = new CacheHolder();
|
||||
cacheHolder.transactionId = transactionId;
|
||||
return cacheHolder;
|
||||
}
|
||||
|
||||
public boolean isRollbackOnly() {
|
||||
return rollbackOnly;
|
||||
return this.rollbackOnly;
|
||||
}
|
||||
|
||||
public void setRollbackOnly() {
|
||||
rollbackOnly = true;
|
||||
this.rollbackOnly = true;
|
||||
}
|
||||
|
||||
public TransactionId getTransactionId() {
|
||||
return this.transactionId;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
|
||||
import com.gemstone.gemfire.cache.GemFireCache;
|
||||
import com.gemstone.gemfire.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.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.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<Object, Object> 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<Object, Object> exampleRegion(GemFireCache gemfireCache) {
|
||||
LocalRegionFactoryBean<Object, Object> example = new LocalRegionFactoryBean<Object, Object>();
|
||||
|
||||
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<Object, Object> example;
|
||||
|
||||
SuspendAndResumeCacheTransactionsRepository(Region<Object, Object> 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!");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -17,6 +17,7 @@
|
||||
<context:exclude-filter type="annotation" expression="org.springframework.context.annotation.Configuration"/>
|
||||
<context:exclude-filter type="annotation" expression="org.springframework.stereotype.Service"/>
|
||||
<context:exclude-filter type="assignable" expression="org.springframework.data.gemfire.AutoRegionLookupWithAutowiringIntegrationTests$TestComponent"/>
|
||||
<context:exclude-filter type="assignable" expression="org.springframework.data.gemfire.GemfireTransactionManagerIntegrationTests$SuspendAndResumeCacheTransactionsRepository"/>
|
||||
<context:exclude-filter type="regex" expression="org.springframework.data.gemfire.*.sample.*"/>
|
||||
</context:component-scan>
|
||||
|
||||
|
||||
Reference in New Issue
Block a user