flyfish-group/flyfish-framework
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 使用设计模式优化代码

Browse Source
This commit is contained in:
wangyu 2024-06-30 14:50:09 +08:00
parent 7758cede09
commit 27ec8bf22d
15 changed files with 759 additions and 248 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

145
flyfish-data/flyfish-data-r2dbc/src/main/java/com/flyfish/framework/r2dbc/config/callback/AssociationFetcher.java Normal file
View File

@ -0,0 +1,145 @@
package com.flyfish.framework.r2dbc.config.callback;
import com.flyfish.framework.domain.base.Domain;
import com.flyfish.framework.query.Queries;
import com.flyfish.framework.r2dbc.metadata.reference.R2dbcAssociation;
import com.flyfish.framework.r2dbc.metadata.reference.R2dbcCollection;
import com.flyfish.framework.r2dbc.metadata.reference.R2dbcRelation;
import com.flyfish.framework.r2dbc.metadata.reference.field.FieldAccessor;
import com.flyfish.framework.r2dbc.metadata.visitor.MetadataHandler;
import com.flyfish.framework.repository.DefaultReactiveRepository;
import lombok.RequiredArgsConstructor;
import org.apache.commons.lang3.ObjectUtils;
import org.springframework.data.mapping.PersistentProperty;
import org.springframework.data.mapping.PersistentPropertyAccessor;
import org.springframework.data.r2dbc.core.R2dbcEntityOperations;
import org.springframework.data.r2dbc.core.StatementMapper;
import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
import org.springframework.data.relational.core.query.Criteria;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.r2dbc.core.PreparedOperation;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.function.Supplier;
@RequiredArgsConstructor
class AssociationFetcher implements MetadataHandler {
private final RepositoryProvider repositories;
private final Supplier<R2dbcEntityOperations> operations;
/**
* 处理association
*
* @param association 一对一关联
* @param accessor 对象访问器
* @return 结果
*/
@Override
public Mono<Domain> handleAssociation(R2dbcAssociation association, PersistentPropertyAccessor<Domain> accessor) {
// 准备查询， 获取对方的仓库
DefaultReactiveRepository<Domain> repository = repositories.get(association.getEntity());
// 准备设置器
FieldAccessor field = association.getAccessor();
if (association.isInner()) {
// 处理内部查询，以对象内的值作为查询条件
Object value = accessor.getProperty(association.getField());
if (ObjectUtils.isEmpty(value)) return null;
// 查询
return repository.findById(String.valueOf(value))
.map(result -> field.set(accessor.getBean(), result))
.defaultIfEmpty(accessor.getBean());
} else {
// 处理外部关联查询
String id = accessor.getBean().getId();
// 直接通过外部字段查询
return repository.findOne(Queries.where(getColumnName(association.getField())).eq(id).wrap())
.map(result -> field.set(accessor.getBean(), result))
.defaultIfEmpty(accessor.getBean());
}
}
/**
* 处理collection
*
* @param collection 一对多关联
* @param accessor 对象访问器
* @return 结果
*/
@Override
public Mono<Domain> handleCollection(R2dbcCollection collection, PersistentPropertyAccessor<Domain> accessor) {
// 获取条件
String id = accessor.getBean().getId();
// 准备查询
DefaultReactiveRepository<Domain> repository = repositories.get(collection.getEntity());
// 准备设置
FieldAccessor field = collection.getAccessor();
// 直接通过外部字段查询
return repository.findAll(Queries.where(getColumnName(collection.getField())).eq(id).wrap())
.collectList()
.map(result -> field.set(accessor.getBean(), result))
.defaultIfEmpty(accessor.getBean());
}
/**
* 处理relation
*
* @param relation 多对多关联
* @param accessor 访问器
* @return 结果
*/
@Override
public Mono<Domain> handleRelation(R2dbcRelation relation, PersistentPropertyAccessor<Domain> accessor) {
// 准备目标对象的repo
DefaultReactiveRepository<Domain> repository = repositories.get(relation.getEntity());
// 准备查询参数
String id = accessor.getBean().getId();
// 查询关联id集合
Flux<String> ids = fetchRelationIds(relation, id);
// 准备设置
FieldAccessor field = relation.getAccessor();
// 查询最终结果
return repository.findAllById(ids)
.collectList()
.map(result -> field.set(accessor.getBean(), result))
.defaultIfEmpty(accessor.getBean());
}
/**
* 查询当前实体关联的ids
*
* @param relation 关联信息
* @return 结果
*/
@SuppressWarnings("deprecation")
private Flux<String> fetchRelationIds(R2dbcRelation relation, String id) {
// 准备查询器
R2dbcEntityOperations entityOperations = operations.get();
StatementMapper statementMapper = entityOperations.getDataAccessStrategy().getStatementMapper();
DatabaseClient databaseClient = entityOperations.getDatabaseClient();
// 构建查询
StatementMapper.SelectSpec select = statementMapper.createSelect(relation.getTableName())
.withProjection(relation.getForeignField()).withCriteria(Criteria.where(relation.getField()).is(id));
PreparedOperation<?> operation = statementMapper.getMappedObject(select);
// 执行查询
return databaseClient.sql(operation)
.map(row -> row.get(0, String.class))
.all();
}
/**
* 通过持久化属性获取真实的列名
*
* @param property 属性
* @return 结果
*/
private String getColumnName(PersistentProperty<?> property) {
if (property instanceof RelationalPersistentProperty) {
return ((RelationalPersistentProperty) property).getColumnName().getReference();
}
return property.getName();
}
}

189
flyfish-data/flyfish-data-r2dbc/src/main/java/com/flyfish/framework/r2dbc/config/callback/AssociationWriter.java Normal file
View File

