io.reactivex.Single#zip ( )源码实例Demo

protected Single<Map<ReferenceType, Map<Permission, Set<Acl>>>> findAllPermissions(User user, String organizationId, String environmentId, String domainId, String applicationId) {

        List<Single<Map<Permission, Set<Acl>>>> permissionObs = new ArrayList<>();

        permissionObs.add(applicationId != null ? permissionService.findAllPermissions(user, ReferenceType.APPLICATION, applicationId) : Single.just(emptyMap()));
        permissionObs.add(domainId != null ? permissionService.findAllPermissions(user, ReferenceType.DOMAIN, domainId) : Single.just(emptyMap()));
        permissionObs.add(environmentId != null ? permissionService.findAllPermissions(user, ReferenceType.ENVIRONMENT, environmentId) : Single.just(emptyMap()));
        permissionObs.add(organizationId != null ? permissionService.findAllPermissions(user, ReferenceType.ORGANIZATION, organizationId) : Single.just(emptyMap()));

        return Single.zip(permissionObs, objects -> {
            Map<ReferenceType, Map<Permission, Set<Acl>>> permissionsPerType = new HashMap<>();
            permissionsPerType.put(ReferenceType.APPLICATION, (Map<Permission, Set<Acl>>) objects[0]);
            permissionsPerType.put(ReferenceType.DOMAIN, (Map<Permission, Set<Acl>>) objects[1]);
            permissionsPerType.put(ReferenceType.ENVIRONMENT, (Map<Permission, Set<Acl>>) objects[2]);
            permissionsPerType.put(ReferenceType.ORGANIZATION, (Map<Permission, Set<Acl>>) objects[3]);

            return permissionsPerType;
        });
    }
 

@Override public Single<List<PlexItem>> browseMediaType(MediaType mt, int offset) {
  Single<List<PlexItem>> browseItems;

  if (mt.type() == Type.ARTIST) {
    browseItems = browseArtists(mt, offset);
  } else if (mt.type() == Type.ALBUM) {
    browseItems = browseAlbums(mt, offset);
  } else {
    browseItems = browseTracks(mt, offset);
  }

  return Single.zip(browseHeaders(mt), browseItems, (headers, items) -> {
    List<PlexItem> plexItems = new ArrayList<>();

    for (int i = 0; i < items.size(); ++i) {
      // The headers need to be offset by the current offset!
      if (headers.containsKey(i + offset)) {
        plexItems.add(headers.get(i + offset));
      }
      plexItems.add(items.get(i));
    }

    return plexItems;
  });
}
 

@Override
protected void callApi(int page, OnCallApiDone<Repo> onCallApiDone) {
    Single<Pageable<Repo>> single;
    if (staredCount != null && (staredCount.getValue() == null || staredCount.getValue() < 0)) {
         single = Single.zip(userRestService.getStarred(targetUser, page),
                userRestService.getStarredCount(targetUser), (repoPageable, count) -> {
                    if (count != null){
                        staredCount.setValue(count.getLast());
                    }
                    return repoPageable;
                });
    } else {
        single = userRestService.getStarred(targetUser, page);
    }
    execute(true, single, repoPageable -> {
        onCallApiDone.onDone(repoPageable.getLast(), page == 1, repoPageable.getItems());
    });
}
 

@Override
public Single<List<PlaylistEntity>> getPlaylistsBy(List<String> categories) {
  if (categories != null) {
    Single<List<PlaylistEntity>> start = Single.just(new LinkedList<>());
    for (String category : categories) {
      start = Single.zip(start, service.searchPlaylists(PlaylistEntity
              .Filter.start()
              .byName(category)
              .limit(100)
              .createOptions())
              .onErrorResumeNext(Single.just(new ArrayList<>())), (first, second) -> {
        if (second != null) {
          first.addAll(second);
        }
        return first;
      });
    }
    return start.map(filter::filterPlaylists);
  }
  return Single.error(new IllegalArgumentException("categories are null"));
}
 

