Relism/Flash5
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: introduce WebSocket support with new endpoints and transaction propagation enhancements

Browse Source
This commit is contained in:
Relism
2026-05-11 16:14:26 +02:00
parent a4a16bdb00
commit ccc5550598
55 changed files with 2705 additions and 752 deletions
Download Patch File Download Diff File Expand all files Collapse all files
flash-extensions/flash-ext-data-jdbc/src/main/java/dev/relism/flash/ext/data/jdbc/JdbcRepository.java
+95 -63
View File
@@ -3,87 +3,99 @@ package dev.relism.flash.ext.data.jdbc;
import dev.relism.flash.ext.data.core.*;
import java.sql.*;
import java.util.*;
import java.util.stream.Collectors;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
public abstract class JdbcRepository<T, ID> extends Repository<T, ID> {
private final String table;
private final String idColumn;
protected JdbcRepository(String table, String idColumn) {
this.table = table;
protected JdbcRepository(Tx tx, String table, String idColumn) {
super(tx);
this.table = table;
this.idColumn = idColumn;
}
protected Connection connection() {
return Tx.resource(Connection.class);
return tx().resource(Connection.class);
}
// ── Subclass contract ─────────────────────────────────────────────────────
protected abstract T mapRow(ResultSet rs) throws SQLException;
protected abstract void bindInsert(PreparedStatement ps, T entity) throws SQLException;
protected abstract void bindUpdate(PreparedStatement ps, T entity) throws SQLException;
protected abstract T mapRow(ResultSet rs) throws SQLException;
protected abstract void bindInsert(PreparedStatement ps, T entity) throws SQLException;
protected abstract void bindUpdate(PreparedStatement ps, T entity) throws SQLException;
protected abstract String insertSql();
protected abstract String updateSql();
// ── Repository abstract impl ──────────────────────────────────────────────
@Override
protected Optional<T> doFindById(ID id) {
return queryOne("select * from " + table + " where " + idColumn + " = ?",
ps -> ps.setObject(1, id));
return queryOne("select * from " + table + " where " + idColumn + " = ?", ps -> ps.setObject(1, id));
}
@Override
protected List<T> doFindAll() {
return queryMany("select * from " + table, ps -> {});
}
protected List<T> doFind(Query<T> query) {
JdbcSpecContext ctx = new JdbcSpecContext();
String where = query.spec() != null ? " where " + query.spec().toFragment(ctx) : "";
String order = query.sort() != null && query.sort().isSorted() ? " order by " + orderClause(query.sort()) : "";
String paging = query.isPaged() ? " limit ? offset ?" : "";
@Override
protected List<T> doFindAll(int page, int size) {
return queryMany("select * from " + table + " limit ? offset ?", ps -> {
ps.setInt(1, size);
ps.setInt(2, page * size);
return queryMany("select * from " + table + where + order + paging, ps -> {
if (query.isPaged()) {
ctx.applyParameters(ps);
int base = ctx.size();
ps.setInt(base + 1, query.size());
ps.setInt(base + 2, query.page() * query.size());
return;
}
ctx.applyParameters(ps);
});
}
@Override
protected List<T> doFindAll(Sort sort) {
return queryMany("select * from " + table + orderClause(sort), ps -> {});
protected Optional<T> doFindOne(Spec<T> spec) {
return doFind(Query.<T>all().where(spec).page(0, 1)).stream().findFirst();
}
@Override
protected List<T> doFindAll(int page, int size, Sort sort) {
return queryMany("select * from " + table + orderClause(sort) + " limit ? offset ?",
ps -> {
ps.setInt(1, size);
ps.setInt(2, page * size);
});
protected Page<T> doFindPage(Query<T> query) {
if (!query.isPaged()) {
throw new IllegalArgumentException("Paged query requires page and size");
}
long total = countWhere(query.spec());
List<T> content = doFind(query);
return new Page<>(content, query.page(), query.size(), total);
}
@Override
protected Page<T> doFindPage(int page, int size) {
long total = doCount();
return new Page<>(doFindAll(page, size), page, size, total);
protected boolean doExistsById(ID id) {
return queryOne("select 1 from " + table + " where " + idColumn + " = ?", ps -> ps.setObject(1, id), rs -> rs.getInt(1)).isPresent();
}
@Override
protected Page<T> doFindPage(int page, int size, Sort sort) {
long total = doCount();
return new Page<>(doFindAll(page, size, sort), page, size, total);
protected long doCount() {
return queryOne("select count(*) from " + table, ps -> {}, rs -> rs.getLong(1)).orElse(0L);
}
@Override
protected T doSave(T entity) {
try (PreparedStatement ps = connection().prepareStatement(
insertSql(), Statement.RETURN_GENERATED_KEYS)) {
try (PreparedStatement ps = connection().prepareStatement(insertSql(), Statement.RETURN_GENERATED_KEYS)) {
bindInsert(ps, entity);
ps.executeUpdate();
applyGeneratedKey(ps, entity);
return entity;
} catch (SQLException e) { throw new TxException(e); }
} catch (SQLException e) {
throw new TxException(e);
}
}
@Override
protected List<T> doSaveAll(Iterable<T> entities) {
List<T> saved = new ArrayList<>();
for (T entity : entities) {
saved.add(doSave(entity));
}
return saved;
}
@Override
@@ -92,7 +104,9 @@ public abstract class JdbcRepository<T, ID> extends Repository<T, ID> {
bindUpdate(ps, entity);
ps.executeUpdate();
return entity;
} catch (SQLException e) { throw new TxException(e); }
} catch (SQLException e) {
throw new TxException(e);
}
}
@Override
@@ -102,38 +116,35 @@ public abstract class JdbcRepository<T, ID> extends Repository<T, ID> {
@Override
protected void doDeleteById(ID id) {
mutate("delete from " + table + " where " + idColumn + " = ?",
ps -> ps.setObject(1, id));
mutate("delete from " + table + " where " + idColumn + " = ?", ps -> ps.setObject(1, id));
}
@Override
protected boolean doExistsById(ID id) {
return queryOne("select 1 from " + table + " where " + idColumn + " = ?",
ps -> ps.setObject(1, id),
rs -> rs.getInt(1)).isPresent();
protected int doDeleteAll(Spec<T> spec) {
JdbcSpecContext ctx = new JdbcSpecContext();
String where = " where " + spec.toFragment(ctx);
return mutate("delete from " + table + where, ctx::applyParameters);
}
@Override
protected long doCount() {
return queryOne("select count(*) from " + table, ps -> {},
rs -> rs.getLong(1)).orElse(0L);
protected int doUpdateAll(Spec<T> spec, T patch) {
throw new UnsupportedOperationException("Override doUpdateAll() for bulk UPDATE support");
}
// ── Query helpers ─────────────────────────────────────────────────────────
protected Optional<T> queryOne(String sql, SqlBinder params) {
List<T> r = queryMany(sql, params);
return r.isEmpty() ? Optional.empty() : Optional.of(r.get(0));
}
protected <R> Optional<R> queryOne(String sql, SqlBinder params,
SqlMapper<R> mapper) {
protected <R> Optional<R> queryOne(String sql, SqlBinder params, SqlMapper<R> mapper) {
try (PreparedStatement ps = connection().prepareStatement(sql)) {
params.bind(ps);
try (ResultSet rs = ps.executeQuery()) {
return rs.next() ? Optional.of(mapper.map(rs)) : Optional.empty();
}
} catch (SQLException e) { throw new TxException(e); }
} catch (SQLException e) {
throw new TxException(e);
}
}
protected List<T> queryMany(String sql, SqlBinder params) {
@@ -144,26 +155,47 @@ public abstract class JdbcRepository<T, ID> extends Repository<T, ID> {
while (rs.next()) results.add(mapRow(rs));
return results;
}
} catch (SQLException e) { throw new TxException(e); }
} catch (SQLException e) {
throw new TxException(e);
}
}
protected int mutate(String sql, SqlBinder params) {
try (PreparedStatement ps = connection().prepareStatement(sql)) {
params.bind(ps);
return ps.executeUpdate();
} catch (SQLException e) { throw new TxException(e); }
} catch (SQLException e) {
throw new TxException(e);
}
}
protected void applyGeneratedKey(PreparedStatement ps, T entity) throws SQLException {
// override when entity has a generated PK
}
private String orderClause(Sort sort) {
return " order by " + sort.columns().stream()
.map(c -> c.column() + (c.asc() ? " ASC" : " DESC"))
.collect(Collectors.joining(", "));
protected Class<T> entityType() {
return null;
}
@FunctionalInterface public interface SqlBinder { void bind(PreparedStatement ps) throws SQLException; }
@FunctionalInterface public interface SqlMapper<R> { R map(ResultSet rs) throws SQLException; }
}
private long countWhere(Spec<T> spec) {
JdbcSpecContext ctx = new JdbcSpecContext();
String where = spec != null ? " where " + spec.toFragment(ctx) : "";
return queryOne("select count(*) from " + table + where, ctx::applyParameters, rs -> rs.getLong(1)).orElse(0L);
}
private String orderClause(Sort sort) {
return sort.columns().stream()
.map(c -> c.column() + (c.asc() ? " ASC" : " DESC"))
.collect(java.util.stream.Collectors.joining(", "));
}
@FunctionalInterface
public interface SqlBinder {
void bind(PreparedStatement ps) throws SQLException;
}
@FunctionalInterface
public interface SqlMapper<R> {
R map(ResultSet rs) throws SQLException;
}
}
flash-extensions/flash-ext-data-jdbc/src/main/java/dev/relism/flash/ext/data/jdbc/JdbcSpecContext.java
+28
View File
@@ -0,0 +1,28 @@
package dev.relism.flash.ext.data.jdbc;
import dev.relism.flash.ext.data.core.SpecContext;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
final class JdbcSpecContext implements SpecContext {
private final List<Object> params = new ArrayList<>();
@Override
public String bind(Object value) {
params.add(value);
return "?";
}
void applyParameters(PreparedStatement ps) throws SQLException {
for (int i = 0; i < params.size(); i++) {
ps.setObject(i + 1, params.get(i));
}
}
int size() {
return params.size();
}
}
flash-extensions/flash-ext-data-jdbc/src/main/java/dev/relism/flash/ext/data/jdbc/JdbcTxManager.java
+69 -7
View File
@@ -9,6 +9,7 @@ import java.util.Objects;
public class JdbcTxManager implements TxManager {
private static final TxResourceKey JDBC_STATUS_KEY = TxResourceKey.of("jdbc.tx.status");
private static final TxResourceKey JDBC_SUSPENDED_KEY = TxResourceKey.of("jdbc.tx.suspended");
private final DataSource ds;
@@ -23,22 +24,27 @@ public class JdbcTxManager implements TxManager {
? joinExisting(definition)
: beginNew(definition);
case REQUIRES_NEW -> beginNew(definition);
case SUPPORTS -> ResourceRegistry.isBound(JDBC_STATUS_KEY)
? joinExisting(definition)
: noOp(definition);
case MANDATORY -> {
if (!ResourceRegistry.isBound(JDBC_STATUS_KEY))
throw new IllegalStateException("MANDATORY: no active transaction");
yield joinExisting(definition);
}
case NOT_SUPPORTED -> throw new UnsupportedOperationException("NOT_SUPPORTED is not implemented");
case NOT_SUPPORTED -> {
JdbcTxStatus suspended = suspendIfNeeded();
yield noOp(definition, suspended);
}
};
}
private TxStatus beginNew(TxDefinition definition) {
Connection conn = null;
JdbcTxStatus suspended = suspendIfNeeded();
boolean bound = false;
try {
JdbcTxStatus suspended = ResourceRegistry.getOrNull(JDBC_STATUS_KEY, JdbcTxStatus.class);
if (suspended != null) {
ResourceRegistry.unbind(JDBC_STATUS_KEY);
}
Connection conn = ds.getConnection();
conn = ds.getConnection();
conn.setAutoCommit(false);
if (definition.readOnly()) conn.setReadOnly(true);
if (definition.isolation() != TransactionIsolation.DEFAULT) {
@@ -52,9 +58,16 @@ public class JdbcTxManager implements TxManager {
new JdbcTxStatus.RollbackMarker()
);
ResourceRegistry.bind(JDBC_STATUS_KEY, status);
bound = true;
return status;
} catch (SQLException e) {
silentClose(conn);
throw new TxException(e);
} finally {
if (!bound && suspended != null) {
ResourceRegistry.unbind(JDBC_SUSPENDED_KEY);
ResourceRegistry.bind(JDBC_STATUS_KEY, suspended);
}
}
}
@@ -72,10 +85,29 @@ public class JdbcTxManager implements TxManager {
);
}
private TxStatus noOp(TxDefinition definition) {
return noOp(definition, null);
}
private TxStatus noOp(TxDefinition definition, JdbcTxStatus suspended) {
return new JdbcTxStatus(null, false, definition.readOnly(), suspended, new JdbcTxStatus.RollbackMarker());
}
private JdbcTxStatus suspendIfNeeded() {
JdbcTxStatus suspended = ResourceRegistry.getOrNull(JDBC_STATUS_KEY, JdbcTxStatus.class);
if (suspended != null) {
ResourceRegistry.unbind(JDBC_STATUS_KEY);
ResourceRegistry.bind(JDBC_SUSPENDED_KEY, suspended);
}
return suspended;
}
@Override
public void commit(TxStatus status) {
JdbcTxStatus s = (JdbcTxStatus) status;
if (!s.isNewTransaction()) {
resumeIfNeeded(s);
cleanupIfIdle();
return;
}
try {
@@ -90,6 +122,7 @@ public class JdbcTxManager implements TxManager {
throw new TxException(e);
} finally {
cleanupAndResume(s);
cleanupIfIdle();
}
}
@@ -98,6 +131,8 @@ public class JdbcTxManager implements TxManager {
JdbcTxStatus s = (JdbcTxStatus) status;
if (!s.isNewTransaction()) {
s.markRollbackOnly();
resumeIfNeeded(s);
cleanupIfIdle();
return;
}
try {
@@ -107,18 +142,45 @@ public class JdbcTxManager implements TxManager {
throw new TxException(e);
} finally {
cleanupAndResume(s);
cleanupIfIdle();
}
}
private void cleanupAndResume(JdbcTxStatus status) {
ResourceRegistry.unbind(JDBC_STATUS_KEY);
try {
status.connection().close();
if (status.connection() != null) {
status.connection().close();
}
} catch (SQLException ignored) {
}
resumeIfNeeded(status);
}
private void cleanupIfIdle() {
if (!ResourceRegistry.isBound(JDBC_STATUS_KEY) && !ResourceRegistry.isBound(JDBC_SUSPENDED_KEY)) {
ResourceRegistry.cleanup();
}
}
private void resumeIfNeeded(JdbcTxStatus status) {
JdbcTxStatus suspended = status.suspended();
if (suspended == null) {
suspended = ResourceRegistry.getOrNull(JDBC_SUSPENDED_KEY, JdbcTxStatus.class);
}
if (suspended != null) {
ResourceRegistry.unbind(JDBC_SUSPENDED_KEY);
ResourceRegistry.bind(JDBC_STATUS_KEY, suspended);
}
}
private void silentClose(Connection connection) {
if (connection == null) {
return;
}
try {
connection.close();
} catch (SQLException ignored) {
}
}
}
flash-extensions/flash-ext-data-jdbc/src/main/java/dev/relism/flash/ext/data/jdbc/JdbcTxStatus.java
+3
View File
@@ -37,6 +37,9 @@ class JdbcTxStatus implements TxStatus {
@Override
public <R> R resource(Class<R> type) {
if (connection == null) {
throw new IllegalStateException("No connection bound to this transaction status");
}
return type.cast(connection);
}