Skip to content

Datacouch 1145 transaction support #1448

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 16 commits into from
May 31, 2022
Merged

Datacouch 1145 transaction support #1448

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
0394222
Move CouchbaseTransactionalOperator to use SLF4J, same as
programmatix May 27, 2022
1311125
Handle all propagation levels
programmatix May 27, 2022
56201d1
Adding new tests for repository calls inside @Transactional
programmatix May 27, 2022
673a00d
Rename CouchbaseTransactionalIntegrationTests, and check
programmatix May 27, 2022
1be459f
Remove unnecessary methods From ReactiveCouchbaseClientFactory.
mikereiche May 27, 2022
7034e74
Cleanup of test classes.
mikereiche May 27, 2022
76191cb
Removing unused classes
programmatix May 30, 2022
7709c0b
Removing version from CouchbaseDocument
programmatix May 30, 2022
ee9f837
Adding and removing TODOs
programmatix May 30, 2022
8b20018
Adding and removing TODOs
programmatix May 30, 2022
68dcca3
DRYing CouchbaseTransactionalPropagationIntegrationTests
programmatix May 30, 2022
e8ab027
Check propagation tests retry as expected
programmatix May 30, 2022
ebb1cc7
Tidy up PersonWithoutVersion to the minimum required
programmatix May 30, 2022
6ed3bad
Removing unused code
programmatix May 30, 2022
4344bb0
Adding tests for @Transactional removeByQuery and findByQuery
programmatix May 30, 2022
91cd6c0
Merge branch 'programmatix-datacouch_1145_transaction_support' into d…
mikereiche May 31, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
77 changes: 1 addition & 76 deletions src/main/java/com/couchbase/client/java/transactions/AttemptContextReactiveAccessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,22 +66,6 @@ public static ReactiveTransactionAttemptContext reactive(TransactionAttemptConte
return new ReactiveTransactionAttemptContext(getCore(atr), serializer);
}

public static TransactionAttemptContext blocking(ReactiveTransactionAttemptContext atr) {
JsonSerializer serializer;
try {
Field field = ReactiveTransactionAttemptContext.class.getDeclaredField("serializer");
field.setAccessible(true);
serializer = (JsonSerializer) field.get(atr);
} catch (Throwable err) {
throw new RuntimeException(err);
}
return new TransactionAttemptContext(getCore(atr), serializer);
}

public static CoreTransactionLogger getLogger(ReactiveTransactionAttemptContext attemptContextReactive) {
return attemptContextReactive.logger();
}

public static CoreTransactionLogger getLogger(TransactionAttemptContext attemptContextReactive) {
return attemptContextReactive.logger();
}
Expand All @@ -100,6 +84,7 @@ public static CoreTransactionAttemptContext newCoreTranactionAttemptContext(Reac
throw new RuntimeException(err);
}

// todo gpx options need to be loaded from Cluster and TransactionOptions (from TransactionsWrapper)
CoreTransactionOptions perConfig = new CoreTransactionOptions(Optional.empty(), Optional.empty(), Optional.empty(),
Optional.of(Duration.ofMinutes(10)), Optional.empty(), Optional.empty());

Expand All @@ -115,16 +100,6 @@ public static CoreTransactionAttemptContext newCoreTranactionAttemptContext(Reac
return coreTransactionAttemptContext;
}

private static Duration now() {
return Duration.of(System.nanoTime(), ChronoUnit.NANOS);
}

public static ReactiveTransactionAttemptContext from(CoreTransactionAttemptContext coreTransactionAttemptContext,
JsonSerializer serializer) {
TransactionAttemptContext tac = new TransactionAttemptContext(coreTransactionAttemptContext, serializer);
return reactive(tac);
}

public static CoreTransactionAttemptContext getCore(ReactiveTransactionAttemptContext atr) {
CoreTransactionAttemptContext coreTransactionsReactive;
try {
Expand All @@ -147,61 +122,11 @@ public static CoreTransactionAttemptContext getCore(TransactionAttemptContext at
}
}

public static Mono<Void> implicitCommit(ReactiveTransactionAttemptContext atr, boolean b) {
CoreTransactionAttemptContext coreTransactionsReactive = getCore(atr);
try {
// getDeclaredMethod() does not find it (because of primitive arg?)
// CoreTransactionAttemptContext.class.getDeclaredMethod("implicitCommit", Boolean.class);
Method[] methods = CoreTransactionAttemptContext.class.getDeclaredMethods();
Method method = null;
for (Method m : methods) {
if (m.getName().equals("implicitCommit")) {
method = m;
break;
}
}
if (method == null) {
throw new RuntimeException("did not find implicitCommit method");
}
method.setAccessible(true);
return (Mono<Void>) method.invoke(coreTransactionsReactive, b);
} catch (Throwable err) {
throw new RuntimeException(err);
}

}

public static AttemptState getState(ReactiveTransactionAttemptContext atr) {
CoreTransactionAttemptContext coreTransactionsReactive = getCore(atr);
try {
Field field = CoreTransactionAttemptContext.class.getDeclaredField("state");
field.setAccessible(true);
return (AttemptState) field.get(coreTransactionsReactive);
} catch (Throwable err) {
throw new RuntimeException(err);
}
}

public static ReactiveTransactionAttemptContext createReactiveTransactionAttemptContext(
CoreTransactionAttemptContext core, JsonSerializer jsonSerializer) {
return new ReactiveTransactionAttemptContext(core, jsonSerializer);
}

public static CoreTransactionsReactive getCoreTransactionsReactive(ReactiveTransactions transactions) {
try {
Field field = ReactiveTransactions.class.getDeclaredField("internal");
field.setAccessible(true);
return (CoreTransactionsReactive) field.get(transactions);
} catch (Throwable err) {
throw new RuntimeException(err);
}
}

public static TransactionAttemptContext newTransactionAttemptContext(CoreTransactionAttemptContext ctx,
JsonSerializer jsonSerializer) {
return new TransactionAttemptContext(ctx, jsonSerializer);
}

public static TransactionResult run(Transactions transactions, Consumer<TransactionAttemptContext> transactionLogic, CoreTransactionOptions coreTransactionOptions) {
return reactive(transactions).runBlocking(transactionLogic, coreTransactionOptions);
}
Expand Down
Loading