@Override
public Single<List<TrackEntity>> getTracksBy(List<String> categories) {
  if (categories != null) {
    Single<List<TrackEntity>> start = Single.just(new LinkedList<>());
    for (String category : categories) {
      start = Single.zip(start, service.searchTracks(TrackEntity
              .Filter.start()
              .byTags(category)
              .createOptions())
              .onErrorResumeNext(Single.just(new ArrayList<>())), (first, second) -> {
        if (second != null) {
          first.addAll(second);
        }
        return first;
      });
    }
    return start.map(filter::filterTracks);
  }
  return Single.error(new IllegalArgumentException("categories are null"));
}
 

@Override
public Single<UserDetailsEntity> getUserBy(String id) {
  if (id != null) {
    Single<List<TrackEntity>> singleTracks = service.fetchUserTracks(id)
            .subscribeOn(schedulerProvider.multi());
    Single<List<PlaylistEntity>> singlePlaylists = service.fetchUserPlaylists(id)
            .subscribeOn(schedulerProvider.multi());
    Single<UserEntity> singleUser = service.fetchUser(id)
            .subscribeOn(schedulerProvider.multi());
    return Single.zip(singleUser,
            singleTracks.onErrorResumeNext(Single.just(new ArrayList<>())),
            singlePlaylists.onErrorResumeNext(Single.just(new ArrayList<>())),
            (user, tracks, playlists) -> {
              UserDetailsEntity userDetails = new UserDetailsEntity();
              userDetails.setUserEntity(user);
              userDetails.setTracks(filter.filterTracks(tracks));
              userDetails.setPlaylists(filter.filterPlaylists(playlists));
              return userDetails;
            });

  }
  return Single.error(new IllegalArgumentException("id is null"));
}
 

@Override
public Single<Map<String, Map<String, Object>>> getMetadata(List<Resource> resources) {
    if (resources == null || resources.isEmpty()) {
        return Single.just(Collections.emptyMap());
    }

    List<String> userIds = resources.stream().filter(resource -> resource.getUserId() != null).map(Resource::getUserId).distinct().collect(Collectors.toList());
    List<String> appIds = resources.stream().filter(resource -> resource.getClientId() != null).map(Resource::getClientId).distinct().collect(Collectors.toList());

    return Single.zip(userService.findByIdIn(userIds), applicationService.findByIdIn(appIds), (users, apps) -> {
        Map<String, Map<String, Object>> metadata = new HashMap<>();
        metadata.put("users", users.stream().collect(Collectors.toMap(User::getId, this::filter)));
        metadata.put("applications", apps.stream().collect(Collectors.toMap(Application::getId, this::filter)));
        return metadata;
    });
}
 

@Override
public Single<Map<String, Map<String, Object>>> getMetadata(List<Membership> memberships) {
    if (memberships == null || memberships.isEmpty()) {
        return Single.just(Collections.emptyMap());
    }

    List<String> userIds = memberships.stream().filter(membership -> MemberType.USER.equals(membership.getMemberType())).map(Membership::getMemberId).distinct().collect(Collectors.toList());
    List<String> groupIds = memberships.stream().filter(membership -> MemberType.GROUP.equals(membership.getMemberType())).map(Membership::getMemberId).distinct().collect(Collectors.toList());
    List<String> roleIds = memberships.stream().map(Membership::getRoleId).distinct().collect(Collectors.toList());

    return Single.zip(userService.findByIdIn(userIds), groupService.findByIdIn(groupIds), roleService.findByIdIn(roleIds), (users, groups, roles) -> {
        Map<String, Map<String, Object>> metadata = new HashMap<>();
        metadata.put("users", users.stream().collect(Collectors.toMap(io.gravitee.am.model.User::getId, this::convert)));
        metadata.put("groups", groups.stream().collect(Collectors.toMap(Group::getId, this::convert)));
        metadata.put("roles", roles.stream().collect(Collectors.toMap(Role::getId, this::filter)));
        return metadata;
    });
}
 

/**
 * Enhance user information with roles and groups if the access token contains those scopes
 * @param user The end user
 * @param accessToken The access token with required scopes
 * @return enhanced user
 */
