feat: 重大进展，关联查询可用

flyfish-data/flyfish-data-common/src/main/java/com/flyfish/framework/domain/base/SimpleQo.java

@@ -0,0 +1,26 @@
+package com.flyfish.framework.domain.base;
+
+import com.flyfish.framework.query.QueryDefinition;
+import lombok.RequiredArgsConstructor;
+
+/**
+ * 基本查询qo
+ *
+ * @param <T> 泛型
+ */
+@RequiredArgsConstructor
+public class SimpleQo<T extends Domain> extends BaseQo<T> {
+
+    private final QueryDefinition query;
+
+    public static <T extends Domain> Qo<T> of(QueryDefinition query) {
+        return new SimpleQo<>(query);
+    }
+
+    @Override
+    public QueryDefinition queryBuilder() {
+        return query;
+    }
+}
+
+

flyfish-data/flyfish-data-common/src/main/java/com/flyfish/framework/query/QueryDefinition.java

@@ -1,6 +1,8 @@
 package com.flyfish.framework.query;
 
-import com.flyfish.framework.query.holder.QueryChainHolder;
+import com.flyfish.framework.domain.base.Domain;
+import com.flyfish.framework.domain.base.Qo;
+import com.flyfish.framework.domain.base.SimpleQo;
 import com.flyfish.framework.query.spi.adaptor.CriteriaAdaptor;
 
 /**
@@ -19,6 +21,16 @@ public interface QueryDefinition {
      */
     <T> T build(CriteriaAdaptor<T> adaptor);
 
+    /**
+     * 包装快捷方法
+     *
+     * @param <D> 实体类型泛型
+     * @return 结果
+     */
+    default <D extends Domain> Qo<D> wrap() {
+        return new SimpleQo<>(this);
+    }
+
     /**
      * 修改，此修改会直接接着查询条件进行拼接
      *

flyfish-data/flyfish-data-common/src/main/java/com/flyfish/framework/query/chain/DefaultQueryChainHolder.java

@@ -44,16 +44,17 @@ class DefaultQueryChainHolder<C> implements QueryChainHolder<C> {
     @Override
     public void with(Function<CriteriaAdaptor<C>, C> criteria) {
         Assert.notNull(operator, "连接操作不可为空");
+        Function<CriteriaAdaptor<C>, BinaryOperator<C>> finalOperator = operator;
+        this.operator = null;
         BiFunction<C, CriteriaAdaptor<C>, C> builder = (previous, adaptor) -> {
             // 连接逻辑
-            BinaryOperator<C> linker = operator.apply(adaptor);
+            BinaryOperator<C> linker = finalOperator.apply(adaptor);
             // 具体条件
             C next = criteria.apply(adaptor);
             // 之前的条件
             return linker.apply(previous, next);
         };
         builders.add(builder);
-        this.operator = null;
     }
 
     /**

flyfish-data/flyfish-data-common/src/main/java/com/flyfish/framework/repository/base/DomainRepository.java

@@ -1,5 +1,7 @@
 package com.flyfish.framework.repository.base;
 
+import org.springframework.data.mapping.PersistentEntity;
+import org.springframework.data.mapping.PersistentProperty;
 import org.springframework.data.repository.core.EntityInformation;
 
 /**
@@ -15,4 +17,12 @@ public interface DomainRepository<T> {
      * @return 结果
      */
     EntityInformation<T, String> getEntityInformation();
+
+    /**
+     * 获取持久化实体信息
+     *
+     * @param <P> 持久化属性泛型
+     * @return 结果
+     */
+    <P extends PersistentProperty<P>> PersistentEntity<T, P> getPersistentEntity();
 }

flyfish-data/flyfish-data-mongodb/src/main/java/com/flyfish/framework/mongodb/repository/impl/DefaultReactiveRepositoryImpl.java

@@ -10,7 +10,10 @@ import org.springframework.dao.IncorrectResultSizeDataAccessException;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
+import org.springframework.data.mapping.PersistentEntity;
+import org.springframework.data.mapping.PersistentProperty;
 import org.springframework.data.mongodb.core.ReactiveMongoOperations;
+import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.repository.query.MongoEntityInformation;
@@ -185,4 +188,9 @@ public class DefaultReactiveRepositoryImpl<T extends Domain> extends SimpleReact
     private Optional<Query> getQuery(Qo<T> qo) {
         return qo.getQuery(entityInformation);
     }
