Merge pull request 'fix(data): fire transaction synchronizations, and scope them to their transaction' (#9) from fix/tx-synchronizations into master
Publish Maven packages / publish (push) Successful in 1m53s
Publish Maven packages / publish (push) Successful in 1m53s
Reviewed-on: #9
This commit was merged in pull request #9.
This commit is contained in:
@@ -1,53 +1,93 @@
|
|||||||
# flash-ext-data-core
|
# flash-ext-data-core
|
||||||
|
|
||||||
Core comune per il layer dati di Flash.
|
Shared core for Flash's data layer.
|
||||||
|
|
||||||
## Scopo
|
## Purpose
|
||||||
|
|
||||||
Questo modulo definisce il contratto transazionale condiviso tra le implementazioni backend.
|
This module defines the transactional contract shared across backend implementations. It does not
|
||||||
Non parla con Hibernate o JDBC direttamente: espone solo astrazioni e un runtime minimale.
|
talk to Hibernate or JDBC directly: it exposes abstractions and a minimal runtime, nothing else.
|
||||||
|
|
||||||
## Componenti
|
## Components
|
||||||
|
|
||||||
- `TxDefinition`: metadata immutabile della transazione.
|
- `TxDefinition`: immutable transaction metadata.
|
||||||
- `TxStatus`: stato runtime restituito dal manager.
|
- `TxStatus`: runtime state returned by the manager.
|
||||||
- `TxManager`: contratto per `begin`, `commit`, `rollback`.
|
- `TxManager`: the `begin`/`commit`/`rollback` contract.
|
||||||
- `Tx`: orchestration runtime e stack transazionale per thread.
|
- `Tx`: runtime orchestration and the per-thread transaction stack.
|
||||||
- `ResourceRegistry`: storage thread-local di risorse e synchronizations.
|
- `ResourceRegistry`: thread-local storage for resources and synchronizations.
|
||||||
- `Repository<T, ID>`: base repository auto-transazionale.
|
- `Repository<T, ID>`: self-transactional base repository.
|
||||||
- `Spec<T>`: predicato componibile.
|
- `Spec<T>`: composable predicate.
|
||||||
- `Query<T>`: oggetto query con spec, sort e paging.
|
- `Query<T>`: query object carrying spec, sort and paging.
|
||||||
- `SpecBuilder<T>`: DSL fluente per costruire spec tipizzate.
|
- `SpecBuilder<T>`: fluent DSL for building typed specs.
|
||||||
- `RepositorySupport<T, ID>`: helper interno condiviso.
|
- `RepositorySupport<T, ID>`: shared internal helper.
|
||||||
- `TransactionPropagation`: semantica di propagazione.
|
- `TransactionPropagation`: propagation semantics.
|
||||||
- `TransactionIsolation`: livello di isolamento.
|
- `TransactionIsolation`: isolation level.
|
||||||
- `TxSynchronization`: hook lifecycle.
|
- `TxSynchronization`: lifecycle hooks (see below).
|
||||||
|
|
||||||
## Modello di esecuzione
|
## Execution model
|
||||||
|
|
||||||
Il flusso è:
|
The flow is:
|
||||||
|
|
||||||
1. `Tx.call(definition, work)` chiama `TxManager.begin(definition)`.
|
1. `Tx.call(definition, work)` calls `TxManager.begin(definition)`.
|
||||||
2. Il `TxManager` crea un `TxStatus` backend-specific.
|
2. The `TxManager` creates a backend-specific `TxStatus`.
|
||||||
3. Lo status viene pushato nello stack thread-local.
|
3. The status is pushed onto the thread-local stack.
|
||||||
4. Il lavoro usa `Tx.resource(Class)` per ottenere la risorsa corrente.
|
4. The work uses `Tx.resource(Class)` to obtain the current resource.
|
||||||
5. A fine lavoro `Tx` decide tra `commit` e `rollback`.
|
5. When the work ends, `Tx` chooses between `commit` and `rollback`.
|
||||||
6. Lo stack viene poppato e il thread-local viene pulito se vuoto.
|
6. The stack is popped, and the thread-local is cleared once it is empty.
|
||||||
|
|
||||||
## Propagation supportata
|
## Supported propagation
|
||||||
|
|
||||||
- `REQUIRED`: usa la tx attiva oppure ne apre una nuova.
|
- `REQUIRED`: use the active transaction, or open a new one.
|
||||||
- `REQUIRES_NEW`: sospende la tx corrente e apre una nuova tx.
|
- `REQUIRES_NEW`: suspend the current transaction and open a new one.
|
||||||
- `SUPPORTS`: se esiste una tx attiva si aggancia, altrimenti esegue senza tx.
|
- `SUPPORTS`: join the active transaction if there is one, otherwise run without a transaction.
|
||||||
- `NOT_SUPPORTED`: sospende la tx corrente ed esegue senza tx.
|
- `NOT_SUPPORTED`: suspend the current transaction and run without one.
|
||||||
- `MANDATORY`: richiede una tx attiva.
|
- `MANDATORY`: require an active transaction.
|
||||||
|
|
||||||
## Uso di `Repository`
|
## Synchronizations (`TxSynchronization`)
|
||||||
|
|
||||||
`Repository` è la base comune per le repository concrete.
|
Lifecycle hooks for **one** transaction, registered through `Data.afterCommit(...)` (or directly
|
||||||
Ogni operazione pubblica usa internamente una tx `REQUIRED` o `REQUIRED` read-only.
|
with `ResourceRegistry.addSynchronization(...)`).
|
||||||
|
|
||||||
Le sottoclassi implementano i metodi `doXxx(...)` del nuovo modello:
|
Every callback belongs to exactly the innermost transaction active at registration time, and fires
|
||||||
|
exactly once, when *that* transaction completes:
|
||||||
|
|
||||||
|
- a **joined** inner transaction (`REQUIRED`) is not a transaction of its own, so callbacks
|
||||||
|
registered inside one wait for the outermost commit;
|
||||||
|
- a `REQUIRES_NEW` transaction is, so completing it fires only its own callbacks and leaves the
|
||||||
|
suspended outer transaction's pending.
|
||||||
|
|
||||||
|
### Which side of the commit each hook sits on
|
||||||
|
|
||||||
|
| hook | when | resource |
|
||||||
|
| --- | --- | --- |
|
||||||
|
| `beforeCommit(readOnly)` | immediately **before** the real commit | session/connection still **bound**, transaction still active |
|
||||||
|
| `afterCommit()` / `afterRollback()` | after completion | resource already **unbound** |
|
||||||
|
| `afterCompletion(outcome)` | after the two above | resource already unbound |
|
||||||
|
|
||||||
|
`beforeCommit` is the only hook that can still write through the same resource and have the write
|
||||||
|
land in the same atomic unit: flush a buffer, stamp an audit row, materialize a derived value. It
|
||||||
|
is skipped when the transaction is already `rollback-only`, since there is no commit to precede.
|
||||||
|
|
||||||
|
Post-completion callbacks run with the resource unbound instead: one that opens its own transaction
|
||||||
|
gets a **fresh** one rather than joining the transaction that just finished. That is what makes
|
||||||
|
them the right place to refresh a cache, enqueue a message, or notify anything outside the
|
||||||
|
database.
|
||||||
|
|
||||||
|
### Failure
|
||||||
|
|
||||||
|
Throwing from `beforeCommit` **vetoes the commit**: the transaction is rolled back,
|
||||||
|
`afterRollback`/`afterCompletion(ROLLED_BACK)` fire, and the exception reaches the caller. That is
|
||||||
|
the reason the hook runs before the commit rather than after — it can still refuse.
|
||||||
|
|
||||||
|
The post-completion hooks have no such power: the transaction is already over by the time they run,
|
||||||
|
so an exception propagates but changes nothing already committed, and stops the callbacks queued
|
||||||
|
behind it.
|
||||||
|
|
||||||
|
## Using `Repository`
|
||||||
|
|
||||||
|
`Repository` is the shared base for concrete repositories. Every public operation internally uses a
|
||||||
|
`REQUIRED` transaction, read-only where applicable.
|
||||||
|
|
||||||
|
Subclasses implement the `doXxx(...)` methods:
|
||||||
|
|
||||||
- `doFind(Query<T>)`
|
- `doFind(Query<T>)`
|
||||||
- `doFindOne(Spec<T>)`
|
- `doFindOne(Spec<T>)`
|
||||||
@@ -55,7 +95,8 @@ Le sottoclassi implementano i metodi `doXxx(...)` del nuovo modello:
|
|||||||
- `doDeleteAll(Spec<T>)`
|
- `doDeleteAll(Spec<T>)`
|
||||||
- `doUpdateAll(Spec<T>, T)`
|
- `doUpdateAll(Spec<T>, T)`
|
||||||
|
|
||||||
I vecchi overload di `findAll(...)` e `findPage(...)` sono stati ridotti a una combinazione di `Query<T>` e `Spec<T>`.
|
The old `findAll(...)` and `findPage(...)` overloads were reduced to a combination of `Query<T>` and
|
||||||
|
`Spec<T>`.
|
||||||
|
|
||||||
```java
|
```java
|
||||||
public abstract class Repository<T, ID> {
|
public abstract class Repository<T, ID> {
|
||||||
@@ -64,18 +105,21 @@ public abstract class Repository<T, ID> {
|
|||||||
}
|
}
|
||||||
```
|
```
|
||||||
|
|
||||||
## Composizione con Flash
|
## Composing with Flash
|
||||||
|
|
||||||
`DataExtension` registra:
|
`DataExtension` registers:
|
||||||
|
|
||||||
- `Tx` nel `FlashContext`
|
- `Tx` in the `FlashContext`
|
||||||
- `TxManager` nel `FlashContext`
|
- `TxManager` in the `FlashContext`
|
||||||
- un annotation processor per `@Transactional`
|
- an annotation processor for `@Transactional`
|
||||||
|
|
||||||
Questo rende il layer dati componibile con il sistema di extension di Flash senza stato globale.
|
This makes the data layer composable with Flash's extension system without global state.
|
||||||
|
|
||||||
## Note implementative
|
## Implementation notes
|
||||||
|
|
||||||
- Lo stack transazionale è thread-local e viene ripulito quando torna vuoto.
|
- The transaction stack is thread-local and is cleared once it becomes empty.
|
||||||
- Le risorse backend sono sospese e ripristinate per `REQUIRES_NEW` e `NOT_SUPPORTED`.
|
- Backend resources are suspended and restored for `REQUIRES_NEW` and `NOT_SUPPORTED`.
|
||||||
- `TxSynchronization` è il punto di aggancio per hook di commit/rollback/completion.
|
- `TxSynchronization` is the hook point for commit/rollback/completion callbacks.
|
||||||
|
- Synchronizations live in a thread-local list; every new transaction records how many were already
|
||||||
|
registered when it opened and fires only its own tail, so a `REQUIRES_NEW` does not drag along
|
||||||
|
the suspended transaction's callbacks.
|
||||||
|
|||||||
+46
-3
@@ -53,9 +53,52 @@ public final class ResourceRegistry {
|
|||||||
SYNCHRONIZATIONS.get().add(Objects.requireNonNull(sync));
|
SYNCHRONIZATIONS.get().add(Objects.requireNonNull(sync));
|
||||||
}
|
}
|
||||||
|
|
||||||
public static void fireSynchronizations(TxOutcome outcome) {
|
/**
|
||||||
List<TxSynchronization> syncs = List.copyOf(SYNCHRONIZATIONS.get());
|
* How many synchronizations are registered right now — captured by a transaction manager when
|
||||||
SYNCHRONIZATIONS.get().clear();
|
* it opens a new transaction, and handed back to {@link #fireSynchronizations} on completion
|
||||||
|
* so that transaction only fires its own. See there for why that matters.
|
||||||
|
*/
|
||||||
|
public static int synchronizationCount() {
|
||||||
|
return SYNCHRONIZATIONS.get().size();
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Runs {@link TxSynchronization#beforeCommit} on the synchronizations registered from
|
||||||
|
* {@code fromIndex} onward, while their transaction is still active and its resource still
|
||||||
|
* bound. Unlike {@link #fireSynchronizations} this leaves them registered: they still have
|
||||||
|
* their post-completion callbacks to come. Exceptions propagate on purpose — a beforeCommit
|
||||||
|
* that throws vetoes the commit, see {@link TxSynchronization}.
|
||||||
|
*
|
||||||
|
* <p>Snapshots before iterating, so a callback that registers further synchronizations (a
|
||||||
|
* nested {@code Data#afterCommit}) doesn't mutate the list mid-loop. Those new ones join the
|
||||||
|
* transaction's post-completion callbacks without getting a {@code beforeCommit} of their own,
|
||||||
|
* which is the only coherent answer once the pass is already running.
|
||||||
|
*/
|
||||||
|
public static void fireBeforeCommit(boolean readOnly, int fromIndex) {
|
||||||
|
List<TxSynchronization> pending = SYNCHRONIZATIONS.get();
|
||||||
|
if (fromIndex >= pending.size()) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
for (TxSynchronization sync : List.copyOf(pending.subList(fromIndex, pending.size()))) {
|
||||||
|
sync.beforeCommit(readOnly);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Fires (and removes) the synchronizations registered from {@code fromIndex} onward — the ones
|
||||||
|
* belonging to the transaction now completing. Everything before that index was registered by
|
||||||
|
* an enclosing transaction that is merely <em>suspended</em>, not finished: a REQUIRES_NEW
|
||||||
|
* inner transaction sets its own baseline, so committing it no longer drags the outer's
|
||||||
|
* pending callbacks along — which fired them early, and with the inner transaction's outcome,
|
||||||
|
* for an outer transaction that might still roll back.
|
||||||
|
*/
|
||||||
|
public static void fireSynchronizations(TxOutcome outcome, int fromIndex) {
|
||||||
|
List<TxSynchronization> pending = SYNCHRONIZATIONS.get();
|
||||||
|
if (fromIndex >= pending.size()) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
List<TxSynchronization> syncs = List.copyOf(pending.subList(fromIndex, pending.size()));
|
||||||
|
pending.subList(fromIndex, pending.size()).clear();
|
||||||
for (TxSynchronization sync : syncs) {
|
for (TxSynchronization sync : syncs) {
|
||||||
if (outcome == TxOutcome.COMMITTED) {
|
if (outcome == TxOutcome.COMMITTED) {
|
||||||
sync.afterCommit();
|
sync.afterCommit();
|
||||||
|
|||||||
+50
@@ -1,8 +1,58 @@
|
|||||||
package dev.relism.flash.ext.data.core;
|
package dev.relism.flash.ext.data.core;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Lifecycle hooks for one transaction, registered through {@code Data#afterCommit} (or
|
||||||
|
* {@link ResourceRegistry#addSynchronization} directly) and fired by the {@link TxManager} that
|
||||||
|
* owns the transaction they were registered in.
|
||||||
|
*
|
||||||
|
* <p>Every callback belongs to exactly one transaction — the innermost one active at registration
|
||||||
|
* time — and fires exactly once, when <em>that</em> transaction completes. A joined
|
||||||
|
* ({@code REQUIRED}) inner transaction is not a transaction of its own, so callbacks registered
|
||||||
|
* inside one wait for the outermost commit; a {@code REQUIRES_NEW} transaction is, so completing
|
||||||
|
* it fires only its own callbacks and leaves the suspended outer transaction's alone.
|
||||||
|
*
|
||||||
|
* <h3>Where each hook sits relative to the commit</h3>
|
||||||
|
* <ul>
|
||||||
|
* <li>{@link #beforeCommit(boolean)} — immediately <b>before</b> the real commit, with the
|
||||||
|
* transaction still active and its session/connection still bound. This is the only hook
|
||||||
|
* that can still write through that same resource and have the write land in the same atomic
|
||||||
|
* unit: flush a buffer, stamp an audit row, materialize a derived value. Skipped entirely
|
||||||
|
* when the transaction is already rollback-only, since there is no commit to precede.</li>
|
||||||
|
* <li>{@link #afterCommit()} / {@link #afterRollback()}, then {@link #afterCompletion(TxOutcome)}
|
||||||
|
* — <b>after</b> the transaction has completed and its resource has been unbound. Nothing
|
||||||
|
* done here is part of the transaction: a callback that opens its own transaction gets a
|
||||||
|
* fresh one instead of joining the one that just finished, which is what makes this the
|
||||||
|
* right place to refresh a cache, enqueue a message, or notify anything outside the
|
||||||
|
* database.</li>
|
||||||
|
* </ul>
|
||||||
|
*
|
||||||
|
* <h3>Failure</h3>
|
||||||
|
* Throwing from {@link #beforeCommit(boolean)} <b>vetoes the commit</b>: the transaction is rolled
|
||||||
|
* back, {@link #afterRollback()}/{@link #afterCompletion(TxOutcome)} fire with
|
||||||
|
* {@link TxOutcome#ROLLED_BACK}, and the exception propagates to the caller. That is the point of
|
||||||
|
* this hook running before the commit rather than after — it can still refuse.
|
||||||
|
*
|
||||||
|
* <p>The post-completion hooks have no such power: the transaction is over by the time they run,
|
||||||
|
* so an exception from one propagates to the caller but changes nothing already committed, and
|
||||||
|
* stops the callbacks queued behind it.
|
||||||
|
*/
|
||||||
public interface TxSynchronization {
|
public interface TxSynchronization {
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Runs inside the transaction, immediately before it commits — see the interface javadoc.
|
||||||
|
* Throwing from here rolls the transaction back instead of committing it.
|
||||||
|
*
|
||||||
|
* @param readOnly whether the transaction was opened read-only, so a callback that would
|
||||||
|
* otherwise write can skip work it is not allowed to do
|
||||||
|
*/
|
||||||
default void beforeCommit(boolean readOnly) {}
|
default void beforeCommit(boolean readOnly) {}
|
||||||
|
|
||||||
|
/** Runs after a successful commit, with the transaction's resource already unbound. */
|
||||||
default void afterCommit() {}
|
default void afterCommit() {}
|
||||||
|
|
||||||
|
/** Runs after a rollback, with the transaction's resource already unbound. */
|
||||||
default void afterRollback() {}
|
default void afterRollback() {}
|
||||||
|
|
||||||
|
/** Runs after {@link #afterCommit()}/{@link #afterRollback()}, whichever applied. */
|
||||||
default void afterCompletion(TxOutcome outcome) {}
|
default void afterCompletion(TxOutcome outcome) {}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,14 +1,15 @@
|
|||||||
# flash-ext-data-hibernate
|
# flash-ext-data-hibernate
|
||||||
|
|
||||||
Backend Hibernate per `flash-ext-data-core`.
|
Hibernate backend for `flash-ext-data-core`.
|
||||||
|
|
||||||
## Scopo
|
## Purpose
|
||||||
|
|
||||||
Questo modulo implementa `TxManager` sopra `SessionFactory` e fornisce una base repository Hibernate-centric.
|
This module implements `TxManager` on top of a `SessionFactory` and provides a Hibernate-centric
|
||||||
|
repository base class.
|
||||||
|
|
||||||
## Come si usa
|
## How to use it
|
||||||
|
|
||||||
### 1. Creare il manager
|
### 1. Create the manager
|
||||||
|
|
||||||
```java
|
```java
|
||||||
SessionFactory sessionFactory = ...;
|
SessionFactory sessionFactory = ...;
|
||||||
@@ -16,12 +17,12 @@ HibernateTxManager txManager = new HibernateTxManager(sessionFactory);
|
|||||||
DataExtension extension = new DataExtension(txManager);
|
DataExtension extension = new DataExtension(txManager);
|
||||||
```
|
```
|
||||||
|
|
||||||
### 2. Installare l’estensione in Flash
|
### 2. Install the extension in Flash
|
||||||
|
|
||||||
L’estensione registra `Tx` e `TxManager` nel `FlashContext`.
|
The extension registers `Tx` and `TxManager` in the `FlashContext`. Class-based handlers annotated
|
||||||
Le handler class-based annotate con `@Transactional` vengono wrappate automaticamente.
|
with `@Transactional` are wrapped automatically.
|
||||||
|
|
||||||
### 3. Definire una repository
|
### 3. Define a repository
|
||||||
|
|
||||||
```java
|
```java
|
||||||
public final class UserRepository extends HibernateRepository<User, Long> {
|
public final class UserRepository extends HibernateRepository<User, Long> {
|
||||||
@@ -31,7 +32,7 @@ public final class UserRepository extends HibernateRepository<User, Long> {
|
|||||||
}
|
}
|
||||||
```
|
```
|
||||||
|
|
||||||
Con il nuovo modello query/spec puoi esporre campi riusabili come costanti:
|
With the query/spec model you can expose reusable fields as constants:
|
||||||
|
|
||||||
```java
|
```java
|
||||||
public final class UserRepository extends HibernateRepository<User, Long> {
|
public final class UserRepository extends HibernateRepository<User, Long> {
|
||||||
@@ -48,7 +49,7 @@ public final class UserRepository extends HibernateRepository<User, Long> {
|
|||||||
}
|
}
|
||||||
```
|
```
|
||||||
|
|
||||||
Le query domain-specific possono usare gli helper della base class:
|
Domain-specific queries can use the base class helpers:
|
||||||
|
|
||||||
```java
|
```java
|
||||||
public List<User> findByEmailDomain(String domain) {
|
public List<User> findByEmailDomain(String domain) {
|
||||||
@@ -58,35 +59,37 @@ public List<User> findByEmailDomain(String domain) {
|
|||||||
}
|
}
|
||||||
```
|
```
|
||||||
|
|
||||||
## Come funziona sotto
|
## How it works underneath
|
||||||
|
|
||||||
- La tx corrente è rappresentata da `HibernateTxStatus`.
|
- The current transaction is represented by `HibernateTxStatus`.
|
||||||
- La risorsa esposta al core è una `Session`.
|
- The resource exposed to the core is a `Session`.
|
||||||
- `Tx.resource(Session.class)` recupera la `Session` dal contesto corrente.
|
- `Tx.resource(Session.class)` retrieves the `Session` from the current context.
|
||||||
- `REQUIRES_NEW` sospende lo status attivo e apre una nuova `Session`.
|
- `REQUIRES_NEW` suspends the active status and opens a new `Session`.
|
||||||
- `NOT_SUPPORTED` sospende la tx attiva e continua senza sessione bindata.
|
- `NOT_SUPPORTED` suspends the active transaction and continues with no session bound.
|
||||||
|
|
||||||
## Repository base
|
## Repository base class
|
||||||
|
|
||||||
`HibernateRepository` fornisce:
|
`HibernateRepository` provides:
|
||||||
|
|
||||||
- `findById`, `findAll`, `findPage`, `findOne`
|
- `findById`, `findAll`, `findPage`, `findOne`
|
||||||
- `save`, `update`, `delete`, `saveAll`
|
- `save`, `update`, `delete`, `saveAll`
|
||||||
- bulk `deleteAll(Spec<T>)` e `updateAll(Spec<T>, T)`
|
- bulk `deleteAll(Spec<T>)` and `updateAll(Spec<T>, T)`
|
||||||
- helper HQL: `hql(...)`, `hqlMutate(...)`
|
- HQL helpers: `hql(...)`, `hqlMutate(...)`
|
||||||
|
|
||||||
Le classi concrete devono solo implementare query di dominio, non il plumbing transazionale.
|
Concrete classes only have to implement domain queries, never the transactional plumbing.
|
||||||
|
|
||||||
## Semantica transazionale
|
## Transactional semantics
|
||||||
|
|
||||||
- `REQUIRED`: join o apertura nuova tx.
|
- `REQUIRED`: join, or open a new transaction.
|
||||||
- `REQUIRES_NEW`: sospensione del contesto corrente.
|
- `REQUIRES_NEW`: suspend the current context.
|
||||||
- `SUPPORTS`: join se c’è tx, altrimenti no-op.
|
- `SUPPORTS`: join if a transaction exists, otherwise no-op.
|
||||||
- `NOT_SUPPORTED`: sospende e prosegue senza tx.
|
- `NOT_SUPPORTED`: suspend and continue without a transaction.
|
||||||
- `MANDATORY`: fallisce se non c’è tx.
|
- `MANDATORY`: fail if there is no transaction.
|
||||||
|
|
||||||
## Note
|
## Notes
|
||||||
|
|
||||||
- `Session` viene chiusa a fine tx nuova.
|
- The `Session` is closed when a new transaction ends.
|
||||||
- Le synchronizations vengono eseguite al commit/rollback.
|
- Synchronizations registered in a transaction fire when *that* transaction completes:
|
||||||
- Il backend è pensato per essere usato tramite la base class, non direttamente.
|
`beforeCommit` while it is still active and the `Session` still bound, the post-completion hooks
|
||||||
|
once it is unbound. See `flash-ext-data-core/docs/README.md` for the full contract.
|
||||||
|
- This backend is meant to be used through the base class, not directly.
|
||||||
|
|||||||
+40
-6
@@ -43,6 +43,9 @@ public class HibernateTxManager implements TxManager {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private TxStatus beginNew(TxDefinition definition, HibernateTxStatus suspended) {
|
private TxStatus beginNew(TxDefinition definition, HibernateTxStatus suspended) {
|
||||||
|
// Anything already registered belongs to an enclosing transaction this one is nested
|
||||||
|
// inside (or suspended over) — see ResourceRegistry#fireSynchronizations.
|
||||||
|
int synchronizationBaseline = ResourceRegistry.synchronizationCount();
|
||||||
Session s = sf.openSession();
|
Session s = sf.openSession();
|
||||||
boolean bound = false;
|
boolean bound = false;
|
||||||
try {
|
try {
|
||||||
@@ -56,7 +59,8 @@ public class HibernateTxManager implements TxManager {
|
|||||||
true,
|
true,
|
||||||
definition.readOnly(),
|
definition.readOnly(),
|
||||||
suspended,
|
suspended,
|
||||||
new HibernateTxStatus.RollbackMarker()
|
new HibernateTxStatus.RollbackMarker(),
|
||||||
|
synchronizationBaseline
|
||||||
);
|
);
|
||||||
ResourceRegistry.bind(HIBERNATE_STATUS_KEY, status);
|
ResourceRegistry.bind(HIBERNATE_STATUS_KEY, status);
|
||||||
bound = true;
|
bound = true;
|
||||||
@@ -80,12 +84,15 @@ public class HibernateTxManager implements TxManager {
|
|||||||
if (definition.readOnly() && !existing.isReadOnly()) {
|
if (definition.readOnly() && !existing.isReadOnly()) {
|
||||||
throw new TxException("Cannot join read-write tx as read-only");
|
throw new TxException("Cannot join read-write tx as read-only");
|
||||||
}
|
}
|
||||||
|
// Baseline 0 is never read: a joined status isn't a new transaction, so commit()/rollback()
|
||||||
|
// hand it straight back to the transaction it joined without firing anything.
|
||||||
return new HibernateTxStatus(
|
return new HibernateTxStatus(
|
||||||
existing.session(),
|
existing.session(),
|
||||||
false,
|
false,
|
||||||
definition.readOnly(),
|
definition.readOnly(),
|
||||||
null,
|
null,
|
||||||
existing.rollbackMarker()
|
existing.rollbackMarker(),
|
||||||
|
0
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -94,7 +101,7 @@ public class HibernateTxManager implements TxManager {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private TxStatus noOp(TxDefinition definition, HibernateTxStatus suspended) {
|
private TxStatus noOp(TxDefinition definition, HibernateTxStatus suspended) {
|
||||||
return new HibernateTxStatus(null, false, definition.readOnly(), suspended, new HibernateTxStatus.RollbackMarker());
|
return new HibernateTxStatus(null, false, definition.readOnly(), suspended, new HibernateTxStatus.RollbackMarker(), 0);
|
||||||
}
|
}
|
||||||
|
|
||||||
private HibernateTxStatus suspendIfNeeded() {
|
private HibernateTxStatus suspendIfNeeded() {
|
||||||
@@ -114,20 +121,45 @@ public class HibernateTxManager implements TxManager {
|
|||||||
cleanupIfIdle();
|
cleanupIfIdle();
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
TxOutcome outcome;
|
TxOutcome outcome = null;
|
||||||
try {
|
try {
|
||||||
if (s.isRollbackOnly() && s.session().getTransaction().isActive()) {
|
if (s.isRollbackOnly() && s.session().getTransaction().isActive()) {
|
||||||
s.session().getTransaction().rollback();
|
s.session().getTransaction().rollback();
|
||||||
outcome = TxOutcome.ROLLED_BACK;
|
outcome = TxOutcome.ROLLED_BACK;
|
||||||
} else {
|
} else {
|
||||||
|
// Still inside the transaction, session still bound: a beforeCommit callback can
|
||||||
|
// write through it and land in this same commit. Throwing from there vetoes the
|
||||||
|
// commit — see TxSynchronization.
|
||||||
|
ResourceRegistry.fireBeforeCommit(s.isReadOnly(), s.synchronizationBaseline());
|
||||||
s.session().getTransaction().commit();
|
s.session().getTransaction().commit();
|
||||||
outcome = TxOutcome.COMMITTED;
|
outcome = TxOutcome.COMMITTED;
|
||||||
}
|
}
|
||||||
|
} catch (RuntimeException e) {
|
||||||
|
// A vetoing beforeCommit, or a commit that failed outright: either way nothing was
|
||||||
|
// committed, so roll back and let the remaining callbacks hear ROLLED_BACK rather than
|
||||||
|
// nothing at all. A rollback failure here is swallowed deliberately — it would mask
|
||||||
|
// the exception that actually explains what went wrong, which is the one propagating.
|
||||||
|
if (s.session().getTransaction().isActive()) {
|
||||||
|
try {
|
||||||
|
s.session().getTransaction().rollback();
|
||||||
|
} catch (RuntimeException suppressed) {
|
||||||
|
e.addSuppressed(suppressed);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
outcome = TxOutcome.ROLLED_BACK;
|
||||||
|
throw e;
|
||||||
} finally {
|
} finally {
|
||||||
|
// Order is load-bearing, and getting it wrong is silent: cleanupIfIdle() calls
|
||||||
|
// ResourceRegistry.cleanup(), which removes the very ThreadLocal list of
|
||||||
|
// synchronizations still waiting to be fired — firing afterwards saw a freshly
|
||||||
|
// initialized empty list and dropped every callback on the floor. cleanupAndResume()
|
||||||
|
// still has to come first, so a synchronization that opens its own transaction
|
||||||
|
// (Registry#reload() in Pathway does) starts a fresh one instead of joining the
|
||||||
|
// session that just committed. rollback() below already had this order right.
|
||||||
cleanupAndResume(s);
|
cleanupAndResume(s);
|
||||||
|
if (outcome != null) ResourceRegistry.fireSynchronizations(outcome, s.synchronizationBaseline());
|
||||||
cleanupIfIdle();
|
cleanupIfIdle();
|
||||||
}
|
}
|
||||||
ResourceRegistry.fireSynchronizations(outcome);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
@@ -143,9 +175,11 @@ public class HibernateTxManager implements TxManager {
|
|||||||
if (s.session().getTransaction().isActive()) {
|
if (s.session().getTransaction().isActive()) {
|
||||||
s.session().getTransaction().rollback();
|
s.session().getTransaction().rollback();
|
||||||
}
|
}
|
||||||
ResourceRegistry.fireSynchronizations(TxOutcome.ROLLED_BACK);
|
|
||||||
} finally {
|
} finally {
|
||||||
|
// Same order as commit(): unbind the session first so a callback opening its own
|
||||||
|
// transaction gets a fresh one, fire before cleanupIfIdle() can drop the list.
|
||||||
cleanupAndResume(s);
|
cleanupAndResume(s);
|
||||||
|
ResourceRegistry.fireSynchronizations(TxOutcome.ROLLED_BACK, s.synchronizationBaseline());
|
||||||
cleanupIfIdle();
|
cleanupIfIdle();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+7
-1
@@ -13,19 +13,22 @@ class HibernateTxStatus implements TxStatus {
|
|||||||
private final boolean readOnly;
|
private final boolean readOnly;
|
||||||
private final HibernateTxStatus suspended;
|
private final HibernateTxStatus suspended;
|
||||||
private final RollbackMarker rollbackMarker;
|
private final RollbackMarker rollbackMarker;
|
||||||
|
private final int synchronizationBaseline;
|
||||||
|
|
||||||
HibernateTxStatus(
|
HibernateTxStatus(
|
||||||
Session session,
|
Session session,
|
||||||
boolean newTransaction,
|
boolean newTransaction,
|
||||||
boolean readOnly,
|
boolean readOnly,
|
||||||
HibernateTxStatus suspended,
|
HibernateTxStatus suspended,
|
||||||
RollbackMarker rollbackMarker
|
RollbackMarker rollbackMarker,
|
||||||
|
int synchronizationBaseline
|
||||||
) {
|
) {
|
||||||
this.session = session;
|
this.session = session;
|
||||||
this.newTransaction = newTransaction;
|
this.newTransaction = newTransaction;
|
||||||
this.readOnly = readOnly;
|
this.readOnly = readOnly;
|
||||||
this.suspended = suspended;
|
this.suspended = suspended;
|
||||||
this.rollbackMarker = rollbackMarker;
|
this.rollbackMarker = rollbackMarker;
|
||||||
|
this.synchronizationBaseline = synchronizationBaseline;
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override public boolean isNewTransaction() { return newTransaction; }
|
@Override public boolean isNewTransaction() { return newTransaction; }
|
||||||
@@ -44,4 +47,7 @@ class HibernateTxStatus implements TxStatus {
|
|||||||
Session session() { return session; }
|
Session session() { return session; }
|
||||||
HibernateTxStatus suspended() { return suspended; }
|
HibernateTxStatus suspended() { return suspended; }
|
||||||
RollbackMarker rollbackMarker() { return rollbackMarker; }
|
RollbackMarker rollbackMarker() { return rollbackMarker; }
|
||||||
|
|
||||||
|
/** Index into {@code ResourceRegistry}'s synchronization list where this transaction's own callbacks start. */
|
||||||
|
int synchronizationBaseline() { return synchronizationBaseline; }
|
||||||
}
|
}
|
||||||
|
|||||||
+258
@@ -0,0 +1,258 @@
|
|||||||
|
package dev.relism.flash.ext.data.hibernate;
|
||||||
|
|
||||||
|
import dev.relism.flash.ext.data.core.*;
|
||||||
|
import org.hibernate.Session;
|
||||||
|
import org.hibernate.SessionFactory;
|
||||||
|
import org.junit.jupiter.api.AfterAll;
|
||||||
|
import org.junit.jupiter.api.AfterEach;
|
||||||
|
import org.junit.jupiter.api.BeforeAll;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
import static org.junit.jupiter.api.Assertions.*;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Transaction synchronization semantics — the {@code beforeCommit}/{@code afterCommit}/
|
||||||
|
* {@code afterRollback} callbacks {@code Data#afterCommit} exposes, and the contract callers build
|
||||||
|
* on: "my callback runs once, for my transaction, on the right side of the commit".
|
||||||
|
*
|
||||||
|
* <p>None of this was covered before, and the gap was not academic: {@link #afterCommit_fires_on_commit}
|
||||||
|
* failed against the original {@code commit()}, which fired synchronizations only after
|
||||||
|
* {@code cleanupIfIdle()} had already dropped the ThreadLocal list holding them — so every callback
|
||||||
|
* was silently discarded, on every commit, with no error and no log. Downstream that meant an admin
|
||||||
|
* write landing in Postgres while the in-memory cache it was supposed to refresh never heard about
|
||||||
|
* it until the process restarted.
|
||||||
|
*/
|
||||||
|
class HibernateTxManagerSynchronizationTest {
|
||||||
|
|
||||||
|
static SessionFactory sf;
|
||||||
|
static HibernateTxManager manager;
|
||||||
|
|
||||||
|
@BeforeAll
|
||||||
|
static void setup() {
|
||||||
|
sf = TestHelper.buildSessionFactory();
|
||||||
|
manager = new HibernateTxManager(sf);
|
||||||
|
}
|
||||||
|
|
||||||
|
@AfterAll
|
||||||
|
static void teardown() {
|
||||||
|
if (sf != null) {
|
||||||
|
sf.close();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@AfterEach
|
||||||
|
void cleanup() {
|
||||||
|
ResourceRegistry.clear();
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Records which callbacks ran, in order — order is part of the contract, not just the fact they ran. */
|
||||||
|
private static final class Recorder implements TxSynchronization {
|
||||||
|
final List<String> calls = new ArrayList<>();
|
||||||
|
|
||||||
|
@Override public void beforeCommit(boolean readOnly) { calls.add("beforeCommit:" + readOnly); }
|
||||||
|
@Override public void afterCommit() { calls.add("afterCommit"); }
|
||||||
|
@Override public void afterRollback() { calls.add("afterRollback"); }
|
||||||
|
@Override public void afterCompletion(TxOutcome outcome) { calls.add("afterCompletion:" + outcome); }
|
||||||
|
}
|
||||||
|
|
||||||
|
private static final List<String> COMMITTED = List.of("beforeCommit:false", "afterCommit", "afterCompletion:COMMITTED");
|
||||||
|
private static final List<String> ROLLED_BACK = List.of("afterRollback", "afterCompletion:ROLLED_BACK");
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void afterCommit_fires_on_commit() {
|
||||||
|
Recorder recorder = new Recorder();
|
||||||
|
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
ResourceRegistry.addSynchronization(recorder);
|
||||||
|
|
||||||
|
manager.commit(tx);
|
||||||
|
|
||||||
|
assertEquals(COMMITTED, recorder.calls);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void afterRollback_fires_on_rollback() {
|
||||||
|
Recorder recorder = new Recorder();
|
||||||
|
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
ResourceRegistry.addSynchronization(recorder);
|
||||||
|
|
||||||
|
manager.rollback(tx);
|
||||||
|
|
||||||
|
assertEquals(ROLLED_BACK, recorder.calls);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** A commit() call on a tx already marked rollback-only really rolls back — and has no commit to precede. */
|
||||||
|
@Test
|
||||||
|
void commit_of_a_rollback_only_tx_rolls_back_without_running_beforeCommit() {
|
||||||
|
Recorder recorder = new Recorder();
|
||||||
|
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
tx.markRollbackOnly();
|
||||||
|
ResourceRegistry.addSynchronization(recorder);
|
||||||
|
|
||||||
|
manager.commit(tx);
|
||||||
|
|
||||||
|
assertEquals(ROLLED_BACK, recorder.calls);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** beforeCommit runs inside the transaction: session still bound, transaction still active. */
|
||||||
|
@Test
|
||||||
|
void beforeCommit_runs_while_the_transaction_is_still_active() {
|
||||||
|
List<Boolean> stillActive = new ArrayList<>();
|
||||||
|
List<Session> sessionSeen = new ArrayList<>();
|
||||||
|
|
||||||
|
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
Session session = tx.resource(Session.class);
|
||||||
|
ResourceRegistry.addSynchronization(new TxSynchronization() {
|
||||||
|
@Override
|
||||||
|
public void beforeCommit(boolean readOnly) {
|
||||||
|
stillActive.add(session.getTransaction().isActive());
|
||||||
|
// MANDATORY only succeeds while a transaction is bound — proof this runs inside it.
|
||||||
|
TxStatus joined = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.MANDATORY));
|
||||||
|
sessionSeen.add(joined.resource(Session.class));
|
||||||
|
manager.commit(joined);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
manager.commit(tx);
|
||||||
|
|
||||||
|
assertEquals(List.of(true), stillActive, "the transaction must not have committed yet");
|
||||||
|
assertSame(session, sessionSeen.get(0), "the same session must still be bound, so writes land in this commit");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void beforeCommit_is_told_whether_the_transaction_is_read_only() {
|
||||||
|
Recorder readWrite = new Recorder();
|
||||||
|
TxStatus rw = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
ResourceRegistry.addSynchronization(readWrite);
|
||||||
|
manager.commit(rw);
|
||||||
|
|
||||||
|
Recorder readOnly = new Recorder();
|
||||||
|
TxStatus ro = manager.begin(TxDefinition.DEFAULTS.asReadOnly());
|
||||||
|
ResourceRegistry.addSynchronization(readOnly);
|
||||||
|
manager.commit(ro);
|
||||||
|
|
||||||
|
assertEquals("beforeCommit:false", readWrite.calls.get(0));
|
||||||
|
assertEquals("beforeCommit:true", readOnly.calls.get(0));
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Throwing from beforeCommit is a veto: no commit, the rollback callbacks run, the exception propagates. */
|
||||||
|
@Test
|
||||||
|
void a_throwing_beforeCommit_vetoes_the_commit() {
|
||||||
|
Recorder recorder = new Recorder();
|
||||||
|
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
Session session = tx.resource(Session.class);
|
||||||
|
ResourceRegistry.addSynchronization(new TxSynchronization() {
|
||||||
|
@Override public void beforeCommit(boolean readOnly) { throw new IllegalStateException("veto"); }
|
||||||
|
});
|
||||||
|
ResourceRegistry.addSynchronization(recorder);
|
||||||
|
|
||||||
|
IllegalStateException thrown = assertThrows(IllegalStateException.class, () -> manager.commit(tx));
|
||||||
|
|
||||||
|
assertEquals("veto", thrown.getMessage());
|
||||||
|
assertFalse(session.getTransaction().isActive(), "the vetoed transaction must be rolled back, not left open");
|
||||||
|
assertEquals(ROLLED_BACK, recorder.calls, "the surviving callbacks must hear ROLLED_BACK, not silence");
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* The load-bearing ordering detail: post-completion callbacks run <em>after</em> the committed
|
||||||
|
* session is unbound, so a callback that opens its own transaction (a cache reload, an outbox
|
||||||
|
* drain) gets a fresh one instead of silently joining the transaction that just committed.
|
||||||
|
*/
|
||||||
|
@Test
|
||||||
|
void a_synchronization_may_open_its_own_transaction() {
|
||||||
|
List<Session> sessionsSeen = new ArrayList<>();
|
||||||
|
List<Boolean> wasNewTransaction = new ArrayList<>();
|
||||||
|
|
||||||
|
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
Session committedSession = outer.resource(Session.class);
|
||||||
|
ResourceRegistry.addSynchronization(new TxSynchronization() {
|
||||||
|
@Override
|
||||||
|
public void afterCommit() {
|
||||||
|
TxStatus own = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
sessionsSeen.add(own.resource(Session.class));
|
||||||
|
wasNewTransaction.add(own.isNewTransaction());
|
||||||
|
manager.commit(own);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
manager.commit(outer);
|
||||||
|
|
||||||
|
assertEquals(1, sessionsSeen.size(), "the callback must have run");
|
||||||
|
assertEquals(List.of(true), wasNewTransaction, "must start its own transaction, not join the committed one");
|
||||||
|
assertNotSame(committedSession, sessionsSeen.get(0));
|
||||||
|
}
|
||||||
|
|
||||||
|
/** A joined (REQUIRED) inner commit is not a real commit — callbacks wait for the outermost one. */
|
||||||
|
@Test
|
||||||
|
void a_joined_commit_defers_synchronizations_to_the_outermost_commit() {
|
||||||
|
Recorder recorder = new Recorder();
|
||||||
|
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
TxStatus inner = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.REQUIRED));
|
||||||
|
ResourceRegistry.addSynchronization(recorder);
|
||||||
|
|
||||||
|
manager.commit(inner);
|
||||||
|
assertEquals(List.of(), recorder.calls, "the joined commit did not commit anything yet");
|
||||||
|
|
||||||
|
manager.commit(outer);
|
||||||
|
assertEquals(COMMITTED, recorder.calls);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Each callback belongs to one transaction: a second transaction must not re-run the first's. */
|
||||||
|
@Test
|
||||||
|
void synchronizations_do_not_leak_into_the_next_transaction() {
|
||||||
|
Recorder recorder = new Recorder();
|
||||||
|
TxStatus first = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
ResourceRegistry.addSynchronization(recorder);
|
||||||
|
manager.commit(first);
|
||||||
|
recorder.calls.clear();
|
||||||
|
|
||||||
|
TxStatus second = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
manager.commit(second);
|
||||||
|
|
||||||
|
assertEquals(List.of(), recorder.calls);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A REQUIRES_NEW inner transaction suspends the outer one; committing the inner must not drag
|
||||||
|
* the still-pending outer transaction's callbacks along with it. They belong to a transaction
|
||||||
|
* that has not committed — and may yet roll back, in which case firing {@code afterCommit} for
|
||||||
|
* it would be a straight lie.
|
||||||
|
*/
|
||||||
|
@Test
|
||||||
|
void a_requires_new_commit_leaves_the_suspended_transactions_synchronizations_alone() {
|
||||||
|
Recorder outerSync = new Recorder();
|
||||||
|
Recorder innerSync = new Recorder();
|
||||||
|
|
||||||
|
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
ResourceRegistry.addSynchronization(outerSync);
|
||||||
|
|
||||||
|
TxStatus inner = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.REQUIRES_NEW));
|
||||||
|
ResourceRegistry.addSynchronization(innerSync);
|
||||||
|
manager.commit(inner);
|
||||||
|
|
||||||
|
assertEquals(COMMITTED, innerSync.calls);
|
||||||
|
assertEquals(List.of(), outerSync.calls, "the outer transaction has not committed yet");
|
||||||
|
|
||||||
|
manager.commit(outer);
|
||||||
|
assertEquals(COMMITTED, outerSync.calls);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** A rolled-back inner REQUIRES_NEW must not fire the outer's callbacks either — same reason, opposite outcome. */
|
||||||
|
@Test
|
||||||
|
void a_requires_new_rollback_leaves_the_suspended_transactions_synchronizations_alone() {
|
||||||
|
Recorder outerSync = new Recorder();
|
||||||
|
|
||||||
|
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
ResourceRegistry.addSynchronization(outerSync);
|
||||||
|
|
||||||
|
TxStatus inner = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.REQUIRES_NEW));
|
||||||
|
manager.rollback(inner);
|
||||||
|
|
||||||
|
assertEquals(List.of(), outerSync.calls, "the outer transaction is still open");
|
||||||
|
|
||||||
|
manager.commit(outer);
|
||||||
|
assertEquals(COMMITTED, outerSync.calls);
|
||||||
|
}
|
||||||
|
}
|
||||||
+40
@@ -65,4 +65,44 @@ class HibernateTxManagerTest {
|
|||||||
assertTrue(outer.isRollbackOnly());
|
assertTrue(outer.isRollbackOnly());
|
||||||
manager.rollback(outer);
|
manager.rollback(outer);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** SUPPORTS without an active transaction yields a sessionless status: not a transaction, no session to hand out. */
|
||||||
|
@Test
|
||||||
|
void supports_without_active_transaction_is_a_sessionless_no_op() {
|
||||||
|
TxStatus s = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.SUPPORTS));
|
||||||
|
|
||||||
|
assertFalse(s.isNewTransaction());
|
||||||
|
assertThrows(IllegalStateException.class, () -> s.resource(Session.class));
|
||||||
|
assertDoesNotThrow(() -> manager.commit(s));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void not_supported_suspends_the_active_transaction_and_restores_it_on_commit() {
|
||||||
|
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
Session outerSession = outer.resource(Session.class);
|
||||||
|
|
||||||
|
TxStatus suspended = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.NOT_SUPPORTED));
|
||||||
|
assertFalse(suspended.isNewTransaction());
|
||||||
|
assertThrows(IllegalStateException.class, () -> suspended.resource(Session.class));
|
||||||
|
manager.commit(suspended);
|
||||||
|
|
||||||
|
TxStatus rejoined = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.REQUIRED));
|
||||||
|
assertSame(outerSession, rejoined.resource(Session.class), "the suspended transaction must be back");
|
||||||
|
manager.rollback(outer);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void mandatory_without_active_transaction_is_rejected() {
|
||||||
|
assertThrows(IllegalStateException.class,
|
||||||
|
() -> manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.MANDATORY)));
|
||||||
|
}
|
||||||
|
|
||||||
|
/** A read-only join onto a read-write transaction is a contract violation, not a silent downgrade. */
|
||||||
|
@Test
|
||||||
|
void read_only_cannot_join_a_read_write_transaction() {
|
||||||
|
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
assertThrows(TxException.class,
|
||||||
|
() -> manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.REQUIRED).asReadOnly()));
|
||||||
|
manager.rollback(outer);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,14 +1,15 @@
|
|||||||
# flash-ext-data-jdbc
|
# flash-ext-data-jdbc
|
||||||
|
|
||||||
Backend JDBC per `flash-ext-data-core`.
|
JDBC backend for `flash-ext-data-core`.
|
||||||
|
|
||||||
## Scopo
|
## Purpose
|
||||||
|
|
||||||
Questo modulo implementa `TxManager` sopra `DataSource` e fornisce una base repository SQL raw.
|
This module implements `TxManager` on top of a `DataSource` and provides a raw-SQL repository base
|
||||||
|
class.
|
||||||
|
|
||||||
## Come si usa
|
## How to use it
|
||||||
|
|
||||||
### 1. Creare il manager
|
### 1. Create the manager
|
||||||
|
|
||||||
```java
|
```java
|
||||||
DataSource dataSource = ...;
|
DataSource dataSource = ...;
|
||||||
@@ -16,11 +17,12 @@ JdbcTxManager txManager = new JdbcTxManager(dataSource);
|
|||||||
DataExtension extension = new DataExtension(txManager);
|
DataExtension extension = new DataExtension(txManager);
|
||||||
```
|
```
|
||||||
|
|
||||||
### 2. Installare l’estensione in Flash
|
### 2. Install the extension in Flash
|
||||||
|
|
||||||
Come per Hibernate, `DataExtension` registra `Tx` nel `FlashContext` e abilita `@Transactional` sugli handler class-based.
|
As with Hibernate, `DataExtension` registers `Tx` in the `FlashContext` and enables
|
||||||
|
`@Transactional` on class-based handlers.
|
||||||
|
|
||||||
### 3. Definire una repository
|
### 3. Define a repository
|
||||||
|
|
||||||
```java
|
```java
|
||||||
public final class UserRepository extends JdbcRepository<User, Long> {
|
public final class UserRepository extends JdbcRepository<User, Long> {
|
||||||
@@ -35,7 +37,7 @@ public final class UserRepository extends JdbcRepository<User, Long> {
|
|||||||
}
|
}
|
||||||
```
|
```
|
||||||
|
|
||||||
Anche qui puoi esporre `Spec` riusabili e comporre query dal service layer:
|
Here too you can expose reusable `Spec`s and compose queries from the service layer:
|
||||||
|
|
||||||
```java
|
```java
|
||||||
public final class UserRepository extends JdbcRepository<User, Long> {
|
public final class UserRepository extends JdbcRepository<User, Long> {
|
||||||
@@ -47,7 +49,7 @@ public final class UserRepository extends JdbcRepository<User, Long> {
|
|||||||
}
|
}
|
||||||
```
|
```
|
||||||
|
|
||||||
Per il salvataggio e l’update devi fornire il binding esplicito:
|
Saving and updating need an explicit binding:
|
||||||
|
|
||||||
```java
|
```java
|
||||||
@Override
|
@Override
|
||||||
@@ -61,36 +63,39 @@ protected void bindInsert(PreparedStatement ps, User entity) throws SQLException
|
|||||||
}
|
}
|
||||||
```
|
```
|
||||||
|
|
||||||
## Come funziona sotto
|
## How it works underneath
|
||||||
|
|
||||||
- La tx corrente espone una `Connection`.
|
- The current transaction exposes a `Connection`.
|
||||||
- `Tx.resource(Connection.class)` recupera la connessione bindata al thread.
|
- `Tx.resource(Connection.class)` retrieves the connection bound to the thread.
|
||||||
- `REQUIRES_NEW` sospende la connessione attiva e ne apre una nuova.
|
- `REQUIRES_NEW` suspends the active connection and opens a new one.
|
||||||
- `NOT_SUPPORTED` sospende il contesto e prosegue senza tx.
|
- `NOT_SUPPORTED` suspends the context and continues without a transaction.
|
||||||
|
|
||||||
## Repository base
|
## Repository base class
|
||||||
|
|
||||||
`JdbcRepository` fornisce:
|
`JdbcRepository` provides:
|
||||||
|
|
||||||
- query `select` con `queryOne`, `queryMany`
|
- `select` queries through `queryOne`, `queryMany`
|
||||||
- mutation con `mutate`
|
- mutations through `mutate`
|
||||||
- persistenza con `doSave`, `doUpdate`
|
- persistence through `doSave`, `doUpdate`
|
||||||
- paging con `doFindPage`
|
- paging through `doFindPage`
|
||||||
- bulk `deleteAll(Spec<T>)`
|
- bulk `deleteAll(Spec<T>)`
|
||||||
- helper raw `queryOne(...)`, `queryMany(...)`, `mutate(...)`
|
- raw helpers `queryOne(...)`, `queryMany(...)`, `mutate(...)`
|
||||||
|
|
||||||
Le repository concrete devono solo tradurre tra `ResultSet` e dominio.
|
Concrete repositories only have to translate between `ResultSet` and the domain.
|
||||||
|
|
||||||
## Semantica transazionale
|
## Transactional semantics
|
||||||
|
|
||||||
- `REQUIRED`: join o apertura nuova tx.
|
- `REQUIRED`: join, or open a new transaction.
|
||||||
- `REQUIRES_NEW`: sospensione del contesto corrente.
|
- `REQUIRES_NEW`: suspend the current context.
|
||||||
- `SUPPORTS`: join se c’è tx, altrimenti no-op.
|
- `SUPPORTS`: join if a transaction exists, otherwise no-op.
|
||||||
- `NOT_SUPPORTED`: sospende e prosegue senza tx.
|
- `NOT_SUPPORTED`: suspend and continue without a transaction.
|
||||||
- `MANDATORY`: fallisce se non c’è tx.
|
- `MANDATORY`: fail if there is no transaction.
|
||||||
|
|
||||||
## Note
|
## Notes
|
||||||
|
|
||||||
- La `Connection` viene chiusa a fine tx nuova.
|
- The `Connection` is closed when a new transaction ends.
|
||||||
- Le synchronizations vengono eseguite al commit/rollback.
|
- Synchronizations registered in a transaction fire when *that* transaction completes:
|
||||||
- Se una repository usa `doDelete(T)`, il comportamento predefinito è non supportato: usare `deleteById` o override specifico.
|
`beforeCommit` while it is still active and the `Connection` still bound, the post-completion
|
||||||
|
hooks once it is unbound. See `flash-ext-data-core/docs/README.md` for the full contract.
|
||||||
|
- If a repository uses `doDelete(T)`, the default behaviour is unsupported: use `deleteById` or
|
||||||
|
override it.
|
||||||
|
|||||||
+46
-9
@@ -41,6 +41,9 @@ public class JdbcTxManager implements TxManager {
|
|||||||
|
|
||||||
private TxStatus beginNew(TxDefinition definition) {
|
private TxStatus beginNew(TxDefinition definition) {
|
||||||
Connection conn = null;
|
Connection conn = null;
|
||||||
|
// Anything already registered belongs to an enclosing transaction this one is nested
|
||||||
|
// inside (or suspended over) — see ResourceRegistry#fireSynchronizations.
|
||||||
|
int synchronizationBaseline = ResourceRegistry.synchronizationCount();
|
||||||
JdbcTxStatus suspended = suspendIfNeeded();
|
JdbcTxStatus suspended = suspendIfNeeded();
|
||||||
boolean bound = false;
|
boolean bound = false;
|
||||||
try {
|
try {
|
||||||
@@ -55,7 +58,8 @@ public class JdbcTxManager implements TxManager {
|
|||||||
true,
|
true,
|
||||||
definition.readOnly(),
|
definition.readOnly(),
|
||||||
suspended,
|
suspended,
|
||||||
new JdbcTxStatus.RollbackMarker()
|
new JdbcTxStatus.RollbackMarker(),
|
||||||
|
synchronizationBaseline
|
||||||
);
|
);
|
||||||
ResourceRegistry.bind(JDBC_STATUS_KEY, status);
|
ResourceRegistry.bind(JDBC_STATUS_KEY, status);
|
||||||
bound = true;
|
bound = true;
|
||||||
@@ -76,12 +80,15 @@ public class JdbcTxManager implements TxManager {
|
|||||||
if (definition.readOnly() && !existing.isReadOnly()) {
|
if (definition.readOnly() && !existing.isReadOnly()) {
|
||||||
throw new TxException("Cannot join read-write tx as read-only");
|
throw new TxException("Cannot join read-write tx as read-only");
|
||||||
}
|
}
|
||||||
|
// Baseline 0 is never read: a joined status isn't a new transaction, so commit()/rollback()
|
||||||
|
// hand it straight back to the transaction it joined without firing anything.
|
||||||
return new JdbcTxStatus(
|
return new JdbcTxStatus(
|
||||||
existing.connection(),
|
existing.connection(),
|
||||||
false,
|
false,
|
||||||
definition.readOnly(),
|
definition.readOnly(),
|
||||||
null,
|
null,
|
||||||
existing.rollbackMarker()
|
existing.rollbackMarker(),
|
||||||
|
0
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -90,7 +97,7 @@ public class JdbcTxManager implements TxManager {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private TxStatus noOp(TxDefinition definition, JdbcTxStatus suspended) {
|
private TxStatus noOp(TxDefinition definition, JdbcTxStatus suspended) {
|
||||||
return new JdbcTxStatus(null, false, definition.readOnly(), suspended, new JdbcTxStatus.RollbackMarker());
|
return new JdbcTxStatus(null, false, definition.readOnly(), suspended, new JdbcTxStatus.RollbackMarker(), 0);
|
||||||
}
|
}
|
||||||
|
|
||||||
private JdbcTxStatus suspendIfNeeded() {
|
private JdbcTxStatus suspendIfNeeded() {
|
||||||
@@ -110,22 +117,51 @@ public class JdbcTxManager implements TxManager {
|
|||||||
cleanupIfIdle();
|
cleanupIfIdle();
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
TxOutcome outcome = null;
|
||||||
try {
|
try {
|
||||||
if (s.isRollbackOnly()) {
|
if (s.isRollbackOnly()) {
|
||||||
s.connection().rollback();
|
s.connection().rollback();
|
||||||
ResourceRegistry.fireSynchronizations(TxOutcome.ROLLED_BACK);
|
outcome = TxOutcome.ROLLED_BACK;
|
||||||
return;
|
} else {
|
||||||
}
|
// Still inside the transaction, connection still bound: a beforeCommit callback
|
||||||
|
// can write through it and land in this same commit. Throwing from there vetoes
|
||||||
|
// the commit — see TxSynchronization.
|
||||||
|
ResourceRegistry.fireBeforeCommit(s.isReadOnly(), s.synchronizationBaseline());
|
||||||
s.connection().commit();
|
s.connection().commit();
|
||||||
ResourceRegistry.fireSynchronizations(TxOutcome.COMMITTED);
|
outcome = TxOutcome.COMMITTED;
|
||||||
|
}
|
||||||
} catch (SQLException e) {
|
} catch (SQLException e) {
|
||||||
throw new TxException(e);
|
TxException wrapped = new TxException(e);
|
||||||
|
outcome = rollbackAfterFailedCommit(s, wrapped);
|
||||||
|
throw wrapped;
|
||||||
|
} catch (RuntimeException e) {
|
||||||
|
outcome = rollbackAfterFailedCommit(s, e);
|
||||||
|
throw e;
|
||||||
} finally {
|
} finally {
|
||||||
|
// Unbind the connection before firing, so a callback that opens its own transaction
|
||||||
|
// gets a fresh one instead of joining the connection that just committed — and fire
|
||||||
|
// before cleanupIfIdle(), whose ResourceRegistry.cleanup() drops the pending list.
|
||||||
cleanupAndResume(s);
|
cleanupAndResume(s);
|
||||||
|
if (outcome != null) ResourceRegistry.fireSynchronizations(outcome, s.synchronizationBaseline());
|
||||||
cleanupIfIdle();
|
cleanupIfIdle();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Nothing was committed — a vetoing {@code beforeCommit}, or a commit that failed outright —
|
||||||
|
* so undo whatever the transaction had done and report {@code ROLLED_BACK} to the callbacks
|
||||||
|
* still queued behind it. A failure to roll back is attached to the exception already on its
|
||||||
|
* way out rather than replacing it: that one explains what actually went wrong.
|
||||||
|
*/
|
||||||
|
private static TxOutcome rollbackAfterFailedCommit(JdbcTxStatus s, Throwable propagating) {
|
||||||
|
try {
|
||||||
|
s.connection().rollback();
|
||||||
|
} catch (SQLException suppressed) {
|
||||||
|
propagating.addSuppressed(suppressed);
|
||||||
|
}
|
||||||
|
return TxOutcome.ROLLED_BACK;
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void rollback(TxStatus status) {
|
public void rollback(TxStatus status) {
|
||||||
JdbcTxStatus s = (JdbcTxStatus) status;
|
JdbcTxStatus s = (JdbcTxStatus) status;
|
||||||
@@ -137,11 +173,12 @@ public class JdbcTxManager implements TxManager {
|
|||||||
}
|
}
|
||||||
try {
|
try {
|
||||||
s.connection().rollback();
|
s.connection().rollback();
|
||||||
ResourceRegistry.fireSynchronizations(TxOutcome.ROLLED_BACK);
|
|
||||||
} catch (SQLException e) {
|
} catch (SQLException e) {
|
||||||
throw new TxException(e);
|
throw new TxException(e);
|
||||||
} finally {
|
} finally {
|
||||||
|
// Same order as commit() above, for the same two reasons.
|
||||||
cleanupAndResume(s);
|
cleanupAndResume(s);
|
||||||
|
ResourceRegistry.fireSynchronizations(TxOutcome.ROLLED_BACK, s.synchronizationBaseline());
|
||||||
cleanupIfIdle();
|
cleanupIfIdle();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+13
-3
@@ -3,7 +3,6 @@ package dev.relism.flash.ext.data.jdbc;
|
|||||||
import dev.relism.flash.ext.data.core.TxStatus;
|
import dev.relism.flash.ext.data.core.TxStatus;
|
||||||
|
|
||||||
import java.sql.Connection;
|
import java.sql.Connection;
|
||||||
import java.util.Objects;
|
|
||||||
|
|
||||||
class JdbcTxStatus implements TxStatus {
|
class JdbcTxStatus implements TxStatus {
|
||||||
static final class RollbackMarker {
|
static final class RollbackMarker {
|
||||||
@@ -15,19 +14,27 @@ class JdbcTxStatus implements TxStatus {
|
|||||||
private final boolean readOnly;
|
private final boolean readOnly;
|
||||||
private final JdbcTxStatus suspended;
|
private final JdbcTxStatus suspended;
|
||||||
private final RollbackMarker rollbackMarker;
|
private final RollbackMarker rollbackMarker;
|
||||||
|
private final int synchronizationBaseline;
|
||||||
|
|
||||||
|
// No requireNonNull on the connection: a SUPPORTS-without-a-transaction or a NOT_SUPPORTED
|
||||||
|
// status is deliberately connectionless (see JdbcTxManager#noOp), and rejecting null here
|
||||||
|
// turned both of those propagations into an NPE at begin() — the Hibernate manager has
|
||||||
|
// always allowed it. resource() reports the real mistake, asking a connectionless status for
|
||||||
|
// its connection, where it can name it.
|
||||||
JdbcTxStatus(
|
JdbcTxStatus(
|
||||||
Connection connection,
|
Connection connection,
|
||||||
boolean newTransaction,
|
boolean newTransaction,
|
||||||
boolean readOnly,
|
boolean readOnly,
|
||||||
JdbcTxStatus suspended,
|
JdbcTxStatus suspended,
|
||||||
RollbackMarker rollbackMarker
|
RollbackMarker rollbackMarker,
|
||||||
|
int synchronizationBaseline
|
||||||
) {
|
) {
|
||||||
this.connection = Objects.requireNonNull(connection);
|
this.connection = connection;
|
||||||
this.newTransaction = newTransaction;
|
this.newTransaction = newTransaction;
|
||||||
this.readOnly = readOnly;
|
this.readOnly = readOnly;
|
||||||
this.suspended = suspended;
|
this.suspended = suspended;
|
||||||
this.rollbackMarker = rollbackMarker;
|
this.rollbackMarker = rollbackMarker;
|
||||||
|
this.synchronizationBaseline = synchronizationBaseline;
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override public boolean isNewTransaction() { return newTransaction; }
|
@Override public boolean isNewTransaction() { return newTransaction; }
|
||||||
@@ -46,4 +53,7 @@ class JdbcTxStatus implements TxStatus {
|
|||||||
Connection connection() { return connection; }
|
Connection connection() { return connection; }
|
||||||
JdbcTxStatus suspended() { return suspended; }
|
JdbcTxStatus suspended() { return suspended; }
|
||||||
RollbackMarker rollbackMarker() { return rollbackMarker; }
|
RollbackMarker rollbackMarker() { return rollbackMarker; }
|
||||||
|
|
||||||
|
/** Index into {@code ResourceRegistry}'s synchronization list where this transaction's own callbacks start. */
|
||||||
|
int synchronizationBaseline() { return synchronizationBaseline; }
|
||||||
}
|
}
|
||||||
|
|||||||
+216
@@ -0,0 +1,216 @@
|
|||||||
|
package dev.relism.flash.ext.data.jdbc;
|
||||||
|
|
||||||
|
import dev.relism.flash.ext.data.core.*;
|
||||||
|
import org.junit.jupiter.api.AfterEach;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
|
||||||
|
import java.sql.Connection;
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
import static org.junit.jupiter.api.Assertions.*;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Transaction synchronization semantics for the JDBC manager — the same contract {@code
|
||||||
|
* HibernateTxManagerSynchronizationTest} pins down for the Hibernate one, kept deliberately
|
||||||
|
* parallel: the two managers are interchangeable behind {@code TxManager}, so a callback must not
|
||||||
|
* observe a different lifecycle depending on which one is installed.
|
||||||
|
*/
|
||||||
|
class JdbcTxManagerSynchronizationTest {
|
||||||
|
|
||||||
|
private final JdbcTxManager manager = new JdbcTxManager(new TestDataSource());
|
||||||
|
|
||||||
|
@AfterEach
|
||||||
|
void cleanup() {
|
||||||
|
ResourceRegistry.clear();
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Records which callbacks ran, in order — order is part of the contract, not just the fact they ran. */
|
||||||
|
private static final class Recorder implements TxSynchronization {
|
||||||
|
final List<String> calls = new ArrayList<>();
|
||||||
|
|
||||||
|
@Override public void beforeCommit(boolean readOnly) { calls.add("beforeCommit:" + readOnly); }
|
||||||
|
@Override public void afterCommit() { calls.add("afterCommit"); }
|
||||||
|
@Override public void afterRollback() { calls.add("afterRollback"); }
|
||||||
|
@Override public void afterCompletion(TxOutcome outcome) { calls.add("afterCompletion:" + outcome); }
|
||||||
|
}
|
||||||
|
|
||||||
|
private static final List<String> COMMITTED = List.of("beforeCommit:false", "afterCommit", "afterCompletion:COMMITTED");
|
||||||
|
private static final List<String> ROLLED_BACK = List.of("afterRollback", "afterCompletion:ROLLED_BACK");
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void afterCommit_fires_on_commit() {
|
||||||
|
Recorder recorder = new Recorder();
|
||||||
|
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
ResourceRegistry.addSynchronization(recorder);
|
||||||
|
|
||||||
|
manager.commit(tx);
|
||||||
|
|
||||||
|
assertEquals(COMMITTED, recorder.calls);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void afterRollback_fires_on_rollback() {
|
||||||
|
Recorder recorder = new Recorder();
|
||||||
|
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
ResourceRegistry.addSynchronization(recorder);
|
||||||
|
|
||||||
|
manager.rollback(tx);
|
||||||
|
|
||||||
|
assertEquals(ROLLED_BACK, recorder.calls);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void commit_of_a_rollback_only_tx_rolls_back_without_running_beforeCommit() {
|
||||||
|
Recorder recorder = new Recorder();
|
||||||
|
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
tx.markRollbackOnly();
|
||||||
|
ResourceRegistry.addSynchronization(recorder);
|
||||||
|
|
||||||
|
manager.commit(tx);
|
||||||
|
|
||||||
|
assertEquals(ROLLED_BACK, recorder.calls);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** beforeCommit runs inside the transaction: connection still bound, nothing committed yet. */
|
||||||
|
@Test
|
||||||
|
void beforeCommit_runs_while_the_transaction_is_still_active() {
|
||||||
|
List<Connection> connectionSeen = new ArrayList<>();
|
||||||
|
|
||||||
|
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
Connection connection = tx.resource(Connection.class);
|
||||||
|
ResourceRegistry.addSynchronization(new TxSynchronization() {
|
||||||
|
@Override
|
||||||
|
public void beforeCommit(boolean readOnly) {
|
||||||
|
// MANDATORY only succeeds while a transaction is bound — proof this runs inside it.
|
||||||
|
TxStatus joined = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.MANDATORY));
|
||||||
|
connectionSeen.add(joined.resource(Connection.class));
|
||||||
|
manager.commit(joined);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
manager.commit(tx);
|
||||||
|
|
||||||
|
assertSame(connection, connectionSeen.get(0), "the same connection must still be bound, so writes land in this commit");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void beforeCommit_is_told_whether_the_transaction_is_read_only() {
|
||||||
|
Recorder readWrite = new Recorder();
|
||||||
|
TxStatus rw = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
ResourceRegistry.addSynchronization(readWrite);
|
||||||
|
manager.commit(rw);
|
||||||
|
|
||||||
|
Recorder readOnly = new Recorder();
|
||||||
|
TxStatus ro = manager.begin(TxDefinition.DEFAULTS.asReadOnly());
|
||||||
|
ResourceRegistry.addSynchronization(readOnly);
|
||||||
|
manager.commit(ro);
|
||||||
|
|
||||||
|
assertEquals("beforeCommit:false", readWrite.calls.get(0));
|
||||||
|
assertEquals("beforeCommit:true", readOnly.calls.get(0));
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Throwing from beforeCommit is a veto: no commit, the rollback callbacks run, the exception propagates. */
|
||||||
|
@Test
|
||||||
|
void a_throwing_beforeCommit_vetoes_the_commit() {
|
||||||
|
Recorder recorder = new Recorder();
|
||||||
|
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
ResourceRegistry.addSynchronization(new TxSynchronization() {
|
||||||
|
@Override public void beforeCommit(boolean readOnly) { throw new IllegalStateException("veto"); }
|
||||||
|
});
|
||||||
|
ResourceRegistry.addSynchronization(recorder);
|
||||||
|
|
||||||
|
IllegalStateException thrown = assertThrows(IllegalStateException.class, () -> manager.commit(tx));
|
||||||
|
|
||||||
|
assertEquals("veto", thrown.getMessage());
|
||||||
|
assertEquals(ROLLED_BACK, recorder.calls, "the surviving callbacks must hear ROLLED_BACK, not silence");
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Post-completion callbacks run after the committed connection is unbound, so opening a transaction gets a fresh one. */
|
||||||
|
@Test
|
||||||
|
void a_synchronization_may_open_its_own_transaction() {
|
||||||
|
List<Connection> connectionsSeen = new ArrayList<>();
|
||||||
|
List<Boolean> wasNewTransaction = new ArrayList<>();
|
||||||
|
|
||||||
|
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
Connection committedConnection = outer.resource(Connection.class);
|
||||||
|
ResourceRegistry.addSynchronization(new TxSynchronization() {
|
||||||
|
@Override
|
||||||
|
public void afterCommit() {
|
||||||
|
TxStatus own = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
connectionsSeen.add(own.resource(Connection.class));
|
||||||
|
wasNewTransaction.add(own.isNewTransaction());
|
||||||
|
manager.commit(own);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
manager.commit(outer);
|
||||||
|
|
||||||
|
assertEquals(1, connectionsSeen.size(), "the callback must have run");
|
||||||
|
assertEquals(List.of(true), wasNewTransaction, "must start its own transaction, not join the committed one");
|
||||||
|
assertNotSame(committedConnection, connectionsSeen.get(0));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void a_joined_commit_defers_synchronizations_to_the_outermost_commit() {
|
||||||
|
Recorder recorder = new Recorder();
|
||||||
|
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
TxStatus inner = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.REQUIRED));
|
||||||
|
ResourceRegistry.addSynchronization(recorder);
|
||||||
|
|
||||||
|
manager.commit(inner);
|
||||||
|
assertEquals(List.of(), recorder.calls, "the joined commit did not commit anything yet");
|
||||||
|
|
||||||
|
manager.commit(outer);
|
||||||
|
assertEquals(COMMITTED, recorder.calls);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void synchronizations_do_not_leak_into_the_next_transaction() {
|
||||||
|
Recorder recorder = new Recorder();
|
||||||
|
TxStatus first = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
ResourceRegistry.addSynchronization(recorder);
|
||||||
|
manager.commit(first);
|
||||||
|
recorder.calls.clear();
|
||||||
|
|
||||||
|
TxStatus second = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
manager.commit(second);
|
||||||
|
|
||||||
|
assertEquals(List.of(), recorder.calls);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void a_requires_new_commit_leaves_the_suspended_transactions_synchronizations_alone() {
|
||||||
|
Recorder outerSync = new Recorder();
|
||||||
|
Recorder innerSync = new Recorder();
|
||||||
|
|
||||||
|
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
ResourceRegistry.addSynchronization(outerSync);
|
||||||
|
|
||||||
|
TxStatus inner = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.REQUIRES_NEW));
|
||||||
|
ResourceRegistry.addSynchronization(innerSync);
|
||||||
|
manager.commit(inner);
|
||||||
|
|
||||||
|
assertEquals(COMMITTED, innerSync.calls);
|
||||||
|
assertEquals(List.of(), outerSync.calls, "the outer transaction has not committed yet");
|
||||||
|
|
||||||
|
manager.commit(outer);
|
||||||
|
assertEquals(COMMITTED, outerSync.calls);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void a_requires_new_rollback_leaves_the_suspended_transactions_synchronizations_alone() {
|
||||||
|
Recorder outerSync = new Recorder();
|
||||||
|
|
||||||
|
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
ResourceRegistry.addSynchronization(outerSync);
|
||||||
|
|
||||||
|
TxStatus inner = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.REQUIRES_NEW));
|
||||||
|
manager.rollback(inner);
|
||||||
|
|
||||||
|
assertEquals(List.of(), outerSync.calls, "the outer transaction is still open");
|
||||||
|
|
||||||
|
manager.commit(outer);
|
||||||
|
assertEquals(COMMITTED, outerSync.calls);
|
||||||
|
}
|
||||||
|
}
|
||||||
+30
-47
@@ -4,15 +4,12 @@ import dev.relism.flash.ext.data.core.*;
|
|||||||
import org.junit.jupiter.api.AfterEach;
|
import org.junit.jupiter.api.AfterEach;
|
||||||
import org.junit.jupiter.api.Test;
|
import org.junit.jupiter.api.Test;
|
||||||
|
|
||||||
import javax.sql.DataSource;
|
|
||||||
import java.sql.Connection;
|
import java.sql.Connection;
|
||||||
import java.sql.DriverManager;
|
|
||||||
import java.sql.SQLException;
|
|
||||||
|
|
||||||
import static org.junit.jupiter.api.Assertions.*;
|
import static org.junit.jupiter.api.Assertions.*;
|
||||||
|
|
||||||
class JdbcTxManagerTest {
|
class JdbcTxManagerTest {
|
||||||
private final JdbcTxManager manager = new JdbcTxManager(dataSource());
|
private final JdbcTxManager manager = new JdbcTxManager(new TestDataSource());
|
||||||
|
|
||||||
@AfterEach
|
@AfterEach
|
||||||
void cleanup() {
|
void cleanup() {
|
||||||
@@ -53,52 +50,38 @@ class JdbcTxManagerTest {
|
|||||||
manager.rollback(outer);
|
manager.rollback(outer);
|
||||||
}
|
}
|
||||||
|
|
||||||
private static DataSource dataSource() {
|
/**
|
||||||
return new DataSource() {
|
* SUPPORTS without an active transaction yields a connectionless status — it must not be a
|
||||||
@Override
|
* transaction, and asking it for a connection must say so rather than NPE. Both propagations
|
||||||
public Connection getConnection() throws SQLException {
|
* that produce one used to throw {@link NullPointerException} straight out of {@code begin()}.
|
||||||
return DriverManager.getConnection("jdbc:h2:mem:tx-jdbc;DB_CLOSE_DELAY=-1");
|
*/
|
||||||
|
@Test
|
||||||
|
void supports_without_active_transaction_is_a_connectionless_no_op() {
|
||||||
|
TxStatus s = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.SUPPORTS));
|
||||||
|
|
||||||
|
assertFalse(s.isNewTransaction());
|
||||||
|
assertThrows(IllegalStateException.class, () -> s.resource(Connection.class));
|
||||||
|
assertDoesNotThrow(() -> manager.commit(s));
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Test
|
||||||
public Connection getConnection(String username, String password) throws SQLException {
|
void not_supported_suspends_the_active_transaction_and_restores_it_on_commit() {
|
||||||
return DriverManager.getConnection("jdbc:h2:mem:tx-jdbc;DB_CLOSE_DELAY=-1", username, password);
|
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||||
|
Connection outerConnection = outer.resource(Connection.class);
|
||||||
|
|
||||||
|
TxStatus suspended = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.NOT_SUPPORTED));
|
||||||
|
assertFalse(suspended.isNewTransaction());
|
||||||
|
assertThrows(IllegalStateException.class, () -> suspended.resource(Connection.class));
|
||||||
|
manager.commit(suspended);
|
||||||
|
|
||||||
|
TxStatus rejoined = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.REQUIRED));
|
||||||
|
assertSame(outerConnection, rejoined.resource(Connection.class), "the suspended transaction must be back");
|
||||||
|
manager.rollback(outer);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Test
|
||||||
public <T> T unwrap(Class<T> iface) {
|
void mandatory_without_active_transaction_is_rejected() {
|
||||||
throw new UnsupportedOperationException();
|
assertThrows(IllegalStateException.class,
|
||||||
}
|
() -> manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.MANDATORY)));
|
||||||
|
|
||||||
@Override
|
|
||||||
public boolean isWrapperFor(Class<?> iface) {
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public java.io.PrintWriter getLogWriter() {
|
|
||||||
throw new UnsupportedOperationException();
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public void setLogWriter(java.io.PrintWriter out) {
|
|
||||||
throw new UnsupportedOperationException();
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public void setLoginTimeout(int seconds) {
|
|
||||||
throw new UnsupportedOperationException();
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public int getLoginTimeout() {
|
|
||||||
return 0;
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public java.util.logging.Logger getParentLogger() {
|
|
||||||
throw new UnsupportedOperationException();
|
|
||||||
}
|
|
||||||
};
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+37
@@ -0,0 +1,37 @@
|
|||||||
|
package dev.relism.flash.ext.data.jdbc;
|
||||||
|
|
||||||
|
import javax.sql.DataSource;
|
||||||
|
import java.io.PrintWriter;
|
||||||
|
import java.sql.Connection;
|
||||||
|
import java.sql.DriverManager;
|
||||||
|
import java.sql.SQLException;
|
||||||
|
import java.util.logging.Logger;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A bare in-memory H2 {@link DataSource} — one fresh connection per {@code getConnection()}, which
|
||||||
|
* is all {@code JdbcTxManager} needs to exercise real commit/rollback and suspension. Every method
|
||||||
|
* outside the two {@code getConnection} overloads throws: nothing under test calls them, and a
|
||||||
|
* loud failure beats a silent stub if that ever changes.
|
||||||
|
*/
|
||||||
|
final class TestDataSource implements DataSource {
|
||||||
|
|
||||||
|
static final String URL = "jdbc:h2:mem:tx-jdbc;DB_CLOSE_DELAY=-1";
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public Connection getConnection() throws SQLException {
|
||||||
|
return DriverManager.getConnection(URL);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public Connection getConnection(String username, String password) throws SQLException {
|
||||||
|
return DriverManager.getConnection(URL, username, password);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override public <T> T unwrap(Class<T> iface) { throw new UnsupportedOperationException(); }
|
||||||
|
@Override public boolean isWrapperFor(Class<?> iface) { return false; }
|
||||||
|
@Override public PrintWriter getLogWriter() { throw new UnsupportedOperationException(); }
|
||||||
|
@Override public void setLogWriter(PrintWriter out) { throw new UnsupportedOperationException(); }
|
||||||
|
@Override public void setLoginTimeout(int seconds) { throw new UnsupportedOperationException(); }
|
||||||
|
@Override public int getLoginTimeout() { return 0; }
|
||||||
|
@Override public Logger getParentLogger() { throw new UnsupportedOperationException(); }
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user