private Single<User> enhance(User user, JWT accessToken) {
    return Single.zip(
            loadRoles(user, accessToken) ? roleService.findByIdIn(user.getRoles()).map(Optional::of) : Single.just(Optional.<Set<Role>>empty()),
            loadGroups(accessToken) ? groupService.findByMember(user.getId()).map(Optional::of) : Single.just(Optional.<List<Group>>empty()),
            (optionalRoles, optionalGroups) -> {
                Map<String, Object> userClaims = user.getAdditionalInformation() == null ? new HashMap<>() : new HashMap<>(user.getAdditionalInformation());
                if (optionalRoles.isPresent() && !optionalRoles.get().isEmpty()) {
                    Set<Role> roles = optionalRoles.get();
                    userClaims.putIfAbsent(CustomClaims.ROLES, roles.stream().map(Role::getName).collect(Collectors.toList()));
                }
                if (optionalGroups.isPresent() && !optionalGroups.get().isEmpty()) {
                    List<Group> groups = optionalGroups.get();
                    userClaims.putIfAbsent(CustomClaims.GROUPS, groups.stream().map(Group::getName).collect(Collectors.toList()));
                }
                user.setAdditionalInformation(userClaims);
                return user;
            });
}
 

@Override
public Single<Page<Application>> search(String domain, String query, int page, int size) {
    // currently search on client_id field
    Bson searchQuery = or(eq(FIELD_CLIENT_ID, query), eq(FIELD_NAME, query));
    // if query contains wildcard, use the regex query
    if (query.contains("*")) {
        String compactQuery = query.replaceAll("\\*+", ".*");
        String regex = "^" + compactQuery;
        Pattern pattern = Pattern.compile(regex, Pattern.CASE_INSENSITIVE);
        searchQuery = or(new BasicDBObject(FIELD_CLIENT_ID, pattern), new BasicDBObject(FIELD_NAME, pattern));
    }

    Bson mongoQuery = and(
            eq(FIELD_DOMAIN, domain),
            searchQuery);

    Single<Long> countOperation = Observable.fromPublisher(applicationsCollection.countDocuments(mongoQuery)).first(0l);
    Single<Set<Application>> applicationsOperation = Observable.fromPublisher(applicationsCollection.find(mongoQuery).sort(new BasicDBObject(FIELD_UPDATED_AT, -1)).skip(size * page).limit(size)).map(this::convert).collect(HashSet::new, Set::add);
    return Single.zip(countOperation, applicationsOperation, (count, applications) -> new Page<>(applications, page, count));
}
 

private Single<JsonObject> collectTemperatures() {
  Single<HttpResponse<JsonObject>> r1 = fetchTemperature(3000);
  Single<HttpResponse<JsonObject>> r2 = fetchTemperature(3001);
  Single<HttpResponse<JsonObject>> r3 = fetchTemperature(3002);

  return Single.zip(r1, r2, r3, (j1, j2, j3) -> {
    JsonArray array = new JsonArray()
      .add(j1.body())
      .add(j2.body())
      .add(j3.body());
    return new JsonObject().put("data", array);
  });
}
 

@Override
public Single<Page<Audit>> search(ReferenceType referenceType, String referenceId, AuditReportableCriteria criteria, int page, int size) {
    // build query
    Bson query = query(referenceType, referenceId, criteria);

    // run search query
    Single<Long> countOperation = Observable.fromPublisher(reportableCollection.countDocuments(query)).first(0l);
    Single<List<Audit>> auditsOperation = Observable.fromPublisher(reportableCollection.find(query).sort(new BasicDBObject(FIELD_TIMESTAMP, -1)).skip(size * page).limit(size)).map(this::convert).collect(LinkedList::new, List::add);
    return Single.zip(countOperation, auditsOperation, (count, audits) -> new Page<>(audits, page, count));
}
 