+
+    @Override
+    public <P extends PersistentProperty<P>> PersistentEntity<T, P> getPersistentEntity() {
+        return null;
+    }
 }

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

@@ -2,22 +2,30 @@ 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.R2dbcAssociation;
+import com.flyfish.framework.r2dbc.metadata.reference.R2dbcCollection;
 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.mapping.event.AfterConvertCallback;
+import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 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 reactor.core.publisher.Mono;
 
-import java.util.Map;
+import java.util.*;
-import java.util.Optional;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -49,18 +57,93 @@ public class ReferenceR2dbcCallback implements AfterConvertCallback<Domain> {
      * @return 结果
      */
     private Mono<Domain> doFetch(Domain entity) {
+        Class<? extends Domain> entityClass = entity.getClass();
         // 遍历fields，找到要注入的数据
-        R2dbcTableMetadata metadata = r2dbcMetadataManager.getMetadata(entity.getClass());
+        R2dbcTableMetadata metadata = r2dbcMetadataManager.getMetadata(entityClass);
-        // 尝试填充关联
+        // 获取本实体的持久化对象
-        if (CollectionUtils.isNotEmpty(metadata.getAssociations())) {
+        PersistentEntity<?, ?> persistentEntity = getRepository(entityClass).getPersistentEntity();
+        // 获取属性设置器
+        PersistentPropertyAccessor<Domain> accessor = persistentEntity.getPropertyAccessor(entity);
 
+        // 尝试填充一对一关联
+        List<Mono<Domain>> signals = new ArrayList<>();
+        signals.addAll(fetchAssociation(metadata.getAssociations(), persistentEntity, accessor));
+        signals.addAll(fetchCollections(metadata.getCollections(), persistentEntity, accessor));
+
+        // 尝试填充一对多关联
+        if (CollectionUtils.isNotEmpty(signals)) {
+            return Mono.zip(signals, objs -> entity);
         }
         return Mono.just(entity);
     }
 
+    private DefaultReactiveRepository<Domain> getRepository(Class<?> entityClass) {
+        return CastUtils.cast(repositories.get().get(entityClass));
+    }
+
     @Autowired
     public void setRepositories(ObjectProvider<DefaultReactiveRepository<?>> repositories) {
         this.repositories = Lazy.of(() -> repositories.stream()
                 .collect(Collectors.toMap(repo -> repo.getEntityInformation().getJavaType(), Function.identity())));
     }
+
+    private List<Mono<Domain>> fetchAssociation(List<R2dbcAssociation> associations, PersistentEntity<?, ?> persistentEntity,
+                                                PersistentPropertyAccessor<Domain> accessor) {
+        List<Mono<Domain>> signals = new ArrayList<>();
+        if (CollectionUtils.isNotEmpty(associations)) {
+            // 分区，将判断提在外面，提升性能
+            Map<Boolean, List<R2dbcAssociation>> partitions = associations.stream()
+                    .filter(R2dbcAssociation::isValid)
+                    .collect(Collectors.partitioningBy(R2dbcAssociation::isInner));
+            // 处理内部查询
+            if (CollectionUtils.isNotEmpty(partitions.get(true))) {
+                partitions.get(true).forEach(association -> {
+                    // 获取对方的仓库
+                    DefaultReactiveRepository<Domain> repository = getRepository(association.getEntityClass());
+                    // 内部查询，以对象内的值作为查询条件
+                    PersistentProperty<?> property = persistentEntity.getPersistentProperty(association.getField());
+                    if (null == property) return;
+                    // 得到条件值
+                    Object value = accessor.getProperty(property);
+                    if (ObjectUtils.isEmpty(value)) return;
+                    // 查询
+                    Mono<Domain> signal = repository.findById(String.valueOf(value))
+                            .map(result -> association.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.getEntityClass());
+                    // 获取目标属性名
+                    PersistentProperty<?> property = repository.getPersistentEntity().getPersistentProperty(association.getField());
+                    if (null == property) return;
+                    // 直接通过外部字段查询
+                    Mono<Domain> signal = repository.findOne(Queries.where(getColumnName(property)).eq(id).wrap())
+                            .map(result -> association.setValue(accessor.getBean(), result))
+                            .defaultIfEmpty(accessor.getBean());
+                    signals.add(signal);
+                });
+            }
+        }
+        return signals;
+    }
+
+    private List<Mono<Domain>> fetchCollections(List<R2dbcCollection> collections, PersistentEntity<?, ?> persistentEntity,
+                                                PersistentPropertyAccessor<Domain> accessor) {
+        return Collections.emptyList();
+    }
+
+    private String getColumnName(PersistentProperty<?> property) {
+        if (property instanceof RelationalPersistentProperty) {
+            return ((RelationalPersistentProperty) property).getColumnName().getReference();
+        }
+        return property.getName();
+    }
 }

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

@@ -34,8 +34,8 @@ public class R2dbcTableMetadata {
         this.entityClass = entityClass;
     }
 
-    public void addAssociation(String field, boolean inner) {
+    public void addAssociation(org.springframework.data.mapping.model.Property property, String field, boolean inner) {
-        this.associations.add(R2dbcAssociation.of(entityClass, field, inner));
+        this.associations.add(R2dbcAssociation.of(property, field, inner));
     }
 
 }

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

@@ -4,7 +4,6 @@ import com.flyfish.framework.annotations.relation.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.R2dbcAssociation;
 import org.apache.commons.lang3.ClassUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.BeanUtils;
@@ -19,7 +18,6 @@ import java.lang.reflect.Field;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Map;
-import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.function.Function;
@@ -80,6 +78,7 @@ public class SimpleR2dbcMetadataManager implements R2dbcMetadataManager {
          */
         @Override
         public void doWith(@NonNull Field field) throws IllegalArgumentException, IllegalAccessException {
+            if (!descriptors.containsKey(field.getName())) return;
             Property property = Property.of(typeInformation, field, descriptors.get(field.getName()));
             Class<?> fieldType = property.getType();
 
@@ -90,9 +89,9 @@ public class SimpleR2dbcMetadataManager implements R2dbcMetadataManager {
                 if (ClassUtils.isAssignable(fieldType, Domain.class)) {
                     // 一对一关联
                     if (StringUtils.isNotBlank(association.field())) {
-                        metadata.addAssociation(association.field(), true);
+                        metadata.addAssociation(property, association.field(), true);
                     } else if (StringUtils.isNotBlank(association.foreignField())) {
-                        metadata.addAssociation(association.foreignField(), false);
+                        metadata.addAssociation(property, association.foreignField(), false);
                     } else {
                         // 尚且不支持空白策略
                     }

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

@@ -1,25 +1,48 @@
 package com.flyfish.framework.r2dbc.metadata.reference;
 
 import com.flyfish.framework.annotations.Property;
-import com.flyfish.framework.domain.base.Domain;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 
+import java.lang.reflect.Method;
+
 @Data
 @Property("一对一关联查询")
 @AllArgsConstructor
 public class R2dbcAssociation {
 
     @Property("关联实体类")
-    private Class<? extends Domain> entityClass;
+    private Class<?> entityClass;
 
     @Property("关联字段")
     private String field;
 
+    @Property("目标字段")
+    private org.springframework.data.mapping.model.Property targetField;
+
     @Property("字段是否在当前实体内部")
     boolean inner;
 
-    public static R2dbcAssociation of(Class<? extends Domain> entityClass, String field, boolean inner) {
+    public static R2dbcAssociation of(org.springframework.data.mapping.model.Property targetField, String field, boolean inner) {
-        return new R2dbcAssociation(entityClass, field, inner);
+        return new R2dbcAssociation(targetField.getType(), field, targetField, inner);
+    }
+
+    public boolean isValid() {
+        return targetField.hasAccessor();
+    }
+
+    public <T> T setValue(T obj, Object value) {
+        if (null != value && targetField.hasAccessor()) {
+            targetField.getSetter().ifPresent(method -> invokeSetter(method, obj, value));
+        }
+        return obj;
+    }
+
+    private void invokeSetter(Method method, Object obj, Object value) {
+        try {
+            method.invoke(obj, value);
+        } catch (Exception ignored) {
+
+        }
     }
 }

flyfish-data/flyfish-data-r2dbc/src/main/java/com/flyfish/framework/r2dbc/repository/impl/DefaultReactiveRepositoryImpl.java

@@ -236,6 +236,12 @@ public class DefaultReactiveRepositoryImpl<T extends Domain> extends SimpleR2dbc
         return entity;
     }
 
+    @SuppressWarnings("unchecked")
+    @Override
+    public RelationalPersistentEntity<T> getPersistentEntity() {
+        return persistentEntity.get();
+    }
+
     private Mono<Query> getQuery(Qo<T> qo) {
         return Mono.justOrEmpty(qo.getQuery(entity));
     }

flyfish-data/flyfish-data-r2dbc/src/test/java/com/flyfish/framework/r2dbc/R2DbcRepositoryTest.java

@@ -1,8 +1,9 @@
 package com.flyfish.framework.r2dbc;
 
+import com.alibaba.fastjson.JSON;
 import com.flyfish.framework.r2dbc.config.R2dbcDataConfig;
-import com.flyfish.framework.r2dbc.repository.TestDO;
+import com.flyfish.framework.r2dbc.domain.TestDO;
-import com.flyfish.framework.r2dbc.repository.TestQO;
+import com.flyfish.framework.r2dbc.domain.TestQO;
 import com.flyfish.framework.r2dbc.repository.TestRepository;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -35,7 +36,7 @@ public class R2DbcRepositoryTest {
         test.setName("测试名称");
         test.setOtherId("1");
 
-        System.out.println(testRepository.insert(test).block());
+        System.out.println(JSON.toJSONString(testRepository.insert(test).block()));
-        System.out.println(testRepository.findAll(qo).collectList().block());
+        System.out.println(JSON.toJSONString(testRepository.findAll(qo).collectList().block()));
     }
 }

flyfish-data/flyfish-data-r2dbc/src/test/java/com/flyfish/framework/r2dbc/domain/TestDO.java

@@ -1,9 +1,10 @@
-package com.flyfish.framework.r2dbc.repository;
+package com.flyfish.framework.r2dbc.domain;
 
 import com.flyfish.framework.annotations.relation.Association;
 import com.flyfish.framework.domain.base.AuditDomain;
-import com.flyfish.framework.r2dbc.repository.reference.TestAsso;
+import com.flyfish.framework.r2dbc.domain.reference.TestAsso;
-import com.flyfish.framework.r2dbc.repository.reference.TestChild;
+import com.flyfish.framework.r2dbc.domain.reference.TestChild;
+import com.flyfish.framework.r2dbc.domain.reference.TestOther;
 import lombok.Data;
 import org.springframework.data.relational.core.mapping.Column;
 import org.springframework.data.relational.core.mapping.Table;
@@ -17,9 +18,12 @@ public class TestDO extends AuditDomain {
     @Column("other_id")
     private String otherId;
 
-    @Association(field = "other_id")
+    @Association(field = "otherId")
     private TestAsso asso;
 
-    @Association(foreignField = "parent_id")
+    @Association(foreignField = "testId")
+    private TestOther other;
+
+    @Association(foreignField = "parentId")
     private List<TestChild> children;
 }

flyfish-data/flyfish-data-r2dbc/src/test/java/com/flyfish/framework/r2dbc/domain/TestQO.java

@@ -1,4 +1,4 @@
-package com.flyfish.framework.r2dbc.repository;
+package com.flyfish.framework.r2dbc.domain;
 
 import com.flyfish.framework.domain.base.NameLikeQo;
 

flyfish-data/flyfish-data-r2dbc/src/test/java/com/flyfish/framework/r2dbc/domain/reference/TestAsso.java

@@ -1,8 +1,11 @@
-package com.flyfish.framework.r2dbc.repository.reference;
+package com.flyfish.framework.r2dbc.domain.reference;
 
 import com.flyfish.framework.domain.base.AuditDomain;
+import lombok.Data;
 import org.springframework.data.relational.core.mapping.Table;
 
 @Table("test_asso")
+@Data
 public class TestAsso extends AuditDomain {
+
 }

flyfish-data/flyfish-data-r2dbc/src/test/java/com/flyfish/framework/r2dbc/domain/reference/TestChild.java

@@ -1,4 +1,4 @@
-package com.flyfish.framework.r2dbc.repository.reference;
+package com.flyfish.framework.r2dbc.domain.reference;
 
 import com.flyfish.framework.domain.base.AuditDomain;
 import lombok.Data;

flyfish-data/flyfish-data-r2dbc/src/test/java/com/flyfish/framework/r2dbc/domain/reference/TestOther.java

@@ -0,0 +1,16 @@
+package com.flyfish.framework.r2dbc.domain.reference;
+
+import com.flyfish.framework.annotations.relation.Association;
+import com.flyfish.framework.domain.base.AuditDomain;
+import com.flyfish.framework.r2dbc.domain.TestDO;
+import lombok.Data;
+import org.springframework.data.relational.core.mapping.Column;
+import org.springframework.data.relational.core.mapping.Table;
+
+@Table("test_other")
+@Data
+public class TestOther extends AuditDomain {
+
+    @Column("test_id")
+    private String testId;
+}

flyfish-data/flyfish-data-r2dbc/src/test/java/com/flyfish/framework/r2dbc/repository/TestAssoRepository.java

@@ -0,0 +1,10 @@
+package com.flyfish.framework.r2dbc.repository;
+
+import com.flyfish.framework.r2dbc.domain.reference.TestAsso;
+import com.flyfish.framework.repository.DefaultReactiveRepository;
+
+/**
+ * 测试仓库
+ */
+public interface TestAssoRepository extends DefaultReactiveRepository<TestAsso> {
+}

flyfish-data/flyfish-data-r2dbc/src/test/java/com/flyfish/framework/r2dbc/repository/TestChildRepository.java

@@ -0,0 +1,10 @@
+package com.flyfish.framework.r2dbc.repository;
+
+import com.flyfish.framework.r2dbc.domain.reference.TestChild;
+import com.flyfish.framework.repository.DefaultReactiveRepository;
+
+/**
+ * 测试仓库
+ */
+public interface TestChildRepository extends DefaultReactiveRepository<TestChild> {
+}

flyfish-data/flyfish-data-r2dbc/src/test/java/com/flyfish/framework/r2dbc/repository/TestOtherRepository.java

@@ -0,0 +1,10 @@
+package com.flyfish.framework.r2dbc.repository;
+
+import com.flyfish.framework.r2dbc.domain.reference.TestOther;
+import com.flyfish.framework.repository.DefaultReactiveRepository;
+
+/**
+ * 测试仓库
+ */
+public interface TestOtherRepository extends DefaultReactiveRepository<TestOther> {
+}

flyfish-data/flyfish-data-r2dbc/src/test/java/com/flyfish/framework/r2dbc/repository/TestRepository.java

@@ -1,5 +1,6 @@
 package com.flyfish.framework.r2dbc.repository;
 
+import com.flyfish.framework.r2dbc.domain.TestDO;
 import com.flyfish.framework.repository.DefaultReactiveRepository;
 
 /**

flyfish-data/flyfish-data-r2dbc/src/test/resources/schema.sql

@@ -61,3 +61,24 @@ CREATE TABLE IF NOT EXISTS `test_asso`
 REPLACE INTO `test_asso`
 VALUES ('1', '1', 'baba ', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, null, null, null, null, false),
        ('2', '1', 'mama ', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, null, null, null, null, false);
+
+
+CREATE TABLE IF NOT EXISTS `test_other`
+(
+    `id`          VARCHAR(36)  NOT NULL COMMENT '主键',
+    `code`        VARCHAR(32)  NOT NULL COMMENT '编码',
+    `name`        VARCHAR(100) NOT NULL COMMENT '名称',
+    `test_id`     VARCHAR(36)  NOT NULL COMMENT '外部主键',
+    `create_time` DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+    `modify_time` DATETIME     NOT NULL COMMENT '修改时间',
+    `creator`     VARCHAR(36)  NULL COMMENT '创建人名称',
+    `creator_id`  VARCHAR(36)  NULL COMMENT '创建人id',
+    `modifier`    VARCHAR(36)  NULL COMMENT '修改人名称',
+    `modifier_id` VARCHAR(36)  NULL COMMENT '修改人id',
+    `delete`      BIT(1)       NOT NULL DEFAULT b'0',
+    PRIMARY KEY (`id`)
+    ) COMMENT '测试关联表';
+
+REPLACE INTO `test_other`
+VALUES ('1', '1', 'baba ', 1, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, null, null, null, null, false),
+       ('2', '1', 'mama ', 2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, null, null, null, null, false);