Spring/spring-data-geode
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAGEODE-152 - Add tests for Spring @TransactionalEventListener annotated POJO methods in the context of Apache Geode, Local Cache Transactions.

Browse Source 
Resolves gh-27.
This commit is contained in:
John Blum
2019-11-11 19:50:58 -08:00
parent 6d5e1e2f1e
commit ad4a683afc
3 changed files with 415 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

103
src/main/java/org/springframework/data/gemfire/transaction/GemfireTransactionManager.java
View File

@@ -12,9 +12,7 @@
* 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.springframework.data.gemfire.transaction.GemfireTransactionManager.CacheHolder.newCacheHolder;
@@ -90,8 +88,7 @@ public class GemfireTransactionManager extends AbstractPlatformTransactionManage
/**
* Constructs an instance of the {@link GemfireTransactionManager}.
*/
public GemfireTransactionManager() {
}
public GemfireTransactionManager() { }
/**
* Constructs an instance of the {@link GemfireTransactionManager} initialized with
@@ -102,7 +99,9 @@ public class GemfireTransactionManager extends AbstractPlatformTransactionManage
* @see #afterPropertiesSet()
*/
public GemfireTransactionManager(GemFireCache cache) {
this.cache = cache;
afterPropertiesSet();
}
@@ -111,7 +110,9 @@ public class GemfireTransactionManager extends AbstractPlatformTransactionManage
*/
@Override
public void afterPropertiesSet() {
Assert.notNull(this.cache, "Cache is required");
this.cache.setCopyOnRead(isCopyOnRead());
}
@@ -137,8 +138,11 @@ public class GemfireTransactionManager extends AbstractPlatformTransactionManage
*/
@Override
protected void doBegin(Object transaction, TransactionDefinition definition) throws TransactionException {
try {
CacheTransactionObject cacheTransaction = (CacheTransactionObject) transaction;
GemFireCache cache = getCache();
if (logger.isDebugEnabled()) {
@@ -157,10 +161,10 @@ public class GemfireTransactionManager extends AbstractPlatformTransactionManage
cacheTransaction.setAndGetHolder(newCacheHolder(transactionId)));
}
}
catch (IllegalStateException e) {
catch (Exception cause) {
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);
"An existing, ongoing transaction is already associated with the current thread.",
" Are multiple transaction managers present?"), cause);
}
}
@@ -169,6 +173,7 @@ public class GemfireTransactionManager extends AbstractPlatformTransactionManage
*/
@Override
protected void doCommit(DefaultTransactionStatus status) throws TransactionException {
try {
if (status.isDebug()) {
logger.debug("Committing local cache transaction");
@@ -176,13 +181,13 @@ public class GemfireTransactionManager extends AbstractPlatformTransactionManage
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) {
catch (org.apache.geode.cache.TransactionException cause) {
throw new GemfireTransactionCommitException(
"Unexpected failure occurred on commit of local cache transaction", e);
"Unexpected failure occurred on commit of local cache transaction", cause);
}
catch (Exception cause) {
throw new NoTransactionException(
"No transaction is associated with the current thread; are multiple transaction managers present?", cause);
}
}
@@ -191,6 +196,7 @@ public class GemfireTransactionManager extends AbstractPlatformTransactionManage
*/
@Override
protected Object doSuspend(Object transaction) throws TransactionException {
if (getCacheTransactionManager().suspend() != null) {
TransactionSynchronizationManager.unbindResource(getCache());
return ((CacheTransactionObject) transaction).setAndGetExistingHolder(null);
@@ -204,13 +210,14 @@ public class GemfireTransactionManager extends AbstractPlatformTransactionManage
*/
@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()));
boolean resumeSuccessful = isResumeWaitTimeSet()
? getCacheTransactionManager().tryResume(holder.getTransactionId(), getResumeWaitTime(), getResumeWaitTimeUnit())
: getCacheTransactionManager().tryResume(holder.getTransactionId());
if (resumeSuccessful) {
TransactionSynchronizationManager.bindResource(getCache(),
@@ -224,6 +231,7 @@ public class GemfireTransactionManager extends AbstractPlatformTransactionManage
*/
@Override
protected void doRollback(DefaultTransactionStatus status) throws TransactionException {
try {
if (status.isDebug()) {
logger.debug("Rolling back local cache transaction");
@@ -231,9 +239,12 @@ public class GemfireTransactionManager extends AbstractPlatformTransactionManage
getCacheTransactionManager().rollback();
}
catch (IllegalStateException e) {
throw new NoTransactionException(
"No transaction is associated with the current thread; are multiple transaction managers present?", e);
catch (Exception cause) {
String exceptionMessage =
"No transaction is associated with the current thread. Are multiple transaction managers present?";
throw new NoTransactionException(exceptionMessage, cause);
}
}
@@ -312,7 +323,7 @@ public class GemfireTransactionManager extends AbstractPlatformTransactionManage
* @see #setCopyOnRead(boolean)
*/
public boolean isCopyOnRead() {
return copyOnRead;
return this.copyOnRead;
}
/**
@@ -325,7 +336,9 @@ public class GemfireTransactionManager extends AbstractPlatformTransactionManage
* @see org.apache.geode.cache.Region
*/
public <K, V> void setRegion(Region<K, V> region) {
Assert.notNull(region, "Region must not be null");
this.cache = (GemFireCache) region.getRegionService();
}
@@ -366,8 +379,10 @@ public class GemfireTransactionManager extends AbstractPlatformTransactionManage
* @see #getResumeWaitTime()
*/
protected boolean isResumeWaitTimeSet() {
Long resumeWaitTime = getResumeWaitTime();
return (resumeWaitTime != null && resumeWaitTime > 0);
return resumeWaitTime != null && resumeWaitTime > 0;
}
/**
@@ -393,7 +408,7 @@ public class GemfireTransactionManager extends AbstractPlatformTransactionManage
}
/**
* GemFire local transaction object.
* GemFire local cache transaction object.
*
* @author Costin Leau
* @author John Blum
@@ -402,40 +417,34 @@ public class GemfireTransactionManager extends AbstractPlatformTransactionManage
private CacheHolder cacheHolder;
/* (non-Javadoc) */
static CacheTransactionObject newCacheTransactionObject(CacheHolder cacheHolder) {
CacheTransactionObject transactionObject = new CacheTransactionObject();
transactionObject.setHolder(cacheHolder);
return transactionObject;
}
/* (non-Javadoc) */
boolean isHolding() {
return (getHolder() != null);
}
/* (non-Javadoc) */
CacheHolder getHolder() {
return this.cacheHolder;
}
/* (non-Javadoc) */
void setHolder(CacheHolder holder) {
this.cacheHolder = holder;
}
/* (non-Javadoc) */
CacheHolder setAndGetExistingHolder(CacheHolder cacheHolder) {
CacheHolder existingHolder = getHolder();
setHolder(cacheHolder);
return existingHolder;
}
/* (non-Javadoc) */
CacheHolder setAndGetHolder(CacheHolder holder) {
setHolder(holder);
return getHolder();
}
void setHolder(CacheHolder cacheHolder) {
this.cacheHolder = cacheHolder;
}
CacheHolder getHolder() {
return this.cacheHolder;
}
boolean isHolding() {
return getHolder() != null;
}
}
/**
@@ -447,24 +456,20 @@ public class GemfireTransactionManager extends AbstractPlatformTransactionManage
private TransactionId transactionId;
/* (non-Javadoc) */
static CacheHolder newCacheHolder(TransactionId transactionId) {
CacheHolder cacheHolder = new CacheHolder();
cacheHolder.transactionId = transactionId;
return cacheHolder;
}
/* (non-Javadoc) */
boolean isRollbackOnly() {
return this.rollbackOnly;
}
/* (non-Javadoc) */
void setRollbackOnly() {
this.rollbackOnly = true;
}
/* (non-Javadoc) */
boolean isRollbackOnly() {
return this.rollbackOnly;
}
TransactionId getTransactionId() {
return this.transactionId;
}

76
src/main/java/org/springframework/data/gemfire/transaction/TransactionApplicationEvent.java Normal file
View File

@@ -0,0 +1,76 @@
/*
* Copyright 2019 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
*
* https://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 java.time.Instant;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Optional;
import org.springframework.context.ApplicationEvent;
import org.springframework.util.StringUtils;
/**
* The {@link TransactionApplicationEvent} is an implementation of {@link ApplicationEvent} which is fired during
* a transaction.
*
* @author John Blum
* @see java.time.Instant
* @see java.time.LocalDateTime
* @see org.springframework.context.ApplicationEvent
* @since 2.3.0
*/
public class TransactionApplicationEvent extends ApplicationEvent {
protected static final String TIMESTAMP_PATTERN = "yyyy-MM-dd-hh:mm:ss.S";
private String details;
public TransactionApplicationEvent(Object source) {
this(source, null);
}
public TransactionApplicationEvent(Object source, String details) {
super(source);
this.details = details;
}
public Optional<String> getDetails() {
return Optional.ofNullable(this.details).filter(StringUtils::hasText);
}
public LocalDateTime getTimestampAsLocalDateTime() {
return LocalDateTime.from(Instant.ofEpochMilli(getTimestamp()));
}
public String getTimestampAsString() {
return getTimestampAsString(TIMESTAMP_PATTERN);
}
public String getTimestampAsString(String pattern) {
return getTimestampAsLocalDateTime().format(DateTimeFormatter.ofPattern(pattern));
}
@Override
public String toString() {
return getDetails()
.map(details -> String.format("%s - %s", getTimestampAsString(), details))
.orElse(String.format("%s[%s]", getClass().getSimpleName(), getTimestampAsString()));
}
}