private SingleTransformer<Token[], Token[]> attachDefaultTokens(Wallet wallet)
{
    return upstream -> Single.zip(
            upstream, ethereumNetworkRepository.getBlankOverrideTokens(wallet),
            (tokens, defaultTokens) ->
            {
                List<Token> result = mergeTokens(tokens, defaultTokens);
                return result.toArray(new Token[0]);
            });
}
 

private SingleTransformer<Token[], Token[]> attachEthereumStored(Wallet wallet)
{
    return upstream -> Single.zip(
            upstream, attachCachedEth(wallet),
            (tokens, ethTokens) ->
            {
                List<Token> result = new ArrayList<>();
                result.addAll(ethTokens);
                for (Token t : tokens) if (!t.isEthereum()) result.add(t);
                return result.toArray(new Token[0]);
            });
}
 

@Override
public Single<? extends WhitelistState> startFetchWhitelistState() {
    return Single.zip(FFMService.getDiscussWhitelist(), OpenQQService.getDiscussesInfo(),
            (state, groups) -> {
                state.generateStates(groups);
                return state;
            });
}
 

public static <T> Single<DetailBean<T>> toCommentDetail(Single<T> detailSingle,
                                            Single<List<CommentBean>> bestCommentsSingle,
                                            Single<List<CommentBean>> commentsSingle){
    return Single.zip(detailSingle, bestCommentsSingle, commentsSingle,
            new Function3<T, List<CommentBean>, List<CommentBean>, DetailBean<T>>() {
                @Override
                public DetailBean<T> apply(T t, List<CommentBean> commentBeen,
                                           List<CommentBean> commentBeen2) throws Exception {
                    return new DetailBean<T>(t,commentBeen,commentBeen2);
                }
            });
}
 

public Single<SmsConfig> getSmsModuleConfig() {
    return Single.zip(
            localDbRepository.isModuleEnabled(),
            localDbRepository.getGatewayNumber(),
            localDbRepository.getWaitingForResultEnabled(),
            localDbRepository.getConfirmationSenderNumber(),
            localDbRepository.getWaitingResultTimeout(),
            SmsConfig::new
    );
}
 

@Override
public Single<Page<Application>> findAll(int page, int size) {
    Single<Long> countOperation = Observable.fromPublisher(applicationsCollection.countDocuments()).first(0l);
    Single<Set<Application>> applicationsOperation = Observable.fromPublisher(applicationsCollection.find().sort(new BasicDBObject(FIELD_UPDATED_AT, -1)).skip(size * page).limit(size)).map(this::convert).collect(HashSet::new, Set::add);
    return Single.zip(countOperation, applicationsOperation, (count, applications) -> new Page<>(applications, page, count));
}
 

@Override
public Single<Page<Application>> findByDomain(String domain, int page, int size) {
    Single<Long> countOperation = Observable.fromPublisher(applicationsCollection.countDocuments(eq(FIELD_DOMAIN, domain))).first(0l);
    Single<Set<Application>> applicationsOperation = Observable.fromPublisher(applicationsCollection.find(eq(FIELD_DOMAIN, domain)).sort(new BasicDBObject(FIELD_UPDATED_AT, -1)).skip(size * page).limit(size)).map(this::convert).collect(HashSet::new, Set::add);
    return Single.zip(countOperation, applicationsOperation, (count, applications) -> new Page<>(applications, page, count));
}
 

@Override
public Single<Page<User>> findAll(ReferenceType referenceType, String referenceId, int page, int size) {
    Single<Long> countOperation = Observable.fromPublisher(usersCollection.countDocuments(and(eq(FIELD_REFERENCE_TYPE, referenceType.name()), eq(FIELD_REFERENCE_ID, referenceId)))).first(0l);
    Single<Set<User>> usersOperation = Observable.fromPublisher(usersCollection.find(and(eq(FIELD_REFERENCE_TYPE, referenceType.name()), eq(FIELD_REFERENCE_ID, referenceId))).sort(new BasicDBObject(FIELD_USERNAME, 1)).skip(size * page).limit(size)).map(this::convert).collect(LinkedHashSet::new, Set::add);
    return Single.zip(countOperation, usersOperation, (count, users) -> new Page<>(users, page, count));
}