lxy/redkale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Browse Source
This commit is contained in:
Redkale
2019-01-12 11:44:50 +08:00
parent bb09aea8bb
commit 847f81374b
5 changed files with 291 additions and 291 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/org/redkale/source/DataCacheListener.java
View File

@@ -20,5 +20,5 @@ public interface DataCacheListener {
public <T> int updateCache(Class<T> clazz, T... entitys);
public <T> int deleteCache(Class<T> clazz, Serializable... ids);
public <T> int deleteCache(Class<T> clazz, Serializable... pks);
}

28
src/org/redkale/source/DataJdbcSource.java
View File

@@ -50,18 +50,18 @@ public class DataJdbcSource extends DataSqlSource<Connection> {
}
@Override
protected <T> CompletableFuture<Integer> insertDB(EntityInfo<T> info, T... values) {
protected <T> CompletableFuture<Integer> insertDB(EntityInfo<T> info, T... entitys) {
Connection conn = null;
try {
int c = 0;
conn = writePool.poll();
final String sql = info.getInsertPrepareSQL(values[0]);
final String sql = info.getInsertPrepareSQL(entitys[0]);
final Class primaryType = info.getPrimary().type();
final Attribute primary = info.getPrimary();
Attribute<T, Serializable>[] attrs = info.insertAttributes;
conn.setReadOnly(false);
conn.setAutoCommit(true);
PreparedStatement prestmt = createInsertPreparedStatement(conn, sql, info, values);
PreparedStatement prestmt = createInsertPreparedStatement(conn, sql, info, entitys);
try {
int[] cs = prestmt.executeBatch();
int c1 = 0;
@@ -73,7 +73,7 @@ public class DataJdbcSource extends DataSqlSource<Connection> {
if (info.tableStrategy == null || !info.isTableNotExist(se)) throw se;
synchronized (info.tables) {
final String oldTable = info.table;
final String newTable = info.getTable(values[0]);
final String newTable = info.getTable(entitys[0]);
if (!info.tables.contains(newTable)) {
try {
Statement st = conn.createStatement();
@@ -105,7 +105,7 @@ public class DataJdbcSource extends DataSqlSource<Connection> {
}
}
prestmt.close();
prestmt = createInsertPreparedStatement(conn, sql, info, values);
prestmt = createInsertPreparedStatement(conn, sql, info, entitys);
int[] cs = prestmt.executeBatch();
int c1 = 0;
for (int cc : cs) {
@@ -118,11 +118,11 @@ public class DataJdbcSource extends DataSqlSource<Connection> {
int i = -1;
while (set.next()) {
if (primaryType == int.class) {
primary.set(values[++i], set.getInt(1));
primary.set(entitys[++i], set.getInt(1));
} else if (primaryType == long.class) {
primary.set(values[++i], set.getLong(1));
primary.set(entitys[++i], set.getLong(1));
} else {
primary.set(values[++i], set.getObject(1));
primary.set(entitys[++i], set.getObject(1));
}
}
set.close();
@@ -131,7 +131,7 @@ public class DataJdbcSource extends DataSqlSource<Connection> {
//------------------------------------------------------------
if (info.isLoggable(logger, Level.FINEST)) { //打印调试信息
char[] sqlchars = sql.toCharArray();
for (final T value : values) {
for (final T value : entitys) {
//-----------------------------
StringBuilder sb = new StringBuilder(128);
int i = 0;
@@ -162,11 +162,11 @@ public class DataJdbcSource extends DataSqlSource<Connection> {
}
protected <T> PreparedStatement createInsertPreparedStatement(final Connection conn, final String sql,
final EntityInfo<T> info, T... values) throws SQLException {
final EntityInfo<T> info, T... entitys) throws SQLException {
Attribute<T, Serializable>[] attrs = info.insertAttributes;
final PreparedStatement prestmt = info.autoGenerated ? conn.prepareStatement(sql, Statement.RETURN_GENERATED_KEYS) : conn.prepareStatement(sql);
for (final T value : values) {
for (final T value : entitys) {
if (info.autouuid) info.createPrimaryValue(value);
batchStatementParameters(conn, prestmt, info, attrs, value);
prestmt.addBatch();
@@ -259,19 +259,19 @@ public class DataJdbcSource extends DataSqlSource<Connection> {
}
@Override
protected <T> CompletableFuture<Integer> updateDB(EntityInfo<T> info, T... values) {
protected <T> CompletableFuture<Integer> updateDB(EntityInfo<T> info, T... entitys) {
Connection conn = null;
try {
conn = writePool.poll();
conn.setReadOnly(false);
conn.setAutoCommit(true);
final String updateSQL = info.getUpdatePrepareSQL(values[0]);
final String updateSQL = info.getUpdatePrepareSQL(entitys[0]);
final PreparedStatement prestmt = conn.prepareStatement(updateSQL);
Attribute<T, Serializable>[] attrs = info.updateAttributes;
final boolean debugfinest = info.isLoggable(logger, Level.FINEST);
char[] sqlchars = debugfinest ? updateSQL.toCharArray() : null;
final Attribute<T, Serializable> primary = info.getPrimary();
for (final T value : values) {
for (final T value : entitys) {
int k = batchStatementParameters(conn, prestmt, info, attrs, value);
prestmt.setObject(++k, primary.get(value));
prestmt.addBatch();//------------------------------------------------------------

4
src/org/redkale/source/DataMemorySource.java
View File

@@ -84,7 +84,7 @@ public class DataMemorySource extends DataSqlSource<Void> {
}
@Override
protected <T> CompletableFuture<Integer> insertDB(EntityInfo<T> info, T... values) {
protected <T> CompletableFuture<Integer> insertDB(EntityInfo<T> info, T... entitys) {
return CompletableFuture.completedFuture(0);
}
@@ -104,7 +104,7 @@ public class DataMemorySource extends DataSqlSource<Void> {
}
@Override
protected <T> CompletableFuture<Integer> updateDB(EntityInfo<T> info, T... values) {
protected <T> CompletableFuture<Integer> updateDB(EntityInfo<T> info, T... entitys) {
return CompletableFuture.completedFuture(0);
}

426
src/org/redkale/source/DataSqlSource.java
View File

@@ -128,7 +128,7 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
protected abstract PoolSource<DBChannel> createPoolSource(DataSource source, String rwtype, ArrayBlockingQueue queue, Semaphore semaphore, Properties prop);
//插入纪录
protected abstract <T> CompletableFuture<Integer> insertDB(final EntityInfo<T> info, T... values);
protected abstract <T> CompletableFuture<Integer> insertDB(final EntityInfo<T> info, T... entitys);
//删除记录
protected abstract <T> CompletableFuture<Integer> deleteDB(final EntityInfo<T> info, Flipper flipper, final String sql);
@@ -140,7 +140,7 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
protected abstract <T> CompletableFuture<Integer> dropTableDB(final EntityInfo<T> info, final String sql);
//更新纪录
protected abstract <T> CompletableFuture<Integer> updateDB(final EntityInfo<T> info, T... values);
protected abstract <T> CompletableFuture<Integer> updateDB(final EntityInfo<T> info, T... entitys);
//更新纪录
protected abstract <T> CompletableFuture<Integer> updateDB(final EntityInfo<T> info, Flipper flipper, final String sql, final boolean prepared, Object... params);
@@ -256,10 +256,10 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
}
////检查对象是否都是同一个Entity类
protected <T> CompletableFuture checkEntity(String action, boolean async, T... values) {
if (values.length < 1) return null;
protected <T> CompletableFuture checkEntity(String action, boolean async, T... entitys) {
if (entitys.length < 1) return null;
Class clazz = null;
for (T val : values) {
for (T val : entitys) {
if (clazz == null) {
clazz = val.getClass();
continue;
@@ -280,80 +280,80 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
/**
* 新增对象， 必须是Entity对象
*
* @param <T> Entity类泛型
* @param values Entity对象
* @param <T> Entity类泛型
* @param entitys Entity对象
*
* @return 影响的记录条数
*/
@Override
public <T> int insert(@RpcCall(DataCallArrayAttribute.class) T... values) {
if (values.length == 0) return 0;
checkEntity("insert", false, values);
final EntityInfo<T> info = loadEntityInfo((Class<T>) values[0].getClass());
public <T> int insert(@RpcCall(DataCallArrayAttribute.class) T... entitys) {
if (entitys.length == 0) return 0;
checkEntity("insert", false, entitys);
final EntityInfo<T> info = loadEntityInfo((Class<T>) entitys[0].getClass());
if (info.autouuid) {
for (T value : values) {
for (T value : entitys) {
info.createPrimaryValue(value);
}
}
if (isOnlyCache(info)) return insertCache(info, values);
return insertDB(info, values).whenComplete((rs, t) -> {
if (isOnlyCache(info)) return insertCache(info, entitys);
return insertDB(info, entitys).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
insertCache(info, values);
insertCache(info, entitys);
}
}).join();
}
@Override
public <T> CompletableFuture<Integer> insertAsync(@RpcCall(DataCallArrayAttribute.class) T... values) {
if (values.length == 0) return CompletableFuture.completedFuture(0);
CompletableFuture future = checkEntity("insert", true, values);
public <T> CompletableFuture<Integer> insertAsync(@RpcCall(DataCallArrayAttribute.class) T... entitys) {
if (entitys.length == 0) return CompletableFuture.completedFuture(0);
CompletableFuture future = checkEntity("insert", true, entitys);
if (future != null) return future;
final EntityInfo<T> info = loadEntityInfo((Class<T>) values[0].getClass());
final EntityInfo<T> info = loadEntityInfo((Class<T>) entitys[0].getClass());
if (info.autouuid) {
for (T value : values) {
for (T value : entitys) {
info.createPrimaryValue(value);
}
}
if (isOnlyCache(info)) {
return CompletableFuture.supplyAsync(() -> insertCache(info, values), getExecutor());
return CompletableFuture.supplyAsync(() -> insertCache(info, entitys), getExecutor());
}
if (isAsync()) return insertDB(info, values).whenComplete((rs, t) -> {
if (isAsync()) return insertDB(info, entitys).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
insertCache(info, values);
insertCache(info, entitys);
}
});
return CompletableFuture.supplyAsync(() -> insertDB(info, values).join(), getExecutor()).whenComplete((rs, t) -> {
return CompletableFuture.supplyAsync(() -> insertDB(info, entitys).join(), getExecutor()).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
insertCache(info, values);
insertCache(info, entitys);
}
});
}
protected <T> int insertCache(final EntityInfo<T> info, T... values) {
protected <T> int insertCache(final EntityInfo<T> info, T... entitys) {
final EntityCache<T> cache = info.getCache();
if (cache == null) return 0;
int c = 0;
for (final T value : values) {
for (final T value : entitys) {
c += cache.insert(value);
}
if (cacheListener != null) cacheListener.insertCache(info.getType(), values);
if (cacheListener != null) cacheListener.insertCache(info.getType(), entitys);
return c;
}
@Override
public <T> int insertCache(Class<T> clazz, T... values) {
if (values.length == 0) return 0;
public <T> int insertCache(Class<T> clazz, T... entitys) {
if (entitys.length == 0) return 0;
final EntityInfo<T> info = loadEntityInfo(clazz);
final EntityCache<T> cache = info.getCache();
if (cache == null) return -1;
int c = 0;
for (T value : values) {
for (T value : entitys) {
c += cache.insert(value);
}
return c;
@@ -363,75 +363,75 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
/**
* 删除对象， 必须是Entity对象
*
* @param <T> Entity类泛型
* @param values Entity对象
* @param <T> Entity类泛型
* @param entitys Entity对象
*
* @return 删除的数据条数
*/
@Override
public <T> int delete(T... values) {
if (values.length == 0) return -1;
checkEntity("delete", false, values);
final Class<T> clazz = (Class<T>) values[0].getClass();
public <T> int delete(T... entitys) {
if (entitys.length == 0) return -1;
checkEntity("delete", false, entitys);
final Class<T> clazz = (Class<T>) entitys[0].getClass();
final EntityInfo<T> info = loadEntityInfo(clazz);
final Attribute primary = info.getPrimary();
Serializable[] ids = new Serializable[values.length];
Serializable[] ids = new Serializable[entitys.length];
int i = 0;
for (final T value : values) {
for (final T value : entitys) {
ids[i++] = (Serializable) primary.get(value);
}
return delete(clazz, ids);
}
@Override
public <T> CompletableFuture<Integer> deleteAsync(final T... values) {
if (values.length == 0) return CompletableFuture.completedFuture(-1);
CompletableFuture future = checkEntity("delete", true, values);
public <T> CompletableFuture<Integer> deleteAsync(final T... entitys) {
if (entitys.length == 0) return CompletableFuture.completedFuture(-1);
CompletableFuture future = checkEntity("delete", true, entitys);
if (future != null) return future;
final Class<T> clazz = (Class<T>) values[0].getClass();
final Class<T> clazz = (Class<T>) entitys[0].getClass();
final EntityInfo<T> info = loadEntityInfo(clazz);
final Attribute primary = info.getPrimary();
Serializable[] ids = new Serializable[values.length];
Serializable[] ids = new Serializable[entitys.length];
int i = 0;
for (final T value : values) {
for (final T value : entitys) {
ids[i++] = (Serializable) primary.get(value);
}
return deleteAsync(clazz, ids);
}
@Override
public <T> int delete(Class<T> clazz, Serializable... ids) {
if (ids.length == 0) return -1;
public <T> int delete(Class<T> clazz, Serializable... pks) {
if (pks.length == 0) return -1;
final EntityInfo<T> info = loadEntityInfo(clazz);
if (isOnlyCache(info)) return deleteCache(info, -1, ids);
return deleteCompose(info, ids).whenComplete((rs, t) -> {
if (isOnlyCache(info)) return deleteCache(info, -1, pks);
return deleteCompose(info, pks).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
deleteCache(info, rs, ids);
deleteCache(info, rs, pks);
}
}).join();
}
@Override
public <T> CompletableFuture<Integer> deleteAsync(final Class<T> clazz, final Serializable... ids) {
if (ids.length == 0) return CompletableFuture.completedFuture(-1);
public <T> CompletableFuture<Integer> deleteAsync(final Class<T> clazz, final Serializable... pks) {
if (pks.length == 0) return CompletableFuture.completedFuture(-1);
final EntityInfo<T> info = loadEntityInfo(clazz);
if (isOnlyCache(info)) {
return CompletableFuture.supplyAsync(() -> deleteCache(info, -1, ids), getExecutor());
return CompletableFuture.supplyAsync(() -> deleteCache(info, -1, pks), getExecutor());
}
if (isAsync()) return deleteCompose(info, ids).whenComplete((rs, t) -> {
if (isAsync()) return deleteCompose(info, pks).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
deleteCache(info, rs, ids);
deleteCache(info, rs, pks);
}
});
return CompletableFuture.supplyAsync(() -> deleteCompose(info, ids).join(), getExecutor()).whenComplete((rs, t) -> {
return CompletableFuture.supplyAsync(() -> deleteCompose(info, pks).join(), getExecutor()).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
deleteCache(info, rs, ids);
deleteCache(info, rs, pks);
}
});
}
@@ -481,15 +481,15 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
});
}
protected <T> CompletableFuture<Integer> deleteCompose(final EntityInfo<T> info, final Serializable... ids) {
if (ids.length == 1) {
String sql = "DELETE FROM " + info.getTable(ids[0]) + " WHERE " + info.getPrimarySQLColumn() + " = " + FilterNode.formatToString(info.getSQLValue(info.getPrimarySQLColumn(), ids[0]));
protected <T> CompletableFuture<Integer> deleteCompose(final EntityInfo<T> info, final Serializable... pks) {
if (pks.length == 1) {
String sql = "DELETE FROM " + info.getTable(pks[0]) + " WHERE " + info.getPrimarySQLColumn() + " = " + FilterNode.formatToString(info.getSQLValue(info.getPrimarySQLColumn(), pks[0]));
return deleteDB(info, null, sql);
}
String sql = "DELETE FROM " + info.getTable(ids[0]) + " WHERE " + info.getPrimarySQLColumn() + " IN (";
for (int i = 0; i < ids.length; i++) {
String sql = "DELETE FROM " + info.getTable(pks[0]) + " WHERE " + info.getPrimarySQLColumn() + " IN (";
for (int i = 0; i < pks.length; i++) {
if (i > 0) sql += ',';
sql += FilterNode.formatToString(info.getSQLValue(info.getPrimarySQLColumn(), ids[i]));
sql += FilterNode.formatToString(info.getSQLValue(info.getPrimarySQLColumn(), pks[i]));
}
sql += ")";
if (info.isLoggable(logger, Level.FINEST, sql)) logger.finest(info.getType().getSimpleName() + " delete sql=" + sql);
@@ -640,25 +640,25 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
return count >= 0 ? count : (ids == null ? 0 : ids.length);
}
protected <T> int deleteCache(final EntityInfo<T> info, int count, Serializable... keys) {
protected <T> int deleteCache(final EntityInfo<T> info, int count, Serializable... pks) {
final EntityCache<T> cache = info.getCache();
if (cache == null) return -1;
int c = 0;
for (Serializable key : keys) {
for (Serializable key : pks) {
c += cache.delete(key);
}
if (cacheListener != null) cacheListener.deleteCache(info.getType(), keys);
if (cacheListener != null) cacheListener.deleteCache(info.getType(), pks);
return count >= 0 ? count : c;
}
@Override
public <T> int deleteCache(Class<T> clazz, Serializable... ids) {
if (ids.length == 0) return 0;
public <T> int deleteCache(Class<T> clazz, Serializable... pks) {
if (pks.length == 0) return 0;
final EntityInfo<T> info = loadEntityInfo(clazz);
final EntityCache<T> cache = info.getCache();
if (cache == null) return -1;
int c = 0;
for (Serializable id : ids) {
for (Serializable id : pks) {
c += cache.delete(id);
}
return c;
@@ -679,47 +679,47 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
/**
* 更新对象， 必须是Entity对象
*
* @param <T> Entity类泛型
* @param values Entity对象
* @param <T> Entity类泛型
* @param entitys Entity对象
*
* @return 更新的数据条数
*/
@Override
public <T> int update(T... values) {
if (values.length == 0) return -1;
checkEntity("update", false, values);
final Class<T> clazz = (Class<T>) values[0].getClass();
public <T> int update(T... entitys) {
if (entitys.length == 0) return -1;
checkEntity("update", false, entitys);
final Class<T> clazz = (Class<T>) entitys[0].getClass();
final EntityInfo<T> info = loadEntityInfo(clazz);
if (isOnlyCache(info)) return updateCache(info, -1, values);
return updateDB(info, values).whenComplete((rs, t) -> {
if (isOnlyCache(info)) return updateCache(info, -1, entitys);
return updateDB(info, entitys).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
updateCache(info, rs, values);
updateCache(info, rs, entitys);
}
}).join();
}
@Override
public <T> CompletableFuture<Integer> updateAsync(final T... values) {
if (values.length == 0) return CompletableFuture.completedFuture(-1);
CompletableFuture future = checkEntity("update", true, values);
public <T> CompletableFuture<Integer> updateAsync(final T... entitys) {
if (entitys.length == 0) return CompletableFuture.completedFuture(-1);
CompletableFuture future = checkEntity("update", true, entitys);
if (future != null) return future;
final Class<T> clazz = (Class<T>) values[0].getClass();
final Class<T> clazz = (Class<T>) entitys[0].getClass();
final EntityInfo<T> info = loadEntityInfo(clazz);
if (isOnlyCache(info)) return CompletableFuture.supplyAsync(() -> updateCache(info, -1, values), getExecutor());
if (isAsync()) return updateDB(info, values).whenComplete((rs, t) -> {
if (isOnlyCache(info)) return CompletableFuture.supplyAsync(() -> updateCache(info, -1, entitys), getExecutor());
if (isAsync()) return updateDB(info, entitys).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
updateCache(info, rs, values);
updateCache(info, rs, entitys);
}
});
return CompletableFuture.supplyAsync(() -> updateDB(info, values).join(), getExecutor()).whenComplete((rs, t) -> {
return CompletableFuture.supplyAsync(() -> updateDB(info, entitys).join(), getExecutor()).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
updateCache(info, rs, values);
updateCache(info, rs, entitys);
}
});
}
@@ -729,54 +729,54 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
*
* @param <T> Entity类的泛型
* @param clazz Entity类
* @param id 主键值
* @param pk 主键值
* @param column 过滤字段名
* @param value 过滤字段值
* @param colval 过滤字段值
*
* @return 更新的数据条数
*/
@Override
public <T> int updateColumn(Class<T> clazz, Serializable id, String column, Serializable value) {
public <T> int updateColumn(Class<T> clazz, Serializable pk, String column, Serializable colval) {
final EntityInfo<T> info = loadEntityInfo(clazz);
if (isOnlyCache(info)) return updateCache(info, -1, id, column, value);
return updateColumnCompose(info, id, column, value).whenComplete((rs, t) -> {
if (isOnlyCache(info)) return updateCache(info, -1, pk, column, colval);
return updateColumnCompose(info, pk, column, colval).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
updateCache(info, rs, id, column, value);
updateCache(info, rs, pk, column, colval);
}
}).join();
}
@Override
public <T> CompletableFuture<Integer> updateColumnAsync(final Class<T> clazz, final Serializable id, final String column, final Serializable value) {
public <T> CompletableFuture<Integer> updateColumnAsync(final Class<T> clazz, final Serializable pk, final String column, final Serializable colval) {
final EntityInfo<T> info = loadEntityInfo(clazz);
if (isOnlyCache(info)) {
return CompletableFuture.supplyAsync(() -> updateCache(info, -1, id, column, value), getExecutor());
return CompletableFuture.supplyAsync(() -> updateCache(info, -1, pk, column, colval), getExecutor());
}
if (isAsync()) return updateColumnCompose(info, id, column, value).whenComplete((rs, t) -> {
if (isAsync()) return updateColumnCompose(info, pk, column, colval).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
updateCache(info, rs, id, column, value);
updateCache(info, rs, pk, column, colval);
}
});
return CompletableFuture.supplyAsync(() -> updateColumnCompose(info, id, column, value).join(), getExecutor()).whenComplete((rs, t) -> {
return CompletableFuture.supplyAsync(() -> updateColumnCompose(info, pk, column, colval).join(), getExecutor()).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
updateCache(info, rs, id, column, value);
updateCache(info, rs, pk, column, colval);
}
});
}
protected <T> CompletableFuture<Integer> updateColumnCompose(final EntityInfo<T> info, Serializable id, String column, final Serializable value) {
if (value instanceof byte[]) {
String sql = "UPDATE " + info.getTable(id) + " SET " + info.getSQLColumn(null, column) + " = " + prepareParamSign(1) + " WHERE " + info.getPrimarySQLColumn() + " = " + FilterNode.formatToString(info.getSQLValue(info.getPrimarySQLColumn(), id));
return updateDB(info, null, sql, true, value);
protected <T> CompletableFuture<Integer> updateColumnCompose(final EntityInfo<T> info, Serializable pk, String column, final Serializable colval) {
if (colval instanceof byte[]) {
String sql = "UPDATE " + info.getTable(pk) + " SET " + info.getSQLColumn(null, column) + " = " + prepareParamSign(1) + " WHERE " + info.getPrimarySQLColumn() + " = " + FilterNode.formatToString(info.getSQLValue(info.getPrimarySQLColumn(), pk));
return updateDB(info, null, sql, true, colval);
} else {
String sql = "UPDATE " + info.getTable(id) + " SET " + info.getSQLColumn(null, column) + " = "
+ info.formatToString(info.getSQLValue(column, value)) + " WHERE " + info.getPrimarySQLColumn() + " = " + FilterNode.formatToString(info.getSQLValue(info.getPrimarySQLColumn(), id));
String sql = "UPDATE " + info.getTable(pk) + " SET " + info.getSQLColumn(null, column) + " = "
+ info.formatToString(info.getSQLValue(column, colval)) + " WHERE " + info.getPrimarySQLColumn() + " = " + FilterNode.formatToString(info.getSQLValue(info.getPrimarySQLColumn(), pk));
return updateDB(info, null, sql, false);
}
}
@@ -787,47 +787,47 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
* @param <T> Entity类的泛型
* @param clazz Entity类
* @param column 过滤字段名
* @param value 过滤字段值
* @param colval 过滤字段值
* @param node 过滤node 不能为null
*
* @return 更新的数据条数
*/
@Override
public <T> int updateColumn(Class<T> clazz, String column, Serializable value, FilterNode node) {
public <T> int updateColumn(Class<T> clazz, String column, Serializable colval, FilterNode node) {
final EntityInfo<T> info = loadEntityInfo(clazz);
if (isOnlyCache(info)) return updateCache(info, -1, column, value, node);
return DataSqlSource.this.updateColumnCompose(info, column, value, node).whenComplete((rs, t) -> {
if (isOnlyCache(info)) return updateCache(info, -1, column, colval, node);
return DataSqlSource.this.updateColumnCompose(info, column, colval, node).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
updateCache(info, rs, column, value, node);
updateCache(info, rs, column, colval, node);
}
}).join();
}
@Override
public <T> CompletableFuture<Integer> updateColumnAsync(final Class<T> clazz, final String column, final Serializable value, final FilterNode node) {
public <T> CompletableFuture<Integer> updateColumnAsync(final Class<T> clazz, final String column, final Serializable colval, final FilterNode node) {
final EntityInfo<T> info = loadEntityInfo(clazz);
if (isOnlyCache(info)) {
return CompletableFuture.supplyAsync(() -> updateCache(info, -1, column, value, node), getExecutor());
return CompletableFuture.supplyAsync(() -> updateCache(info, -1, column, colval, node), getExecutor());
}
if (isAsync()) return DataSqlSource.this.updateColumnCompose(info, column, value, node).whenComplete((rs, t) -> {
if (isAsync()) return DataSqlSource.this.updateColumnCompose(info, column, colval, node).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
updateCache(info, rs, column, value, node);
updateCache(info, rs, column, colval, node);
}
});
return CompletableFuture.supplyAsync(() -> DataSqlSource.this.updateColumnCompose(info, column, value, node).join(), getExecutor()).whenComplete((rs, t) -> {
return CompletableFuture.supplyAsync(() -> DataSqlSource.this.updateColumnCompose(info, column, colval, node).join(), getExecutor()).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
updateCache(info, rs, column, value, node);
updateCache(info, rs, column, colval, node);
}
});
}
protected <T> CompletableFuture<Integer> updateColumnCompose(final EntityInfo<T> info, final String column, final Serializable value, final FilterNode node) {
protected <T> CompletableFuture<Integer> updateColumnCompose(final EntityInfo<T> info, final String column, final Serializable colval, final FilterNode node) {
Map<Class, String> joinTabalis = node.getJoinTabalis();
CharSequence join = node.createSQLJoin(this, true, joinTabalis, new HashSet<>(), info);
CharSequence where = node.createSQLExpress(info, joinTabalis);
@@ -840,15 +840,15 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
join2 = multisplit('{', '}', " AND ", new StringBuilder(), joinstr, 0);
}
String alias = "postgresql".equals(writePool.dbtype) ? null : "a"; //postgresql的BUG， UPDATE的SET中不能含别名
if (value instanceof byte[]) {
if (colval instanceof byte[]) {
String sql = "UPDATE " + info.getTable(node) + " a " + (join1 == null ? "" : (", " + join1))
+ " SET " + info.getSQLColumn(alias, column) + " = " + prepareParamSign(1)
+ ((where == null || where.length() == 0) ? (join2 == null ? "" : (" WHERE " + join2))
: (" WHERE " + where + (join2 == null ? "" : (" AND " + join2))));
return updateDB(info, null, sql, true, value);
return updateDB(info, null, sql, true, colval);
} else {
String sql = "UPDATE " + info.getTable(node) + " a " + (join1 == null ? "" : (", " + join1))
+ " SET " + info.getSQLColumn(alias, column) + " = " + info.formatToString(value)
+ " SET " + info.getSQLColumn(alias, column) + " = " + info.formatToString(colval)
+ ((where == null || where.length() == 0) ? (join2 == null ? "" : (" WHERE " + join2))
: (" WHERE " + where + (join2 == null ? "" : (" AND " + join2))));
return updateDB(info, null, sql, false);
@@ -860,49 +860,49 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
*
* @param <T> Entity类的泛型
* @param clazz Entity类
* @param id 主键值
* @param pk 主键值
* @param values 字段值
*
* @return 更新的数据条数
*/
@Override
public <T> int updateColumn(final Class<T> clazz, final Serializable id, final ColumnValue... values) {
public <T> int updateColumn(final Class<T> clazz, final Serializable pk, final ColumnValue... values) {
if (values == null || values.length < 1) return -1;
final EntityInfo<T> info = loadEntityInfo(clazz);
if (isOnlyCache(info)) return updateCache(info, -1, id, values);
return DataSqlSource.this.updateColumnCompose(info, id, values).whenComplete((rs, t) -> {
if (isOnlyCache(info)) return updateCache(info, -1, pk, values);
return DataSqlSource.this.updateColumnCompose(info, pk, values).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
updateCache(info, rs, id, values);
updateCache(info, rs, pk, values);
}
}).join();
}
@Override
public <T> CompletableFuture<Integer> updateColumnAsync(final Class<T> clazz, final Serializable id, final ColumnValue... values) {
public <T> CompletableFuture<Integer> updateColumnAsync(final Class<T> clazz, final Serializable pk, final ColumnValue... values) {
if (values == null || values.length < 1) return CompletableFuture.completedFuture(-1);
final EntityInfo<T> info = loadEntityInfo(clazz);
if (isOnlyCache(info)) {
return CompletableFuture.supplyAsync(() -> updateCache(info, -1, id, values), getExecutor());
return CompletableFuture.supplyAsync(() -> updateCache(info, -1, pk, values), getExecutor());
}
if (isAsync()) return DataSqlSource.this.updateColumnCompose(info, id, values).whenComplete((rs, t) -> {
if (isAsync()) return DataSqlSource.this.updateColumnCompose(info, pk, values).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
updateCache(info, rs, id, values);
updateCache(info, rs, pk, values);
}
});
return CompletableFuture.supplyAsync(() -> DataSqlSource.this.updateColumnCompose(info, id, values).join(), getExecutor()).whenComplete((rs, t) -> {
return CompletableFuture.supplyAsync(() -> DataSqlSource.this.updateColumnCompose(info, pk, values).join(), getExecutor()).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
updateCache(info, rs, id, values);
updateCache(info, rs, pk, values);
}
});
}
protected <T> CompletableFuture<Integer> updateColumnCompose(final EntityInfo<T> info, final Serializable id, final ColumnValue... values) {
protected <T> CompletableFuture<Integer> updateColumnCompose(final EntityInfo<T> info, final Serializable pk, final ColumnValue... values) {
StringBuilder setsql = new StringBuilder();
List<byte[]> blobs = null;
int index = 0;
@@ -921,7 +921,7 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
}
}
if (setsql.length() < 1) return CompletableFuture.completedFuture(0);
String sql = "UPDATE " + info.getTable(id) + " SET " + setsql + " WHERE " + info.getPrimarySQLColumn() + " = " + FilterNode.formatToString(info.getSQLValue(info.getPrimarySQLColumn(), id));
String sql = "UPDATE " + info.getTable(pk) + " SET " + setsql + " WHERE " + info.getPrimarySQLColumn() + " = " + FilterNode.formatToString(info.getSQLValue(info.getPrimarySQLColumn(), pk));
if (blobs == null) return updateDB(info, null, sql, false);
return updateDB(info, null, sql, true, blobs.toArray());
}
@@ -1022,104 +1022,104 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
}
@Override
public <T> int updateColumn(final T bean, final String... columns) {
return updateColumn(bean, SelectColumn.includes(columns));
public <T> int updateColumn(final T entity, final String... columns) {
return updateColumn(entity, SelectColumn.includes(columns));
}
@Override
public <T> CompletableFuture<Integer> updateColumnAsync(final T bean, final String... columns) {
return updateColumnAsync(bean, SelectColumn.includes(columns));
public <T> CompletableFuture<Integer> updateColumnAsync(final T entity, final String... columns) {
return updateColumnAsync(entity, SelectColumn.includes(columns));
}
@Override
public <T> int updateColumn(final T bean, final FilterNode node, final String... columns) {
return updateColumn(bean, node, SelectColumn.includes(columns));
public <T> int updateColumn(final T entity, final FilterNode node, final String... columns) {
return updateColumn(entity, node, SelectColumn.includes(columns));
}
@Override
public <T> CompletableFuture<Integer> updateColumnAsync(final T bean, final FilterNode node, final String... columns) {
return updateColumnAsync(bean, node, SelectColumn.includes(columns));
public <T> CompletableFuture<Integer> updateColumnAsync(final T entity, final FilterNode node, final String... columns) {
return updateColumnAsync(entity, node, SelectColumn.includes(columns));
}
@Override
public <T> int updateColumn(final T bean, final SelectColumn selects) {
if (bean == null || selects == null) return -1;
Class<T> clazz = (Class) bean.getClass();
public <T> int updateColumn(final T entity, final SelectColumn selects) {
if (entity == null || selects == null) return -1;
Class<T> clazz = (Class) entity.getClass();
final EntityInfo<T> info = loadEntityInfo(clazz);
if (isOnlyCache(info)) return updateCache(info, -1, false, bean, null, selects);
return DataSqlSource.this.updateColumnCompose(info, false, bean, null, selects).whenComplete((rs, t) -> {
if (isOnlyCache(info)) return updateCache(info, -1, false, entity, null, selects);
return DataSqlSource.this.updateColumnCompose(info, false, entity, null, selects).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
updateCache(info, rs, false, bean, null, selects);
updateCache(info, rs, false, entity, null, selects);
}
}).join();
}
@Override
public <T> CompletableFuture<Integer> updateColumnAsync(final T bean, final SelectColumn selects) {
if (bean == null || selects == null) return CompletableFuture.completedFuture(-1);
Class<T> clazz = (Class) bean.getClass();
public <T> CompletableFuture<Integer> updateColumnAsync(final T entity, final SelectColumn selects) {
if (entity == null || selects == null) return CompletableFuture.completedFuture(-1);
Class<T> clazz = (Class) entity.getClass();
final EntityInfo<T> info = loadEntityInfo(clazz);
if (isOnlyCache(info)) {
return CompletableFuture.supplyAsync(() -> updateCache(info, -1, false, bean, null, selects), getExecutor());
return CompletableFuture.supplyAsync(() -> updateCache(info, -1, false, entity, null, selects), getExecutor());
}
if (isAsync()) return DataSqlSource.this.updateColumnCompose(info, false, bean, null, selects).whenComplete((rs, t) -> {
if (isAsync()) return DataSqlSource.this.updateColumnCompose(info, false, entity, null, selects).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
updateCache(info, rs, false, bean, null, selects);
updateCache(info, rs, false, entity, null, selects);
}
});
return CompletableFuture.supplyAsync(() -> DataSqlSource.this.updateColumnCompose(info, false, bean, null, selects).join(), getExecutor()).whenComplete((rs, t) -> {
return CompletableFuture.supplyAsync(() -> DataSqlSource.this.updateColumnCompose(info, false, entity, null, selects).join(), getExecutor()).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
updateCache(info, rs, false, bean, null, selects);
updateCache(info, rs, false, entity, null, selects);
}
});
}
@Override
public <T> int updateColumn(final T bean, final FilterNode node, final SelectColumn selects) {
if (bean == null || node == null || selects == null) return -1;
Class<T> clazz = (Class) bean.getClass();
public <T> int updateColumn(final T entity, final FilterNode node, final SelectColumn selects) {
if (entity == null || node == null || selects == null) return -1;
Class<T> clazz = (Class) entity.getClass();
final EntityInfo<T> info = loadEntityInfo(clazz);
if (isOnlyCache(info)) return updateCache(info, -1, true, bean, node, selects);
return DataSqlSource.this.updateColumnCompose(info, true, bean, node, selects).whenComplete((rs, t) -> {
if (isOnlyCache(info)) return updateCache(info, -1, true, entity, node, selects);
return DataSqlSource.this.updateColumnCompose(info, true, entity, node, selects).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
updateCache(info, rs, true, bean, node, selects);
updateCache(info, rs, true, entity, node, selects);
}
}).join();
}
@Override
public <T> CompletableFuture<Integer> updateColumnAsync(final T bean, final FilterNode node, final SelectColumn selects) {
if (bean == null || node == null || selects == null) return CompletableFuture.completedFuture(-1);
Class<T> clazz = (Class) bean.getClass();
public <T> CompletableFuture<Integer> updateColumnAsync(final T entity, final FilterNode node, final SelectColumn selects) {
if (entity == null || node == null || selects == null) return CompletableFuture.completedFuture(-1);
Class<T> clazz = (Class) entity.getClass();
final EntityInfo<T> info = loadEntityInfo(clazz);
if (isOnlyCache(info)) {
return CompletableFuture.supplyAsync(() -> updateCache(info, -1, true, bean, node, selects), getExecutor());
return CompletableFuture.supplyAsync(() -> updateCache(info, -1, true, entity, node, selects), getExecutor());
}
if (isAsync()) return DataSqlSource.this.updateColumnCompose(info, true, bean, node, selects).whenComplete((rs, t) -> {
if (isAsync()) return DataSqlSource.this.updateColumnCompose(info, true, entity, node, selects).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
updateCache(info, rs, true, bean, node, selects);
updateCache(info, rs, true, entity, node, selects);
}
});
return CompletableFuture.supplyAsync(() -> DataSqlSource.this.updateColumnCompose(info, true, bean, node, selects).join(), getExecutor()).whenComplete((rs, t) -> {
return CompletableFuture.supplyAsync(() -> DataSqlSource.this.updateColumnCompose(info, true, entity, node, selects).join(), getExecutor()).whenComplete((rs, t) -> {
if (t != null) {
futureCompleteConsumer.accept(rs, t);
} else {
updateCache(info, rs, true, bean, node, selects);
updateCache(info, rs, true, entity, node, selects);
}
});
}
protected <T> CompletableFuture<Integer> updateColumnCompose(final EntityInfo<T> info, final boolean neednode, final T bean, final FilterNode node, final SelectColumn selects) {
protected <T> CompletableFuture<Integer> updateColumnCompose(final EntityInfo<T> info, final boolean neednode, final T entity, final FilterNode node, final SelectColumn selects) {
StringBuilder setsql = new StringBuilder();
List<byte[]> blobs = null;
int index = 0;
@@ -1128,7 +1128,7 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
if (!selects.test(attr.field())) continue;
if (setsql.length() > 0) setsql.append(", ");
setsql.append(info.getSQLColumn(alias, attr.field()));
Serializable val = info.getFieldValue(attr, bean);
Serializable val = info.getFieldValue(attr, entity);
if (val instanceof byte[]) {
if (blobs == null) blobs = new ArrayList<>();
blobs.add((byte[]) val);
@@ -1154,14 +1154,14 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
if (blobs == null) return updateDB(info, null, sql, false);
return updateDB(info, null, sql, true, blobs.toArray());
} else {
final Serializable id = info.getSQLValue(info.getPrimary(), bean);
final Serializable id = info.getSQLValue(info.getPrimary(), entity);
String sql = "UPDATE " + info.getTable(id) + " a SET " + setsql + " WHERE " + info.getPrimarySQLColumn() + " = " + FilterNode.formatToString(id);
if (blobs == null) return updateDB(info, null, sql, false);
return updateDB(info, null, sql, true, blobs.toArray());
}
}
protected <T> int updateCache(final EntityInfo<T> info, int count, final boolean neednode, final T bean, final FilterNode node, final SelectColumn selects) {
protected <T> int updateCache(final EntityInfo<T> info, int count, final boolean neednode, final T entity, final FilterNode node, final SelectColumn selects) {
final EntityCache<T> cache = info.getCache();
if (cache == null) return count;
final List<Attribute<T, Serializable>> attrs = new ArrayList<>();
@@ -1170,11 +1170,11 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
attrs.add(attr);
}
if (neednode) {
T[] rs = cache.update(bean, attrs, node);
T[] rs = cache.update(entity, attrs, node);
if (cacheListener != null) cacheListener.updateCache(info.getType(), rs);
return count >= 0 ? count : (rs == null ? 0 : rs.length);
} else {
T rs = cache.update(bean, attrs);
T rs = cache.update(entity, attrs);
if (cacheListener != null) cacheListener.updateCache(info.getType(), rs);
return count >= 0 ? count : (rs == null ? 0 : 1);
}
@@ -1197,7 +1197,7 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
return count >= 0 ? count : (rs == null ? 0 : 1);
}
protected <T> int updateCache(final EntityInfo<T> info, int count, final Serializable id, final ColumnValue... values) {
protected <T> int updateCache(final EntityInfo<T> info, int count, final Serializable pk, final ColumnValue... values) {
final EntityCache<T> cache = info.getCache();
if (cache == null) return count;
final List<Attribute<T, Serializable>> attrs = new ArrayList<>();
@@ -1209,58 +1209,58 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
attrs.add(attr);
cols.add(col);
}
T rs = cache.updateColumn(id, attrs, cols);
T rs = cache.updateColumn(pk, attrs, cols);
if (cacheListener != null) cacheListener.updateCache(info.getType(), rs);
return count >= 0 ? count : (rs == null ? 0 : 1);
}
protected <T> int updateCache(final EntityInfo<T> info, int count, String column, final Serializable value, FilterNode node) {
protected <T> int updateCache(final EntityInfo<T> info, int count, String column, final Serializable colval, FilterNode node) {
final EntityCache<T> cache = info.getCache();
if (cache == null) return count;
T[] rs = cache.update(info.getAttribute(column), value, node);
T[] rs = cache.update(info.getAttribute(column), colval, node);
if (cacheListener != null) cacheListener.updateCache(info.getType(), rs);
return count >= 0 ? count : (rs == null ? 0 : 1);
}
protected <T> int updateCache(final EntityInfo<T> info, int count, final Serializable id, final String column, final Serializable value) {
protected <T> int updateCache(final EntityInfo<T> info, int count, final Serializable pk, final String column, final Serializable colval) {
final EntityCache<T> cache = info.getCache();
if (cache == null) return count;
T rs = cache.update(id, info.getAttribute(column), value);
T rs = cache.update(pk, info.getAttribute(column), colval);
if (cacheListener != null) cacheListener.updateCache(info.getType(), rs);
return count >= 0 ? count : (rs == null ? 0 : 1);
}
protected <T> int updateCache(final EntityInfo<T> info, int count, T... values) {
protected <T> int updateCache(final EntityInfo<T> info, int count, T... entitys) {
final EntityCache<T> cache = info.getCache();
if (cache == null) return -1;
int c2 = 0;
for (final T value : values) {
for (final T value : entitys) {
c2 += cache.update(value);
}
if (cacheListener != null) cacheListener.updateCache(info.getType(), values);
if (cacheListener != null) cacheListener.updateCache(info.getType(), entitys);
return count >= 0 ? count : c2;
}
@Override
public <T> int updateCache(Class<T> clazz, T... values) {
if (values.length == 0) return 0;
public <T> int updateCache(Class<T> clazz, T... entitys) {
if (entitys.length == 0) return 0;
final EntityInfo<T> info = loadEntityInfo(clazz);
final EntityCache<T> cache = info.getCache();
if (cache == null) return -1;
int c = 0;
for (T value : values) {
for (T value : entitys) {
c += cache.update(value);
}
return c;
}
public <T> int reloadCache(Class<T> clazz, Serializable... ids) {
public <T> int reloadCache(Class<T> clazz, Serializable... pks) {
final EntityInfo<T> info = loadEntityInfo(clazz);
final EntityCache<T> cache = info.getCache();
if (cache == null) return -1;
String column = info.getPrimary().field();
int c = 0;
for (Serializable id : ids) {
for (Serializable id : pks) {
Sheet<T> sheet = querySheetCompose(false, true, clazz, null, FLIPPER_ONE, FilterNode.create(column, id)).join();
T value = sheet.isEmpty() ? null : sheet.list().get(0);
if (value != null) c += cache.update(value);
@@ -1540,13 +1540,13 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
}
@Override
public <T> T find(final Class<T> clazz, final String column, final Serializable key) {
return find(clazz, null, FilterNode.create(column, key));
public <T> T find(final Class<T> clazz, final String column, final Serializable colval) {
return find(clazz, null, FilterNode.create(column, colval));
}
@Override
public <T> CompletableFuture<T> findAsync(final Class<T> clazz, final String column, final Serializable key) {
return findAsync(clazz, null, FilterNode.create(column, key));
public <T> CompletableFuture<T> findAsync(final Class<T> clazz, final String column, final Serializable colval) {
return findAsync(clazz, null, FilterNode.create(column, colval));
}
@Override
@@ -1782,13 +1782,13 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
//-----------------------list set----------------------------
@Override
public <T, V extends Serializable> HashSet<V> queryColumnSet(final String selectedColumn, Class<T> clazz, String column, Serializable key) {
return new LinkedHashSet<>(queryColumnList(selectedColumn, clazz, null, FilterNode.create(column, key)));
public <T, V extends Serializable> HashSet<V> queryColumnSet(final String selectedColumn, Class<T> clazz, String column, Serializable colval) {
return new LinkedHashSet<>(queryColumnList(selectedColumn, clazz, null, FilterNode.create(column, colval)));
}
@Override
public <T, V extends Serializable> CompletableFuture<HashSet<V>> queryColumnSetAsync(final String selectedColumn, Class<T> clazz, String column, Serializable key) {
return queryColumnListAsync(selectedColumn, clazz, null, FilterNode.create(column, key)).thenApply((list) -> new LinkedHashSet(list));
public <T, V extends Serializable> CompletableFuture<HashSet<V>> queryColumnSetAsync(final String selectedColumn, Class<T> clazz, String column, Serializable colval) {
return queryColumnListAsync(selectedColumn, clazz, null, FilterNode.create(column, colval)).thenApply((list) -> new LinkedHashSet(list));
}
@Override
@@ -1812,13 +1812,13 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
}
@Override
public <T, V extends Serializable> List<V> queryColumnList(final String selectedColumn, final Class<T> clazz, final String column, final Serializable key) {
return queryColumnList(selectedColumn, clazz, null, FilterNode.create(column, key));
public <T, V extends Serializable> List<V> queryColumnList(final String selectedColumn, final Class<T> clazz, final String column, final Serializable colval) {
return queryColumnList(selectedColumn, clazz, null, FilterNode.create(column, colval));
}
@Override
public <T, V extends Serializable> CompletableFuture<List<V>> queryColumnListAsync(final String selectedColumn, final Class<T> clazz, final String column, final Serializable key) {
return queryColumnListAsync(selectedColumn, clazz, null, FilterNode.create(column, key));
public <T, V extends Serializable> CompletableFuture<List<V>> queryColumnListAsync(final String selectedColumn, final Class<T> clazz, final String column, final Serializable colval) {
return queryColumnListAsync(selectedColumn, clazz, null, FilterNode.create(column, colval));
}
@Override
@@ -2028,10 +2028,10 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
public <K extends Serializable, T> CompletableFuture<Map<K, T>> queryMapAsync(final Class<T> clazz, final SelectColumn selects, final Stream<K> keyStream) {
if (keyStream == null) return CompletableFuture.completedFuture(new LinkedHashMap<>());
final EntityInfo<T> info = loadEntityInfo(clazz);
final ArrayList<K> ids = new ArrayList<>();
keyStream.forEach(k -> ids.add(k));
final ArrayList<K> pks = new ArrayList<>();
keyStream.forEach(k -> pks.add(k));
final Attribute<T, Serializable> primary = info.primary;
return queryListAsync(clazz, FilterNode.create(primary.field(), ids)).thenApply((List<T> rs) -> {
return queryListAsync(clazz, FilterNode.create(primary.field(), pks)).thenApply((List<T> rs) -> {
Map<K, T> map = new LinkedHashMap<>();
if (rs.isEmpty()) return new LinkedHashMap<>();
for (T item : rs) {
@@ -2108,18 +2108,18 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
* @param <T> Entity类的泛型
* @param clazz Entity类
* @param column 过滤字段名
* @param key 过滤字段值
* @param colval 过滤字段值
*
* @return Entity对象的集合
*/
@Override
public <T> List<T> queryList(final Class<T> clazz, final String column, final Serializable key) {
return queryList(clazz, (SelectColumn) null, null, FilterNode.create(column, key));
public <T> List<T> queryList(final Class<T> clazz, final String column, final Serializable colval) {
return queryList(clazz, (SelectColumn) null, null, FilterNode.create(column, colval));
}
@Override
public <T> CompletableFuture<List<T>> queryListAsync(final Class<T> clazz, final String column, final Serializable key) {
return queryListAsync(clazz, (SelectColumn) null, null, FilterNode.create(column, key));
public <T> CompletableFuture<List<T>> queryListAsync(final Class<T> clazz, final String column, final Serializable colval) {
return queryListAsync(clazz, (SelectColumn) null, null, FilterNode.create(column, colval));
}
@Override
@@ -2192,13 +2192,13 @@ public abstract class DataSqlSource<DBChannel> extends AbstractService implement
}
@Override
public <T> List<T> queryList(final Class<T> clazz, final Flipper flipper, final String column, final Serializable key) {
return queryList(clazz, null, flipper, FilterNode.create(column, key));
public <T> List<T> queryList(final Class<T> clazz, final Flipper flipper, final String column, final Serializable colval) {
return queryList(clazz, null, flipper, FilterNode.create(column, colval));
}
@Override
public <T> CompletableFuture<List<T>> queryListAsync(final Class<T> clazz, final Flipper flipper, final String column, final Serializable key) {
return queryListAsync(clazz, null, flipper, FilterNode.create(column, key));
public <T> CompletableFuture<List<T>> queryListAsync(final Class<T> clazz, final Flipper flipper, final String column, final Serializable colval) {
return queryListAsync(clazz, null, flipper, FilterNode.create(column, colval));
}
@Override

122
src/org/redkale/source/EntityCache.java
View File

@@ -159,15 +159,15 @@ public final class EntityCache<T> {
return fullloaded;
}
public T find(Serializable id) {
if (id == null) return null;
T rs = map.get(id);
public T find(Serializable pk) {
if (pk == null) return null;
T rs = map.get(pk);
return rs == null ? null : (needcopy ? newReproduce.apply(this.creator.create(), rs) : rs);
}
public T find(final SelectColumn selects, final Serializable id) {
if (id == null) return null;
T rs = map.get(id);
public T find(final SelectColumn selects, final Serializable pk) {
if (pk == null) return null;
T rs = map.get(pk);
if (rs == null) return null;
if (selects == null) return (needcopy ? newReproduce.apply(this.creator.create(), rs) : rs);
T t = this.creator.create();
@@ -192,9 +192,9 @@ public final class EntityCache<T> {
return t;
}
public Serializable findColumn(final String column, final Serializable defValue, final Serializable id) {
if (id == null) return defValue;
T rs = map.get(id);
public Serializable findColumn(final String column, final Serializable defValue, final Serializable pk) {
if (pk == null) return defValue;
T rs = map.get(pk);
if (rs == null) return defValue;
for (Attribute attr : this.info.attributes) {
if (column.equals(attr.field())) {
@@ -221,29 +221,29 @@ public final class EntityCache<T> {
return defValue;
}
public boolean exists(Serializable id) {
if (id == null) return false;
public boolean exists(Serializable pk) {
if (pk == null) return false;
final Class atype = this.primary.type();
if (id.getClass() != atype && id instanceof Number) {
if (pk.getClass() != atype && pk instanceof Number) {
if (atype == int.class || atype == Integer.class) {
id = ((Number) id).intValue();
pk = ((Number) pk).intValue();
} else if (atype == long.class || atype == Long.class) {
id = ((Number) id).longValue();
pk = ((Number) pk).longValue();
} else if (atype == short.class || atype == Short.class) {
id = ((Number) id).shortValue();
pk = ((Number) pk).shortValue();
} else if (atype == float.class || atype == Float.class) {
id = ((Number) id).floatValue();
pk = ((Number) pk).floatValue();
} else if (atype == byte.class || atype == Byte.class) {
id = ((Number) id).byteValue();
pk = ((Number) pk).byteValue();
} else if (atype == double.class || atype == Double.class) {
id = ((Number) id).doubleValue();
pk = ((Number) pk).doubleValue();
} else if (atype == AtomicInteger.class) {
id = new AtomicInteger(((Number) id).intValue());
pk = new AtomicInteger(((Number) pk).intValue());
} else if (atype == AtomicLong.class) {
id = new AtomicLong(((Number) id).longValue());
pk = new AtomicLong(((Number) pk).longValue());
}
}
return this.map.containsKey(id);
return this.map.containsKey(pk);
}
public boolean exists(FilterNode node) {
@@ -440,22 +440,22 @@ public final class EntityCache<T> {
return new Sheet<>(total, rs);
}
public int insert(T value) {
if (value == null) return 0;
final T rs = newReproduce.apply(this.creator.create(), value); //确保同一主键值的map与list中的对象必须共用。
public int insert(T entity) {
if (entity == null) return 0;
final T rs = newReproduce.apply(this.creator.create(), entity); //确保同一主键值的map与list中的对象必须共用。
T old = this.map.putIfAbsent(this.primary.get(rs), rs);
if (old == null) {
this.list.add(rs);
return 1;
} else {
logger.log(Level.WARNING, this.type + " cache repeat insert data: " + value);
logger.log(Level.WARNING, this.type + " cache repeat insert data: " + entity);
return 0;
}
}
public int delete(final Serializable id) {
if (id == null) return 0;
final T rs = this.map.remove(id);
public int delete(final Serializable pk) {
if (pk == null) return 0;
final T rs = this.map.remove(pk);
if (rs == null) return 0;
this.list.remove(rs);
return 1;
@@ -484,44 +484,44 @@ public final class EntityCache<T> {
return clear();
}
public int update(final T value) {
if (value == null) return 0;
T rs = this.map.get(this.primary.get(value));
public int update(final T entity) {
if (entity == null) return 0;
T rs = this.map.get(this.primary.get(entity));
if (rs == null) return 0;
synchronized (rs) {
this.chgReproduce.apply(rs, value);
this.chgReproduce.apply(rs, entity);
}
return 1;
}
public T update(final T value, Collection<Attribute<T, Serializable>> attrs) {
if (value == null) return value;
T rs = this.map.get(this.primary.get(value));
public T update(final T entity, Collection<Attribute<T, Serializable>> attrs) {
if (entity == null) return entity;
T rs = this.map.get(this.primary.get(entity));
if (rs == null) return rs;
synchronized (rs) {
for (Attribute attr : attrs) {
attr.set(rs, attr.get(value));
attr.set(rs, attr.get(entity));
}
}
return rs;
}
public T[] update(final T value, final Collection<Attribute<T, Serializable>> attrs, final FilterNode node) {
if (value == null || node == null) return (T[]) Array.newInstance(type, 0);
public T[] update(final T entity, final Collection<Attribute<T, Serializable>> attrs, final FilterNode node) {
if (entity == null || node == null) return (T[]) Array.newInstance(type, 0);
T[] rms = this.list.stream().filter(node.createPredicate(this)).toArray(len -> (T[]) Array.newInstance(type, len));
for (T rs : rms) {
synchronized (rs) {
for (Attribute attr : attrs) {
attr.set(rs, attr.get(value));
attr.set(rs, attr.get(entity));
}
}
}
return rms;
}
public <V> T update(final Serializable id, Attribute<T, V> attr, final V fieldValue) {
if (id == null) return null;
T rs = this.map.get(id);
public <V> T update(final Serializable pk, Attribute<T, V> attr, final V fieldValue) {
if (pk == null) return null;
T rs = this.map.get(pk);
if (rs != null) attr.set(rs, fieldValue);
return rs;
}
@@ -535,9 +535,9 @@ public final class EntityCache<T> {
return rms;
}
public <V> T updateColumn(final Serializable id, List<Attribute<T, Serializable>> attrs, final List<ColumnValue> values) {
if (id == null || attrs == null || attrs.isEmpty()) return null;
T rs = this.map.get(id);
public <V> T updateColumn(final Serializable pk, List<Attribute<T, Serializable>> attrs, final List<ColumnValue> values) {
if (pk == null || attrs == null || attrs.isEmpty()) return null;
T rs = this.map.get(pk);
if (rs == null) return rs;
synchronized (rs) {
for (int i = 0; i < attrs.size(); i++) {
@@ -566,40 +566,40 @@ public final class EntityCache<T> {
return rms;
}
public <V> T updateColumnOr(final Serializable id, Attribute<T, V> attr, final long orvalue) {
if (id == null) return null;
T rs = this.map.get(id);
public <V> T updateColumnOr(final Serializable pk, Attribute<T, V> attr, final long orvalue) {
if (pk == null) return null;
T rs = this.map.get(pk);
if (rs == null) return rs;
synchronized (rs) {
return updateColumn(attr, rs, ColumnExpress.ORR, orvalue);
}
}
public <V> T updateColumnAnd(final Serializable id, Attribute<T, V> attr, final long andvalue) {
if (id == null) return null;
T rs = this.map.get(id);
public <V> T updateColumnAnd(final Serializable pk, Attribute<T, V> attr, final long andvalue) {
if (pk == null) return null;
T rs = this.map.get(pk);
if (rs == null) return rs;
synchronized (rs) {
return updateColumn(attr, rs, ColumnExpress.AND, andvalue);
}
}
public <V> T updateColumnIncrement(final Serializable id, Attribute<T, V> attr, final long incvalue) {
if (id == null) return null;
T rs = this.map.get(id);
public <V> T updateColumnIncrement(final Serializable pk, Attribute<T, V> attr, final long incvalue) {
if (pk == null) return null;
T rs = this.map.get(pk);
if (rs == null) return rs;
synchronized (rs) {
return updateColumn(attr, rs, ColumnExpress.INC, incvalue);
}
}
private <V> T updateColumn(Attribute<T, V> attr, final T rs, final ColumnExpress express, Serializable val) {
private <V> T updateColumn(Attribute<T, V> attr, final T entity, final ColumnExpress express, Serializable val) {
final Class ft = attr.type();
Number numb = null;
Serializable newval = null;
switch (express) {
case INC:
numb = (Number) attr.get(rs);
numb = (Number) attr.get(entity);
if (numb == null) {
numb = (Number) val;
} else {
@@ -607,7 +607,7 @@ public final class EntityCache<T> {
}
break;
case MUL:
numb = (Number) attr.get(rs);
numb = (Number) attr.get(entity);
if (numb == null) {
numb = 0;
} else {
@@ -615,7 +615,7 @@ public final class EntityCache<T> {
}
break;
case AND:
numb = (Number) attr.get(rs);
numb = (Number) attr.get(entity);
if (numb == null) {
numb = 0;
} else {
@@ -623,7 +623,7 @@ public final class EntityCache<T> {
}
break;
case ORR:
numb = (Number) attr.get(rs);
numb = (Number) attr.get(entity);
if (numb == null) {
numb = 0;
} else {
@@ -659,8 +659,8 @@ public final class EntityCache<T> {
newval = new AtomicLong(((Number) newval).longValue());
}
}
attr.set(rs, (V) newval);
return rs;
attr.set(entity, (V) newval);
return entity;
}
public Attribute<T, Serializable> getAttribute(String fieldname) {