@ -0,0 +1,189 @@
package com.flyfish.framework.r2dbc.config.callback;
import com.flyfish.framework.domain.base.Domain;
import com.flyfish.framework.r2dbc.metadata.reference.R2dbcAssociation;
import com.flyfish.framework.r2dbc.metadata.reference.R2dbcCollection;
import com.flyfish.framework.r2dbc.metadata.reference.R2dbcRelation;
import com.flyfish.framework.r2dbc.metadata.visitor.MetadataHandler;
import com.flyfish.framework.repository.DefaultReactiveRepository;
import lombok.RequiredArgsConstructor;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.mapping.PersistentEntity;
import org.springframework.data.mapping.PersistentProperty;
import org.springframework.data.mapping.PersistentPropertyAccessor;
import org.springframework.data.r2dbc.core.R2dbcEntityOperations;
import org.springframework.data.r2dbc.core.StatementMapper;
import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
import org.springframework.data.relational.core.query.Criteria;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.r2dbc.core.Parameter;
import org.springframework.r2dbc.core.PreparedOperation;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@RequiredArgsConstructor
class AssociationWriter implements MetadataHandler {
private final RepositoryProvider repositories;
private final Supplier<R2dbcEntityOperations> operations;
/**
* 通过持久化属性获取真实的列名
*
* @param property 属性
* @return 结果
*/
private String getColumnName(PersistentProperty<?> property) {
if (property instanceof RelationalPersistentProperty) {
return ((RelationalPersistentProperty) property).getColumnName().getReference();
}
return property.getName();
}
/**
* 处理association
*
* @param association 一对一关联
* @param accessor 对象访问器
* @return 结果
*/
@Override
public Mono<Domain> handleAssociation(R2dbcAssociation association, PersistentPropertyAccessor<Domain> accessor) {
Domain entity = accessor.getBean();
// 获取目标值
Domain value = association.getAccessor().get(entity);
if (null == value) return null;
// 准备仓库
DefaultReactiveRepository<Domain> repository = repositories.get(association.getEntity());
// 执行逻辑
if (association.isInner()) {
// 保存内部关联对象，并设置关联值。如果有id，更新，没有id，插入
if (StringUtils.isNotBlank(value.getId())) {
// 设置内部id关联
accessor.setProperty(association.getField(), value.getId());
// 尝试保存
return repository.save(value).thenReturn(entity);
} else {
// 先保存，再设置
return repository.insert(value).map(saved -> {
// 设置内部id关联
accessor.setProperty(association.getField(), saved.getId());
return entity;
});
}
} else {
// 保存外部关联对象，并更新外部关联当前实体的id
PersistentPropertyAccessor<Domain> otherAccessor = association.getEntity().getPropertyAccessor(value);
// 先设置目标id
otherAccessor.setProperty(association.getField(), entity.getId());
// 再保存
return repository.save(value).thenReturn(entity);
}
}
/**
* 处理collection
*
* @param collection 一对多关联
* @param accessor 对象访问器
* @return 结果
*/
@Override
public Mono<Domain> handleCollection(R2dbcCollection collection, PersistentPropertyAccessor<Domain> accessor) {
Domain entity = accessor.getBean();
// 获取目标值
List<Domain> value = collection.getAccessor().get(entity);
if (CollectionUtils.isEmpty(value)) return null;
// 获取目标实体信息
PersistentEntity<?, ?> targetEntity = collection.getEntity();
// 保存外部关联对象，并更新外部关联当前实体的id
value.forEach(item -> {
PersistentPropertyAccessor<Domain> otherAccessor = collection.getEntity().getPropertyAccessor(item);
otherAccessor.setProperty(collection.getField(), entity.getId());
});
// 准备仓库
DefaultReactiveRepository<Domain> repository = repositories.get(targetEntity);
// 批量保存
return repository.saveAll(value).collectList()
.map(list -> collection.getAccessor().set(entity, list));
}
/**
* 处理relation
*
* @param relation 多对多关联
* @param accessor 访问器
* @return 结果
*/
@Override
public Mono<Domain> handleRelation(R2dbcRelation relation, PersistentPropertyAccessor<Domain> accessor) {
// 保存多对多关系，需要先分别插入两边的数据，然后在关联表新增或更新
Domain entity = accessor.getBean();
// 获取目标值
List<Domain> value = relation.getAccessor().get(entity);
if (CollectionUtils.isEmpty(value)) return null;
// 获取目标实体信息
PersistentEntity<?, ?> targetEntity = relation.getEntity();
// 准备仓库
DefaultReactiveRepository<Domain> repository = repositories.get(targetEntity);
// 批量保存
return repository.saveAll(value).collectList()
.flatMap(list -> {
// 回填数据
relation.getAccessor().set(entity, list);
// 关系表保存
return saveRelationIds(relation, entity);
});
}
/**
* 保存关系表
*
* @param relation 关系信息
* @param entity 实体信息
*/
@SuppressWarnings("deprecation")
private Mono<Domain> saveRelationIds(R2dbcRelation relation, Domain entity) {
// 准备查询器
R2dbcEntityOperations entityOperations = operations.get();
StatementMapper statementMapper = entityOperations.getDataAccessStrategy().getStatementMapper();
DatabaseClient databaseClient = entityOperations.getDatabaseClient();
// 构建保存语句，先删后增
StatementMapper.DeleteSpec deleteSpec = statementMapper.createDelete(relation.getTableName())
.withCriteria(Criteria.where(relation.getField()).is(entity.getId()));
PreparedOperation<?> deletion = statementMapper.getMappedObject(deleteSpec);
// 构建插入语句，多条
List<Domain> items = relation.getAccessor().get(entity);
Stream<PreparedOperation<?>> insertions = items.stream().map(item -> {
StatementMapper.InsertSpec insertion = statementMapper.createInsert(relation.getTableName())
.withColumn(relation.getField(), Parameter.from(entity.getId()))
.withColumn(relation.getForeignField(), Parameter.from(item.getId()));
return statementMapper.getMappedObject(insertion);
});
// 执行查询
return execute(databaseClient, deletion, entity)
.map(e -> insertions.map(operation -> execute(databaseClient, operation, e)).collect(Collectors.toList()))
.flatMap(list -> Mono.zip(list, results -> entity));
}
/**
* 执行更新语句的逻辑，保证多行每一行都有执行到
*
* @param databaseClient 数据库客户端
* @param operation 预编译的sql
* @param entity 实体
* @return 结果
*/
private Mono<Domain> execute(DatabaseClient databaseClient, PreparedOperation<?> operation, Domain entity) {
return databaseClient.sql(operation)
.map((row -> entity))
.all()
.last();
}
}

