feat: support get all org-members page by page

Author: lvhuichao

server/api-service/openblocks-domain/src/main/java/com/openblocks/domain/organization/service/OrgMemberService.java

@@ -6,16 +6,16 @@
 
 import com.openblocks.domain.organization.model.MemberRole;
 import com.openblocks.domain.organization.model.OrgMember;
-import com.openblocks.infra.annotation.NonEmptyMono;
 import com.openblocks.infra.annotation.PossibleEmptyMono;
 
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 public interface OrgMemberService {
 
-    @NonEmptyMono
-    Mono<List<OrgMember>> getOrganizationMembers(String orgId, int page, int count);
+    Flux<OrgMember> getOrganizationMembers(String orgId);
+
+    Flux<OrgMember> getOrganizationMembers(String orgId, int page, int count);
 
     Mono<OrgMember> getCurrentOrgMember(String userId);
 

server/api-service/openblocks-domain/src/main/java/com/openblocks/domain/organization/service/OrgMemberServiceImpl.java

@@ -10,6 +10,7 @@
 import org.apache.commons.lang3.StringUtils;
 import org.bson.Document;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.domain.PageRequest;
 import org.springframework.stereotype.Service;
 
 import com.openblocks.domain.group.model.GroupMember;
@@ -24,6 +25,7 @@
 import com.openblocks.infra.birelation.BiRelation;
 import com.openblocks.infra.birelation.BiRelationService;
 import com.openblocks.infra.mongo.MongoUpsertHelper;
+import com.openblocks.infra.util.FluxHelper;
 import com.openblocks.sdk.config.CommonConfig;
 import com.openblocks.sdk.config.CommonConfig.Workspace;
 import com.openblocks.sdk.constants.WorkspaceMode;
@@ -55,10 +57,15 @@ public class OrgMemberServiceImpl implements OrgMemberService {
     private MongoUpsertHelper mongoUpsertHelper;
 
     @Override
-    public Mono<List<OrgMember>> getOrganizationMembers(String orgId, int page, int count) {
-        return biRelationService.getBySourceId(ORG_MEMBER, orgId)
-                .map(OrgMember::from)
-                .collectList();
+    public Flux<OrgMember> getOrganizationMembers(String orgId) {
+        return FluxHelper.getAllPageByPage(pageable -> biRelationService.getBySourceId(ORG_MEMBER, orgId, pageable), PageRequest.ofSize(100))
+                .map(OrgMember::from);
+    }
+
+    @Override
+    public Flux<OrgMember> getOrganizationMembers(String orgId, int page, int count) {
+        return biRelationService.getBySourceId(ORG_MEMBER, orgId, PageRequest.of(page, count))
+                .map(OrgMember::from);
     }
 
     /**

server/api-service/openblocks-domain/src/test/java/com/openblocks/domain/organization/service/OrgMemberServiceTest.java

@@ -0,0 +1,48 @@
+package com.openblocks.domain.organization.service;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import org.apache.commons.lang3.RandomUtils;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import com.openblocks.domain.organization.model.MemberRole;
+import com.openblocks.domain.organization.model.OrgMember;
+import com.openblocks.sdk.util.IDUtils;
+
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+@SpringBootTest
+@RunWith(SpringRunner.class)
+public class OrgMemberServiceTest {
+
+    @Autowired
+    private OrgMemberService orgMemberService;
+
+    @Test
+    public void testGetAllOrganizationMembers() {
+        int totalCount = RandomUtils.nextInt(100, 500);
+        String orgId = IDUtils.generate();
+        Set<String> userIds = IntStream.rangeClosed(1, totalCount).mapToObj(i -> IDUtils.generate()).collect(Collectors.toSet());
+        Mono<List<String>> listMono = orgMemberService.bulkAddMember(orgId, userIds, MemberRole.MEMBER)
+                .thenMany(orgMemberService.getOrganizationMembers(orgId))
+                .map(OrgMember::getUserId)
+                .collectList();
+
+        StepVerifier.create(listMono)
+                .assertNext(list -> {
+                    Assert.assertEquals(totalCount, list.size());
+                    Assert.assertEquals(userIds, new HashSet<>(list));
+                })
+                .verifyComplete();
+    }
+}

server/api-service/openblocks-infra/src/main/java/com/openblocks/infra/birelation/BiRelationRepository.java

@@ -2,6 +2,7 @@
 
 import java.util.Collection;
 
+import org.springframework.data.domain.Pageable;
 import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
 
 import reactor.core.publisher.Flux;
@@ -11,6 +12,8 @@ public interface BiRelationRepository extends ReactiveMongoRepository<BiRelation
 
     Flux<BiRelation> findByBizTypeAndSourceId(BiRelationBizType bizType, String sourceId);
 
+    Flux<BiRelation> findByBizTypeAndSourceId(BiRelationBizType bizType, String sourceId, Pageable pageable);
+
     Flux<BiRelation> findByBizTypeAndSourceIdIn(BiRelationBizType bizType, Collection<String> sourceId);
 
     Flux<BiRelation> findByBizTypeAndTargetId(BiRelationBizType bizType, String targetId);

server/api-service/openblocks-infra/src/main/java/com/openblocks/infra/birelation/BiRelationService.java

@@ -8,6 +8,7 @@
 import java.util.List;
 
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.domain.Pageable;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.stereotype.Service;
@@ -89,6 +90,10 @@ public Flux<BiRelation> getBySourceId(BiRelationBizType bizType, String sourceId
         return biRelationRepository.findByBizTypeAndSourceId(bizType, sourceId);
     }
 
+    public Flux<BiRelation> getBySourceId(BiRelationBizType bizType, String sourceId, Pageable pageable) {
+        return biRelationRepository.findByBizTypeAndSourceId(bizType, sourceId, pageable);
+    }
+
     public Flux<BiRelation> getByTargetIds(BiRelationBizType bizType, Collection<String> targetId) {
         return biRelationRepository.findByBizTypeAndTargetIdIn(bizType, targetId);
     }

server/api-service/openblocks-infra/src/main/java/com/openblocks/infra/util/FluxHelper.java

@@ -0,0 +1,24 @@
+package com.openblocks.infra.util;
+
+import org.springframework.data.domain.Pageable;
+
+import reactor.core.publisher.Flux;
+
+public class FluxHelper {
+
+    public static <T> Flux<T> getAllPageByPage(PageFetcher<T> pageFetcher, Pageable first) {
+        Flux<T> currentFluxPage = pageFetcher.fetch(first).cache();
+        return currentFluxPage.hasElements()
+                .flatMapMany(hasElement -> {
+                    if (hasElement) {
+                        return currentFluxPage.concatWith(getAllPageByPage(pageFetcher, first.next()));
+                    }
+                    return currentFluxPage;
+                });
+    }
+
+    public interface PageFetcher<T> {
+
+        Flux<T> fetch(Pageable pageable);
+    }
+}

server/api-service/openblocks-plugins/restApiPlugin/src/main/java/com/openblocks/plugin/restapi/model/RestApiQueryConfig.java

@@ -16,7 +16,9 @@
 
 import lombok.Builder;
 import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
 
+@Slf4j
 @Getter
 @Builder
 public class RestApiQueryConfig {
@@ -44,6 +46,7 @@ private RestApiQueryConfig(HttpMethod httpMethod, boolean disableEncodingParams,
     public static RestApiQueryConfig from(Map<String, Object> queryConfigs) {
         RestApiQueryConfig queryConfig = fromJson(toJson(queryConfigs), RestApiQueryConfig.class);
         if (queryConfig == null) {
+            log.error("deserialize query config fail:{}", toJson(queryConfigs));
             throw new PluginException(INVALID_QUERY_SETTINGS, "INVALID_RESTAPI");
         }
         return queryConfig;

server/api-service/openblocks-server/src/main/java/com/openblocks/api/authentication/service/AuthenticationApiServiceImpl.java

@@ -247,8 +247,7 @@ public Mono<Boolean> disableAuthConfig(String authId) {
 
     private Mono<Void> removeTokensByAuthId(String authId) {
         return sessionUserService.getVisitorOrgMemberCache()
-                .flatMap(orgMember -> orgMemberService.getOrganizationMembers(orgMember.getOrgId(), 0, Integer.MAX_VALUE))
-                .flatMapIterable(Function.identity())
+                .flatMapMany(orgMember -> orgMemberService.getOrganizationMembers(orgMember.getOrgId()))
                 .map(OrgMember::getUserId)
                 .flatMap(userId -> userApiService.getTokensByAuthId(userId, authId))
                 .delayUntil(token -> sessionUserService.removeUserSession(token))

server/api-service/openblocks-server/src/main/java/com/openblocks/api/usermanagement/OrgApiServiceImpl.java

@@ -87,6 +87,7 @@ public Mono<OrgMemberListView> getOrganizationMembers(String orgId, int page, in
 
     private Mono<OrgMemberListView> getOrgMemberListView(String orgId, int page, int count) {
         return orgMemberService.getOrganizationMembers(orgId, page, count)
+                .collectList()
                 .flatMap(orgMembers -> {
                     List<String> userIds = orgMembers.stream()
                             .map(OrgMember::getUserId)
@@ -97,6 +98,7 @@ private Mono<OrgMemberListView> getOrgMemberListView(String orgId, int page, int
                             .map(orgMember -> {
                                 User user = map.get(orgMember.getUserId());
                                 if (user == null) {
+                                    log.warn("user {} not exist and will be removed from the result.", orgMember.getUserId());
                                     return null;
                                 }
                                 return build(user, orgMember);