#2134 - Modify ChangeLog to send the last set of changes Pre Commit (rather than post commit). Expose this for spring transactions.

This commit is contained in:
rob bygrave committed 2020-12-18 14:54:44 +13:00
1 parent e9081a176f
commit 1bdc6d8f7f
8 files changed
+71 -23

No files matched your search

@@ -319,6 +319,11 @@ public interface SpiTransaction extends Transaction {
*/
boolean isSkipCacheExplicit();
/**
* Fire pre commit processing/listeners.
*/
void preCommit();
/**
* Fire post commit events and listeners.
*/
@@ -418,6 +418,11 @@ public abstract class SpiTransactionProxy implements SpiTransaction {
transaction.flushBatchOnCollection();
}
@Override
public void preCommit() {
transaction.preCommit();
}
@Override
public void postCommit() {
transaction.postCommit();
@@ -586,6 +586,11 @@ class ImplicitReadOnlyTransaction implements SpiTransaction, TxnProfileEventCode
}
}
@Override
public void preCommit() {
// do nothing
}
@Override
public void postCommit() {
// do nothing
@@ -373,6 +373,9 @@ class JdbcTransaction implements SpiTransaction, TxnProfileEventCodes {
private void firePreCommit() {
withEachCallback(TransactionCallback::preCommit);
if (changeLogHolder != null) {
changeLogHolder.preCommit();
}
}
private void firePostCommit() {
@@ -981,9 +984,7 @@ class JdbcTransaction implements SpiTransaction, TxnProfileEventCodes {
* Batch flush, jdbc commit, trigger registered TransactionCallbacks, notify l2 cache etc.
*/
private void flushCommitAndNotify() throws SQLException {
internalBatchFlush();
firePreCommit();
// only performCommit can throw an exception
preCommit();
performCommit();
postCommit();
}
@@ -994,6 +995,12 @@ class JdbcTransaction implements SpiTransaction, TxnProfileEventCodes {
notifyCommit();
}
@Override
public void preCommit() {
internalBatchFlush();
firePreCommit();
}
/**
* Perform a commit, fire callbacks and notify l2 cache etc.
* <p>
@@ -190,7 +190,7 @@ public class JtaTransactionManager implements ExternalTransactionManager {
@Override
public void beforeCompletion() {
transaction.flush();
transaction.preCommit();
}
@Override
@@ -88,6 +88,11 @@ class NoTransaction implements SpiTransaction {
// do nothing
}
@Override
public void preCommit() {
// do nothing
}
@Override
public void postCommit() {
// do nothing
@@ -56,20 +56,31 @@ public class TChangeLogHolder {
* Add a bean change to the change set.
*/
public void addBeanChange(BeanChange change) {
changes.addBeanChange(change);
if (++count >= batchSize) {
// we hit the batch size so send what we have knowing
// that the transaction has not completed yet and
// reset the changes and count
owner.sendChangeLog(changes);
changes = new ChangeSet(transactionId, ++batchId);
count = 0;
sendChanges();
}
}
private void sendChanges() {
owner.sendChangeLog(changes);
changes = new ChangeSet(transactionId, ++batchId);
count = 0;
}
/**
* On post commit send the changes we have collected.
* Send the changes held prior to transaction commit.
*/
public void preCommit() {
sendChanges();
}
/**
* On post commit send the changes we have collected. This should be
* only the COMMITTED state and with all changes sent prior to commit.
*/
public void postCommit() {
changes.setTxnState(TxnState.COMMITTED);
@@ -15,11 +15,15 @@ import io.ebean.event.changelog.ChangeLogPrepare;
import io.ebean.event.changelog.ChangeLogRegister;
import io.ebean.event.changelog.ChangeSet;
import io.ebean.event.changelog.ChangeType;
import io.ebean.event.changelog.TxnState;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.tests.model.basic.EBasicChangeLog;
import java.util.ArrayList;
import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
@@ -43,6 +47,13 @@ public class TestChangeLog extends BaseTestCase {
server.shutdown();
}
private BeanChange firstChange() {
ChangeSet changeSet0 = changeLogListener.changes.get(0);
BeanChange change = changeSet0.getChanges().get(0);
changeLogListener.changes.clear();
return change;
}
@Test
public void test() {
@@ -50,8 +61,13 @@ public class TestChangeLog extends BaseTestCase {
bean.setName("logBean");
bean.setShortDescription("hello");
server.save(bean);
BeanChange change = changeLogListener.changes.getChanges().get(0);
final List<ChangeSet> allChanges = changeLogListener.changes;
assertThat(allChanges).hasSize(2);
assertThat(allChanges.get(0).getTxnState()).isEqualTo(TxnState.IN_PROGRESS);
assertThat(allChanges.get(1).getTxnState()).isEqualTo(TxnState.COMMITTED);
BeanChange change = firstChange();
assertThat(change.getEvent()).isEqualTo(ChangeType.INSERT);
assertThat(change.getData())
.contains("\"name\":\"logBean\"")
@@ -61,8 +77,7 @@ public class TestChangeLog extends BaseTestCase {
bean.setName("ChangedName");
server.save(bean);
change = changeLogListener.changes.getChanges().get(0);
change = firstChange();
assertThat(change.getEvent()).isEqualTo(ChangeType.UPDATE);
assertThat(change.getOldData()).contains("\"name\":\"logBean\"");
assertThat(change.getData()).contains("\"name\":\"ChangedName\"");
@@ -70,14 +85,11 @@ public class TestChangeLog extends BaseTestCase {
server.delete(bean);
change = changeLogListener.changes.getChanges().get(0);
change = firstChange();
assertThat(change.getEvent()).isEqualTo(ChangeType.DELETE);
assertThat(change.getData()).isNull();
}
@Test
public void testWithNull() {
@@ -85,7 +97,7 @@ public class TestChangeLog extends BaseTestCase {
bean.setName(null);
bean.setShortDescription("hello");
server.save(bean);
BeanChange change = changeLogListener.changes.getChanges().get(0);
BeanChange change = firstChange();
assertThat(change.getEvent()).isEqualTo(ChangeType.INSERT);
assertThat(change.getData())
@@ -99,7 +111,7 @@ public class TestChangeLog extends BaseTestCase {
bean.setShortDescription("hello");
server.save(bean);
change = changeLogListener.changes.getChanges().get(0);
change = firstChange();
assertThat(change.getEvent()).isEqualTo(ChangeType.UPDATE);
assertThat(change.getOldData())
@@ -113,11 +125,10 @@ public class TestChangeLog extends BaseTestCase {
server.delete(bean);
change = changeLogListener.changes.getChanges().get(0);
change = firstChange();
assertThat(change.getEvent()).isEqualTo(ChangeType.DELETE);
assertThat(change.getData()).isNull();
}
private Database createServer() {
@@ -153,16 +164,15 @@ public class TestChangeLog extends BaseTestCase {
ObjectMapper objectMapper = new ObjectMapper();
ChangeSet changes;
List<ChangeSet> changes = new ArrayList<>();
public TDChangeLogListener() {
objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
}
@Override
public void log(ChangeSet changes) {
this.changes = changes;
this.changes.add(changes);
try {
String json = objectMapper.writeValueAsString(changes);
ChangeSet changes1 = objectMapper.readValue(json, ChangeSet.class);