220
flyfish-data/flyfish-data-r2dbc/src/main/java/com/flyfish/framework/r2dbc/config/callback/ReferenceR2dbcCallback.java
View File

@ -2,51 +2,45 @@ package com.flyfish.framework.r2dbc.config.callback;
import com.flyfish.framework.domain.base.Domain;
import com.flyfish.framework.domain.base.Qo;
import com.flyfish.framework.query.Queries;
import com.flyfish.framework.r2dbc.metadata.R2dbcMetadataManager;
import com.flyfish.framework.r2dbc.metadata.R2dbcTableMetadata;
import com.flyfish.framework.r2dbc.metadata.reference.FieldSetter;
import com.flyfish.framework.r2dbc.metadata.reference.R2dbcAssociation;
import com.flyfish.framework.r2dbc.metadata.reference.R2dbcCollection;
import com.flyfish.framework.r2dbc.metadata.reference.R2dbcRelation;
import com.flyfish.framework.r2dbc.metadata.visitor.MetadataHandler;
import com.flyfish.framework.r2dbc.metadata.visitor.R2dbcTableMetadataVisitor;
import com.flyfish.framework.repository.DefaultReactiveRepository;
import lombok.RequiredArgsConstructor;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.ObjectUtils;
import org.reactivestreams.Publisher;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mapping.PersistentEntity;
import org.springframework.data.mapping.PersistentProperty;
import org.springframework.data.mapping.PersistentPropertyAccessor;
import org.springframework.data.r2dbc.core.R2dbcEntityOperations;
import org.springframework.data.r2dbc.core.StatementMapper;
import org.springframework.data.r2dbc.mapping.OutboundRow;
import org.springframework.data.r2dbc.mapping.event.AfterConvertCallback;
import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
import org.springframework.data.relational.core.query.Criteria;
import org.springframework.data.r2dbc.mapping.event.AfterSaveCallback;
import org.springframework.data.relational.core.sql.SqlIdentifier;
import org.springframework.data.util.CastUtils;
import org.springframework.data.util.Lazy;
import org.springframework.lang.NonNull;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.r2dbc.core.PreparedOperation;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
@RequiredArgsConstructor
public class ReferenceR2dbcCallback implements AfterConvertCallback<Domain> {
public class ReferenceR2dbcCallback implements AfterConvertCallback<Domain>, AfterSaveCallback<Domain>, RepositoryProvider, Supplier<R2dbcEntityOperations> {
private final R2dbcMetadataManager r2dbcMetadataManager;
private final MetadataHandler fetchHandler;
private final MetadataHandler writeHandler;
private Lazy<R2dbcEntityOperations> operations;
private Lazy<Map<Class<?>, DefaultReactiveRepository<?>>> repositories;
public ReferenceR2dbcCallback(R2dbcMetadataManager r2dbcMetadataManager) {
this.r2dbcMetadataManager = r2dbcMetadataManager;
this.fetchHandler = new AssociationFetcher(this, this);
this.writeHandler = new AssociationWriter(this, this);
}
@Override
@NonNull
public Publisher<Domain> onAfterConvert(@NonNull Domain entity, @NonNull SqlIdentifier table) {
@ -56,42 +50,40 @@ public class ReferenceR2dbcCallback implements AfterConvertCallback<Domain> {
// 判断是否需要获取关联信息
boolean fetchRefs = query.map(Qo::isFetchRef).orElse(false);
if (fetchRefs) {
return doFetch(entity);
R2dbcTableMetadata metadata = r2dbcMetadataManager.getMetadata(entity.getClass());
R2dbcTableMetadataVisitor visitor = new R2dbcTableMetadataVisitor(entity, fetchHandler);
metadata.visit(visitor);
return visitor.getResult();
}
return Mono.just(entity);
});
}
/**
* 完成填充
* 保存完成后的回调，用于保存额外的关联对象
*
* @param entity 实体
* @return 结果
* @param entity 已经保存的实体，带有id信息
* @param outboundRow {@link OutboundRow} 输出行，包含实体的完整内容 {@code entity}.
* @param table 表名
* @return 当前已经实例化的对象
*/
private Mono<Domain> doFetch(Domain entity) {
@Override
@NonNull
public Publisher<Domain> onAfterSave(@NonNull Domain entity, @NonNull OutboundRow outboundRow, @NonNull SqlIdentifier table) {
Class<? extends Domain> entityClass = entity.getClass();
// 遍历fields，找到要注入的数据
R2dbcTableMetadata metadata = r2dbcMetadataManager.getMetadata(entityClass);
// 获取本实体的持久化对象
PersistentEntity<?, ?> persistentEntity = metadata.getEntity();
// 获取属性设置器
PersistentPropertyAccessor<Domain> accessor = persistentEntity.getPropertyAccessor(entity);
// 尝试填充各种关联方式
List<Mono<Domain>> signals = new ArrayList<>();
signals.addAll(fetchAssociation(metadata.getAssociations(), accessor));
signals.addAll(fetchCollections(metadata.getCollections(), accessor));
signals.addAll(fetchRelations(metadata.getRelations(), accessor));
// 尝试填充
if (CollectionUtils.isNotEmpty(signals)) {
return Mono.zip(signals, objs -> entity);
if (metadata.hasAssociation()) {
R2dbcTableMetadataVisitor visitor = new R2dbcTableMetadataVisitor(entity, writeHandler);
metadata.visit(visitor);
return visitor.getResult();
}
return Mono.just(entity);
}
private DefaultReactiveRepository<Domain> getRepository(Class<?> entityClass) {
return CastUtils.cast(repositories.get().get(entityClass));
@Override
public <T extends Domain, R extends DefaultReactiveRepository<T>> R get(PersistentEntity<?, ?> entity) {
return CastUtils.cast(repositories.get().get(entity.getType()));
}
@Autowired
@ -106,148 +98,10 @@ public class ReferenceR2dbcCallback implements AfterConvertCallback<Domain> {
}
/**
* 拉取一对一关联
*
* @param associations 一对一关联集合
* @param accessor 本对象的属性访问器
* @return 结果
* 获取实体操作
*/
private List<Mono<Domain>> fetchAssociation(List<R2dbcAssociation> associations, PersistentPropertyAccessor<Domain> accessor) {
List<Mono<Domain>> signals = new ArrayList<>();
if (CollectionUtils.isNotEmpty(associations)) {
// 分区，将判断提在外面，提升性能
Map<Boolean, List<R2dbcAssociation>> partitions = associations.stream()
.collect(Collectors.partitioningBy(R2dbcAssociation::isInner));
// 处理内部查询
if (CollectionUtils.isNotEmpty(partitions.get(true))) {
partitions.get(true).forEach(association -> {
// 获取对方的仓库
DefaultReactiveRepository<Domain> repository = getRepository(association.getEntity().getType());
// 内部查询，以对象内的值作为查询条件
Object value = accessor.getProperty(association.getField());
if (ObjectUtils.isEmpty(value)) return;
// 准备设置
FieldSetter setter = association.getSetter();
// 查询
Mono<Domain> signal = repository.findById(String.valueOf(value))
.map(result -> setter.setValue(accessor.getBean(), result))
.defaultIfEmpty(accessor.getBean());
signals.add(signal);
});
}
// 处理外部查询
if (CollectionUtils.isNotEmpty(partitions.get(false))) {
// 外部查询，以当前实体的id为条件，去外部表的字段上查询
partitions.get(false).forEach(association -> {
// 获取条件
String id = accessor.getBean().getId();
// 准备查询
DefaultReactiveRepository<Domain> repository = getRepository(association.getEntity().getType());
// 准备设置
FieldSetter setter = association.getSetter();
// 直接通过外部字段查询
Mono<Domain> signal = repository.findOne(Queries.where(getColumnName(association.getField())).eq(id).wrap())
.map(result -> setter.setValue(accessor.getBean(), result))
.defaultIfEmpty(accessor.getBean());
signals.add(signal);
});
}
}
return signals;
}
/**
* 拉取一对多关联实体
*
* @param collections 一对多关联
* @param accessor 本对象的属性访问器
* @return 结果
*/
private List<Mono<Domain>> fetchCollections(List<R2dbcCollection> collections,
PersistentPropertyAccessor<Domain> accessor) {
List<Mono<Domain>> signals = new ArrayList<>();
if (CollectionUtils.isNotEmpty(collections)) {
// 外部查询，以当前实体的id为条件，去外部表的字段上查询
collections.forEach(collection -> {
// 获取条件
String id = accessor.getBean().getId();
// 准备查询
DefaultReactiveRepository<Domain> repository = getRepository(collection.getEntity().getType());
// 准备设置
FieldSetter setter = collection.getSetter();
// 直接通过外部字段查询
Mono<Domain> signal = repository.findAll(Queries.where(getColumnName(collection.getField())).eq(id).wrap())
.collectList()
.map(result -> setter.setValue(accessor.getBean(), result))
.defaultIfEmpty(accessor.getBean());
signals.add(signal);
});
}
return signals;
}
/**
* 拉取多对多关联数据
*
* @param relations 多对多关联信息
* @param accessor 本对象的属性访问器
* @return 结果
*/
private List<Mono<Domain>> fetchRelations(List<R2dbcRelation> relations, PersistentPropertyAccessor<Domain> accessor) {
List<Mono<Domain>> signals = new ArrayList<>();
if (CollectionUtils.isNotEmpty(relations)) {
// 生成DatabaseClient，执行sql语句查询
relations.forEach(relation -> {
// 准备目标对象的repo
DefaultReactiveRepository<Domain> repository = getRepository(relation.getEntity().getType());
// 准备查询参数
String id = accessor.getBean().getId();
// 查询关联id集合
Flux<String> ids = fetchRelationIds(relation, id);
// 准备设置
FieldSetter setter = relation.getSetter();
// 查询最终结果
Mono<Domain> signal = repository.findAllById(ids)
.collectList()
.map(result -> setter.setValue(accessor.getBean(), result))
.defaultIfEmpty(accessor.getBean());
signals.add(signal);
});
}
return signals;
}
/**
* 查询当前实体关联的ids
*
* @param relation 关联信息
* @return 结果
*/
private Flux<String> fetchRelationIds(R2dbcRelation relation, String id) {
// 准备查询器
R2dbcEntityOperations entityOperations = operations.get();
StatementMapper statementMapper = entityOperations.getDataAccessStrategy().getStatementMapper();
DatabaseClient databaseClient = entityOperations.getDatabaseClient();
// 构建查询
StatementMapper.SelectSpec select = statementMapper.createSelect(relation.getTableName())
.withProjection(relation.getForeignField()).withCriteria(Criteria.where(relation.getField()).is(id));
PreparedOperation<?> operation = statementMapper.getMappedObject(select);
// 执行查询
return databaseClient.sql(operation)
.map(row -> row.get(0, String.class))
.all();
}
/**
* 通过持久化属性获取真实的列名
*
* @param property 属性
* @return 结果
*/
private String getColumnName(PersistentProperty<?> property) {
if (property instanceof RelationalPersistentProperty) {
return ((RelationalPersistentProperty) property).getColumnName().getReference();
}
return property.getName();
@Override
public R2dbcEntityOperations get() {
return operations.get();
}
}

11
flyfish-data/flyfish-data-r2dbc/src/main/java/com/flyfish/framework/r2dbc/config/callback/RepositoryProvider.java Normal file
View File

@ -0,0 +1,11 @@
package com.flyfish.framework.r2dbc.config.callback;
import com.flyfish.framework.domain.base.Domain;
import com.flyfish.framework.repository.DefaultReactiveRepository;
import org.springframework.data.mapping.PersistentEntity;
@FunctionalInterface
interface RepositoryProvider {
<T extends Domain, R extends DefaultReactiveRepository<T>> R get(PersistentEntity<?, ?> entity);
}

22
flyfish-data/flyfish-data-r2dbc/src/main/java/com/flyfish/framework/r2dbc/metadata/R2dbcTableMetadata.java
View File

@ -1,14 +1,15 @@
package com.flyfish.framework.r2dbc.metadata;
import com.flyfish.framework.annotations.Property;
import com.flyfish.framework.relational.mapping.Association;
import com.flyfish.framework.r2dbc.metadata.reference.FieldSetter;
import com.flyfish.framework.r2dbc.metadata.reference.R2dbcAssociation;
import com.flyfish.framework.r2dbc.metadata.reference.R2dbcCollection;
import com.flyfish.framework.r2dbc.metadata.reference.R2dbcRelation;
import com.flyfish.framework.r2dbc.metadata.reference.field.FieldAccessor;
import com.flyfish.framework.relational.mapping.Association;
import lombok.Data;
import org.springframework.data.mapping.PersistentEntity;
import org.springframework.data.mapping.PersistentProperty;
import org.springframework.data.relational.core.sql.Visitable;
import java.util.ArrayList;
import java.util.List;
@ -20,7 +21,7 @@ import java.util.List;
*/
@Data
@Property("表的元数据，用于关联查询")
public class R2dbcTableMetadata {
public class R2dbcTableMetadata implements Visitable {
@Property("表实体信息")
private PersistentEntity<?, ?> entity;
@ -38,16 +39,19 @@ public class R2dbcTableMetadata {
this.entity = entity;
}
public void addAssociation(PersistentEntity<?, ?> entity, PersistentProperty<?> property, FieldSetter setter, boolean inner) {
this.associations.add(R2dbcAssociation.of(entity, property, setter, inner));
public void addAssociation(PersistentEntity<?, ?> entity, PersistentProperty<?> property, FieldAccessor accessor, boolean inner) {
this.associations.add(R2dbcAssociation.of(entity, property, accessor, inner));
}
public void addCollection(PersistentEntity<?, ?> entity, PersistentProperty<?> property, FieldSetter setter) {
this.collections.add(R2dbcCollection.of(entity, property, setter));
public void addCollection(PersistentEntity<?, ?> entity, PersistentProperty<?> property, FieldAccessor accessor) {
this.collections.add(R2dbcCollection.of(entity, property, accessor));
}
public void addRelation(PersistentEntity<?, ?> entity, Association association, FieldSetter setter) {
this.relations.add(R2dbcRelation.of(entity, association, setter));
public void addRelation(PersistentEntity<?, ?> entity, Association association, FieldAccessor accessor) {
this.relations.add(R2dbcRelation.of(entity, association, accessor));
}
public boolean hasAssociation() {
return !associations.isEmpty() || !collections.isEmpty() || !relations.isEmpty();
}
}

43
flyfish-data/flyfish-data-r2dbc/src/main/java/com/flyfish/framework/r2dbc/metadata/impl/SimpleR2dbcMetadataManager.java
View File

@ -1,15 +1,14 @@
package com.flyfish.framework.r2dbc.metadata.impl;
import com.flyfish.framework.relational.mapping.Association;
import com.flyfish.framework.domain.base.Domain;
import com.flyfish.framework.r2dbc.metadata.R2dbcMetadataManager;
import com.flyfish.framework.r2dbc.metadata.R2dbcTableMetadata;
import com.flyfish.framework.r2dbc.metadata.reference.FieldSetter;
import com.flyfish.framework.r2dbc.metadata.reference.field.FieldAccessor;
import com.flyfish.framework.relational.mapping.Association;
import lombok.RequiredArgsConstructor;
import org.apache.commons.lang3.ClassUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.core.ResolvableType;
import org.springframework.core.annotation.AnnotatedElementUtils;
import org.springframework.data.annotation.Reference;
@ -17,9 +16,7 @@ import org.springframework.data.mapping.PersistentEntity;
import org.springframework.data.mapping.PersistentProperty;
import org.springframework.data.r2dbc.mapping.R2dbcMappingContext;
import java.beans.PropertyDescriptor;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.Collection;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
@ -74,21 +71,6 @@ public class SimpleR2dbcMetadataManager implements R2dbcMetadataManager {
this.metadata = new R2dbcTableMetadata(entity);
}
private void invokeSetter(Method method, Object obj, Object value) {
try {
method.invoke(obj, value);
} catch (Exception ignored) {
}
}
private FieldSetter getSetter(Field field) {
PropertyDescriptor descriptor = BeanUtils.getPropertyDescriptor(entity.getType(), field.getName());
return Optional.ofNullable(descriptor).map(PropertyDescriptor::getWriteMethod)
.map(method -> (FieldSetter) (obj, value) -> invokeSetter(method, obj, value))
.orElse(null);
}
/**
* 处理所有关联属性
*
@ -98,8 +80,9 @@ public class SimpleR2dbcMetadataManager implements R2dbcMetadataManager {
public void accept(Field field) {
Association association = AnnotatedElementUtils.findMergedAnnotation(field, Association.class);
if (null == association) return;
FieldSetter setter = getSetter(field);
if (null == setter) return;
// 获取字段访问器，只支持标准的bean
FieldAccessor accessor = FieldAccessor.of(entity.getType(), field);
if (!accessor.isAvailable()) return;
// 获取具体类型
Class<?> type = field.getType();
if (ClassUtils.isAssignable(type, Domain.class)) {
@ -107,7 +90,7 @@ public class SimpleR2dbcMetadataManager implements R2dbcMetadataManager {
PersistentEntity<?, ?> targetEntity = r2dbcMappingContext.getRequiredPersistentEntity(type);
// 多对多关联
if (StringUtils.isNotBlank(association.relationTable())) {
metadata.addRelation(targetEntity, association, setter);
metadata.addRelation(targetEntity, association, accessor);
return;
}
// 一对一关联
@ -115,19 +98,19 @@ public class SimpleR2dbcMetadataManager implements R2dbcMetadataManager {
// 获取内部属性
PersistentProperty<?> property = entity.getRequiredPersistentProperty(association.field());
// 添加内部关联
metadata.addAssociation(targetEntity, property, setter, true);
metadata.addAssociation(targetEntity, property, accessor, true);
} else if (StringUtils.isNotBlank(association.foreignField())) {
// 获取外部属性
PersistentProperty<?> property = targetEntity.getRequiredPersistentProperty(association.foreignField());
// 添加外部关联
metadata.addAssociation(targetEntity, property, setter, false);
metadata.addAssociation(targetEntity, property, accessor, false);
} else {
// 都为空，从当前对象查找目标类型的reference，并自动添加关联
Optional<PersistentProperty<?>> inner = findInnerKey(type);
if (inner.isPresent()) {
metadata.addAssociation(targetEntity, inner.get(), setter, true);
metadata.addAssociation(targetEntity, inner.get(), accessor, true);
} else {
findForeignKey(targetEntity).ifPresent(property -> metadata.addAssociation(targetEntity, property, setter, false));
findForeignKey(targetEntity).ifPresent(property -> metadata.addAssociation(targetEntity, property, accessor, false));
}
}
} else if (ClassUtils.isAssignable(type, Collection.class)) {
@ -138,7 +121,7 @@ public class SimpleR2dbcMetadataManager implements R2dbcMetadataManager {
PersistentEntity<?, ?> targetEntity = r2dbcMappingContext.getRequiredPersistentEntity(entityType);
// 多对多关联
if (StringUtils.isNotBlank(association.relationTable())) {
metadata.addRelation(targetEntity, association, setter);
metadata.addRelation(targetEntity, association, accessor);
return;
}
// 一对多关联，仅支持外部字段
@ -146,10 +129,10 @@ public class SimpleR2dbcMetadataManager implements R2dbcMetadataManager {
// 获取外部属性
PersistentProperty<?> property = targetEntity.getRequiredPersistentProperty(association.foreignField());
// 添加外部关联
metadata.addCollection(targetEntity, property, setter);
metadata.addCollection(targetEntity, property, accessor);
} else {
// 都为空，从当前对象查找目标类型的reference，并自动添加关联
findForeignKey(targetEntity).ifPresent(property -> metadata.addCollection(targetEntity, property, setter));
findForeignKey(targetEntity).ifPresent(property -> metadata.addCollection(targetEntity, property, accessor));
}
}
}

14
flyfish-data/flyfish-data-r2dbc/src/main/java/com/flyfish/framework/r2dbc/metadata/reference/FieldSetter.java
View File

@ -1,14 +0,0 @@
package com.flyfish.framework.r2dbc.metadata.reference;
import com.flyfish.framework.domain.base.Domain;
import java.util.function.BiConsumer;
@FunctionalInterface
public interface FieldSetter extends BiConsumer<Object, Object> {
default <T extends Domain> T setValue(T obj, Object value) {
accept(obj, value);
return obj;
}
}

9
flyfish-data/flyfish-data-r2dbc/src/main/java/com/flyfish/framework/r2dbc/metadata/reference/R2dbcAssociation.java
View File

@ -1,6 +1,7 @@
package com.flyfish.framework.r2dbc.metadata.reference;
import com.flyfish.framework.annotations.Property;
import com.flyfish.framework.r2dbc.metadata.reference.field.FieldAccessor;
import lombok.AllArgsConstructor;
import lombok.Data;
import org.springframework.data.mapping.PersistentEntity;
@ -17,13 +18,13 @@ public class R2dbcAssociation {
@Property("关联字段")
private PersistentProperty<?> field;
@Property("目标字段")
private FieldSetter setter;
@Property("目标值访问器")
private FieldAccessor accessor;
@Property("字段是否在当前实体内部")
boolean inner;
public static R2dbcAssociation of(PersistentEntity<?, ?> entity, PersistentProperty<?> field, FieldSetter setter, boolean inner) {
return new R2dbcAssociation(entity, field, setter, inner);
public static R2dbcAssociation of(PersistentEntity<?, ?> entity, PersistentProperty<?> field, FieldAccessor accessor, boolean inner) {
return new R2dbcAssociation(entity, field, accessor, inner);
}
}

9
flyfish-data/flyfish-data-r2dbc/src/main/java/com/flyfish/framework/r2dbc/metadata/reference/R2dbcCollection.java
View File

@ -1,6 +1,7 @@
package com.flyfish.framework.r2dbc.metadata.reference;
import com.flyfish.framework.annotations.Property;
import com.flyfish.framework.r2dbc.metadata.reference.field.FieldAccessor;
import lombok.AllArgsConstructor;
import lombok.Data;
import org.springframework.data.mapping.PersistentEntity;
@ -17,10 +18,10 @@ public class R2dbcCollection {
@Property("关联字段")
private PersistentProperty<?> field;
@Property("目标字段")
private FieldSetter setter;
@Property("目标值访问器")
private FieldAccessor accessor;
public static R2dbcCollection of(PersistentEntity<?, ?> entity, PersistentProperty<?> field, FieldSetter setter) {
return new R2dbcCollection(entity, field, setter);
public static R2dbcCollection of(PersistentEntity<?, ?> entity, PersistentProperty<?> field, FieldAccessor accessor) {
return new R2dbcCollection(entity, field, accessor);
}
}

9
flyfish-data/flyfish-data-r2dbc/src/main/java/com/flyfish/framework/r2dbc/metadata/reference/R2dbcRelation.java
View File

@ -1,6 +1,7 @@
package com.flyfish.framework.r2dbc.metadata.reference;
import com.flyfish.framework.annotations.Property;
import com.flyfish.framework.r2dbc.metadata.reference.field.FieldAccessor;
import com.flyfish.framework.relational.mapping.Association;
import lombok.AllArgsConstructor;
import lombok.Data;
@ -28,10 +29,10 @@ public class R2dbcRelation {
@Property("关系表外部实体id字段")
private String foreignField;
@Property("目标值设置器")
private FieldSetter setter;
@Property("目标值访问器")
private FieldAccessor accessor;
public static R2dbcRelation of(PersistentEntity<?, ?> entity, Association association, FieldSetter setter) {
return new R2dbcRelation(entity, association.relationTable(), association.field(), association.foreignField(), setter);
public static R2dbcRelation of(PersistentEntity<?, ?> entity, Association association, FieldAccessor accessor) {
return new R2dbcRelation(entity, association.relationTable(), association.field(), association.foreignField(), accessor);
}
}

86
flyfish-data/flyfish-data-r2dbc/src/main/java/com/flyfish/framework/r2dbc/metadata/reference/field/DefaultFieldAccessor.java Normal file
View File

@ -0,0 +1,86 @@
package com.flyfish.framework.r2dbc.metadata.reference.field;
import com.flyfish.framework.domain.base.Domain;
import java.beans.PropertyDescriptor;
import java.lang.reflect.Method;
/**
* 默认访问器实现
*
* @author wangyu
*/
class DefaultFieldAccessor implements FieldAccessor {
private final PropertyDescriptor descriptor;
DefaultFieldAccessor(PropertyDescriptor descriptor) {
this.descriptor = descriptor;
}
/**
* 设置字段值
*
* @param obj 具体对象
* @param value 值
* @return 设置后的对象
*/
@Override
public <T extends Domain> T set(T obj, Object value) {
if (null != descriptor.getWriteMethod()) {
invokeSetter(descriptor.getWriteMethod(), obj, value);
}
return obj;
}
/**
* 取得字段值
*
* @param obj 具体对象
* @return 具体字段值
*/
@Override
public <T extends Domain, R> R get(T obj) {
if (null != descriptor.getReadMethod()) {
return invokeGetter(descriptor.getWriteMethod(), obj);
}
return null;
}
/**
* setter是否可用
*
* @return 结果
*/
@Override
public boolean hasSetter() {
return null != descriptor.getWriteMethod();
}
/**
* getter是否可用
*
* @return 结果
*/
@Override
public boolean hasGetter() {
return null != descriptor.getReadMethod();
}
private void invokeSetter(Method method, Object obj, Object value) {
try {
method.invoke(obj, value);
} catch (Exception ignored) {
}
}
@SuppressWarnings("unchecked")
private <T> T invokeGetter(Method method, Object obj) {
try {
return (T) method.invoke(obj);
} catch (Exception ignored) {
return null;
}
}
}

63
flyfish-data/flyfish-data-r2dbc/src/main/java/com/flyfish/framework/r2dbc/metadata/reference/field/FieldAccessor.java Normal file
View File

@ -0,0 +1,63 @@
package com.flyfish.framework.r2dbc.metadata.reference.field;
import com.flyfish.framework.domain.base.Domain;
import org.springframework.beans.BeanUtils;
import java.beans.PropertyDescriptor;
import java.lang.reflect.Field;
/**
* 字段访问器
*
* @author wangyu
*/
public interface FieldAccessor {
static FieldAccessor of(Class<?> type, Field field) {
PropertyDescriptor descriptor = BeanUtils.getPropertyDescriptor(type, field.getName());
return new DefaultFieldAccessor(descriptor);
}
/**
* 设置字段值
*
* @param obj 具体对象
* @param value 值
* @param <T> 对象的泛型
* @return 设置后的对象
*/
<T extends Domain> T set(T obj, Object value);
/**
* 取得字段值
*
* @param obj 具体对象
* @param <T> 对象的泛型
* @param <R> 值的泛型
* @return 具体字段值
*/
<T extends Domain, R> R get(T obj);
/**
* setter是否可用
*
* @return 结果
*/
boolean hasSetter();
/**
* getter是否可用
*
* @return 结果
*/
boolean hasGetter();
/**
* 是否可用，即支持读写
*
* @return 结果
*/
default boolean isAvailable() {
return hasGetter() && hasSetter();
}
}

47
flyfish-data/flyfish-data-r2dbc/src/main/java/com/flyfish/framework/r2dbc/metadata/visitor/MetadataHandler.java Normal file
View File

@ -0,0 +1,47 @@
package com.flyfish.framework.r2dbc.metadata.visitor;
import com.flyfish.framework.domain.base.Domain;
import com.flyfish.framework.r2dbc.metadata.reference.R2dbcAssociation;
import com.flyfish.framework.r2dbc.metadata.reference.R2dbcCollection;
import com.flyfish.framework.r2dbc.metadata.reference.R2dbcRelation;
import org.springframework.data.mapping.PersistentPropertyAccessor;
import org.springframework.lang.Nullable;
import reactor.core.publisher.Mono;
/**
* 元数据处理器
*
* @author wangyu
*/
public interface MetadataHandler {
/**
* 处理association
*
* @param association 一对一关联
* @param accessor 对象访问器
* @return 结果
*/
@Nullable
Mono<Domain> handleAssociation(R2dbcAssociation association, PersistentPropertyAccessor<Domain> accessor);
/**
* 处理collection
*
* @param collection 一对多关联
* @param accessor 对象访问器
* @return 结果
*/
@Nullable
Mono<Domain> handleCollection(R2dbcCollection collection, PersistentPropertyAccessor<Domain> accessor);
/**
* 处理relation
*
* @param relation 多对多关联
* @param accessor 访问器
* @return 结果
*/
@Nullable
Mono<Domain> handleRelation(R2dbcRelation relation, PersistentPropertyAccessor<Domain> accessor);
}

28
flyfish-data/flyfish-data-r2dbc/src/main/java/com/flyfish/framework/r2dbc/metadata/visitor/MetadataVisitor.java Normal file
View File

@ -0,0 +1,28 @@
package com.flyfish.framework.r2dbc.metadata.visitor;
import com.flyfish.framework.r2dbc.metadata.R2dbcTableMetadata;
import org.springframework.data.relational.core.sql.Visitable;
import org.springframework.data.relational.core.sql.Visitor;
import org.springframework.lang.NonNull;
/**
* 元数据访问者
*
* @author wangyu
*/
public interface MetadataVisitor extends Visitor {
/**
* 访问元数据的方法
*
* @param metadata 元数据
*/
void enter(R2dbcTableMetadata metadata);
@Override
default void enter(@NonNull Visitable segment) {
if (segment instanceof R2dbcTableMetadata) {
enter((R2dbcTableMetadata) segment);
}
}
}

112
flyfish-data/flyfish-data-r2dbc/src/main/java/com/flyfish/framework/r2dbc/metadata/visitor/R2dbcTableMetadataVisitor.java Normal file
View File

@ -0,0 +1,112 @@
package com.flyfish.framework.r2dbc.metadata.visitor;
import com.flyfish.framework.domain.base.Domain;
import com.flyfish.framework.r2dbc.metadata.R2dbcTableMetadata;
import com.flyfish.framework.r2dbc.metadata.reference.R2dbcAssociation;
import com.flyfish.framework.r2dbc.metadata.reference.R2dbcCollection;
import com.flyfish.framework.r2dbc.metadata.reference.R2dbcRelation;
import lombok.Getter;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.data.mapping.PersistentEntity;
import org.springframework.data.mapping.PersistentPropertyAccessor;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.Objects;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
/**
* r2dbc表元数据访问器
*
* @author wangyu
*/
public final class R2dbcTableMetadataVisitor implements MetadataVisitor {
@Getter
private Mono<Domain> result;
private final Domain entity;
private final MetadataHandler handler;
private PersistentPropertyAccessor<Domain> accessor;
private R2dbcTableMetadata metadata;
public R2dbcTableMetadataVisitor(Domain entity, MetadataHandler handler) {
this.entity = entity;
this.handler = handler;
this.result = Mono.just(entity);
}
/**
* 访问元数据的方法
*
* @param metadata 元数据
*/
@Override
public void enter(R2dbcTableMetadata metadata) {
// 不含有关联关系，直接跳过
if (!metadata.hasAssociation()) {
return;
}
this.metadata = metadata;
// 获取本实体的持久化对象
PersistentEntity<?, ?> persistentEntity = metadata.getEntity();
// 获取属性设置器
this.accessor = persistentEntity.getPropertyAccessor(entity);
// 尝试填充各种关联方式
List<Mono<Domain>> signals = Stream.of(visitAssociations(), visitCollections(), visitRelations())
.flatMap(Function.identity())
.filter(Objects::nonNull)
.collect(Collectors.toList());
// 尝试填充
if (CollectionUtils.isNotEmpty(signals)) {
this.result = Mono.zip(signals, objs -> entity);
}
}
/**
* 访问一对一关联
*
* @return 结果
*/
private Stream<Mono<Domain>> visitAssociations() {
List<R2dbcAssociation> associations = metadata.getAssociations();
if (CollectionUtils.isNotEmpty(associations)) {
return associations.stream().map(association -> handler.handleAssociation(association, accessor));
}
return Stream.empty();
}
/**
* 拉取一对多关联实体
*
* @return 结果
*/
private Stream<Mono<Domain>> visitCollections() {
List<R2dbcCollection> collections = metadata.getCollections();
if (CollectionUtils.isNotEmpty(collections)) {
return collections.stream().map(collection -> handler.handleCollection(collection, accessor));
}
return Stream.empty();
}
/**
* 拉取多对多关联数据
*
* @return 结果
*/
private Stream<Mono<Domain>> visitRelations() {
List<R2dbcRelation> relations = metadata.getRelations();
if (CollectionUtils.isNotEmpty(relations)) {
return relations.stream().map(relation -> handler.handleRelation(relation, accessor));
}
return Stream.empty();
}
}