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

public Single<String> encryptAuthorization(String signedJwt, Client client) {
    //Return input without encryption if client does not require JWE or algorithm is set to none
    if (client.getAuthorizationEncryptedResponseAlg() == null || JWEAlgorithm.NONE.getName().equals(client.getAuthorizationEncryptedResponseAlg())) {
        return Single.just(signedJwt);
    }

    JWEObject jwe = new JWEObject(
            new JWEHeader.Builder(
                    JWEAlgorithm.parse(client.getAuthorizationEncryptedResponseAlg()),
                    EncryptionMethod.parse(client.getAuthorizationEncryptedResponseEnc()!=null?client.getAuthorizationEncryptedResponseEnc(): JWAlgorithmUtils.getDefaultAuthorizationResponseEnc())
            ).contentType("JWT").build(),
            new Payload(signedJwt)
    );

    return encrypt(jwe,client)
            .onErrorResumeNext(throwable -> {
                if(throwable instanceof OAuth2Exception) {
                    return Single.error(throwable);
                }
                LOGGER.error(throwable.getMessage(), throwable);
                return Single.error(new ServerErrorException("Unable to encrypt authorization"));
            });
}
 

private Single<Domain> upgradeDomain(Domain domain) {
    if(domain.getOidc()!=null) {
        return Single.just(domain);
    }

    PatchClientRegistrationSettings clientRegistrationPatch = new PatchClientRegistrationSettings();
    clientRegistrationPatch.setDynamicClientRegistrationEnabled(Optional.of(false));
    clientRegistrationPatch.setOpenDynamicClientRegistrationEnabled(Optional.of(false));
    clientRegistrationPatch.setAllowHttpSchemeRedirectUri(Optional.of(true));
    clientRegistrationPatch.setAllowLocalhostRedirectUri(Optional.of(true));
    clientRegistrationPatch.setAllowWildCardRedirectUri(Optional.of(true));

    PatchOIDCSettings oidcPatch = new PatchOIDCSettings();
    oidcPatch.setClientRegistrationSettings(Optional.of(clientRegistrationPatch));

    PatchDomain patchDomain = new PatchDomain();
    patchDomain.setOidc(Optional.of(oidcPatch));

    return domainService.patch(domain.getId(),patchDomain);
}
 

@Path("nego")
@GET
public Single<Template> nego() {
	return Single.just(new Template("templates/nego")
			.set("title", "my title")
			.set("message", "my message"));
}
 

private Single<DynamicClientRegistrationRequest> validateAuthorizationSigningAlgorithm(DynamicClientRegistrationRequest request) {
    // Signing an authorization response is required
    // As per https://bitbucket.org/openid/fapi/src/master/Financial_API_JWT_Secured_Authorization_Response_Mode.md#markdown-header-5-client-metadata
    // If unspecified, the default algorithm to use for signing authorization responses is RS256. The algorithm none is not allowed.
    if (request.getAuthorizationSignedResponseAlg() == null || !request.getAuthorizationSignedResponseAlg().isPresent()) {
        request.setAuthorizationSignedResponseAlg(Optional.of(JWSAlgorithm.RS256.getName()));
    }

    if (!JWAlgorithmUtils.isValidAuthorizationSigningAlg(request.getAuthorizationSignedResponseAlg().get())) {
        return Single.error(new InvalidClientMetadataException("Unsupported authorization signing algorithm"));
    }

    return Single.just(request);
}
 

private Single<DynamicClientRegistrationRequest> validateUserinfoSigningAlgorithm(DynamicClientRegistrationRequest request) {
    //if userinfo_signed_response_alg is provided, it must be valid.
    if(request.getUserinfoSignedResponseAlg()!=null && request.getUserinfoSignedResponseAlg().isPresent()) {
        if(!JWAlgorithmUtils.isValidUserinfoSigningAlg(request.getUserinfoSignedResponseAlg().get())) {
            return Single.error(new InvalidClientMetadataException("Unsupported userinfo signing algorithm"));
        }
    }
    return Single.just(request);
}
 

@Override
public final Single<UnitResponse> execute(Unit daoUnit, Map<String, Object> map, XianConnection connection, String msgId) {
    ///Not needed anymore until/unless we have make all sql execution on xian thread pool managed thread pool
    // set msgId in order make sure compatibility of synchronous and asynchronous dao.
    boolean msgIdWritten = MsgIdHolder.set(msgId);
    try {
        this.msgId = msgId;
        this.map = map;
        this.connection = connection;
        this.daoUnit = (DaoUnit) daoUnit;
        sqlDriver = getSqlDriver();
        if (ignore()) {
            return Single.just(UnitResponse.createSuccess(
                    String.format("This sql action '%s.%s' is ignored for execution.",
                            this.daoUnit.getName(), getClass().getSimpleName())
            ));
        } else {
            UnitResponse response = check();
            if (!response.succeeded()) {
                return Single.just(response);
            }
        }
        if (XianConfig.getBoolean(CONFIG_LOG_DETAILED_SQL, true)) {
            logSql(map);
        }
        final long before = System.nanoTime();
        return executeSql()
                .flatMap(sqlExecutionResult -> Single.just(UnitResponse.createSuccess(sqlExecutionResult)))
                .doOnSuccess(unitResponse -> after(before))
                .onErrorReturn(error -> {
                    LOG.error(error);
                    return getSqlDriver().handleException(error, this);
                })
                ;
    } finally {
        if (msgIdWritten) {
            MsgIdHolder.clear();
        }
    }
}
 

@Override
public Single<Map<Object, Object>> statistics(AnalyticsQuery query) {
    switch (query.getField()) {
        case Field.USER_STATUS:
            return usersStatusRepartition(query);
        case Field.USER_REGISTRATION:
            return registrationsStatusRepartition(query);
    }

    return Single.just(Collections.emptyMap());
}
 

@Test
public void oneToMany() {
    RxGreeterGrpc.RxGreeterStub stub = RxGreeterGrpc.newRxStub(channel);
    Single<String> rxRequest = Single.just("World");
    Flowable<String> rxResponse = rxRequest
            .map(this::toRequest)
            .as(stub::sayHelloRespStream)
            .map(this::fromResponse);

    TestSubscriber<String> test = rxResponse.test();
    test.awaitTerminalEvent(1, TimeUnit.SECONDS);

    test.assertNoErrors();
    test.assertValues("Hello World", "Hi World", "Greetings World");
}
 

@Get("/guilds/{id}")
Single<HttpResponse> guild(@RequestAttribute("session") Session session, long id) {
    if (!session.hasGuild(id)) {
        return Single.just(ErrorResponse.get(ErrorCode.UNKNOWN_GUILD));
    }

    return Flowable.fromFuture(Main.getCatnip().rest().guild()
            .getGuild(String.valueOf(id))
            .thenApply(KGuildExt::new)
            .toCompletableFuture())
            .singleOrError()
            .flatMap(guild -> Single.<MutableHttpResponse<?>>just(HttpResponse.ok(guild))
                    .onErrorReturn(err -> ErrorResponse.get(ErrorCode.UNKNOWN_GUILD)));
}
 

private Single<Event<SaveResult>> onSave() {
  if (originalObservation == null) {
    Timber.e("Save attempted before observation loaded");
    return Single.just(Event.create(SaveResult.NO_CHANGES_TO_SAVE));
  }

  if (hasValidationErrors()) {
    return Single.just(Event.create(SaveResult.HAS_VALIDATION_ERRORS));
  }
  if (!hasUnsavedChanges()) {
    return Single.just(Event.create(SaveResult.NO_CHANGES_TO_SAVE));
  }
  return save();
}
 

public static Single<ChanBoard> fetchBoard(final String boardName) {
    if (TextUtils.isEmpty(boardName)) {
        return Single.just(new ChanBoard());
    }
    return DatabaseUtils.fetchTable(Board.class, Board.TABLE_NAME, null, Board.KEY_NAME + "=?", boardName)
            .flatMap((Function<List<Board>, Single<ChanBoard>>) boards -> {
                if (boards != null && boards.size() > 0) {
                    return Single.just(convertBoardDbModelToBoard(boards.get(0)));
                }
                return Single.just(new ChanBoard());
            });
}
 

public BenchmarkRxServerServiceImpl(int times) {
    Messages.SimpleResponse[] array = new Messages.SimpleResponse[times];
    Arrays.fill(array, Messages.SimpleResponse.getDefaultInstance());

    this.responseFlux = Flowable.fromArray(array);
    this.responseMono = Single.just(Messages.SimpleResponse.getDefaultInstance());
}
 

@Path("indexWithTemplateExtension")
@GET
public Single<Template> indexWithTemplateExtension() {
	return Single.just(new Template("templates/index.ftl")
			.set("title", "my title")
			.set("message", "my message"));
}
 

private Single<String> sign(CertificateProvider certificateProvider, JWT jwt) {
    return Single.just(certificateProvider.getJwtBuilder().sign(jwt));
}
 

@Override
public Single<List<CurrencyTransferEntity>> getLastTransfers() {
    return Single.just(new ArrayList<>());
}
 

@Override
public Single<RxBleDeviceServices> discoverServices() {
    return Single.just(rxBleDeviceServices);
}
 

@Path("reactive")
@GET
public Single<String> helloReactive() {
	return Single.just("Hello Reactive World");
}
 

Single<Post> save(Post post) {
    long id = DATA.size() + 1;
    Post saved = Post.builder().id(id).title(post.getTitle()).content(post.getContent()).build();
    DATA.add(saved);
    return Single.just(saved);
}
 

private Single<DynamicClientRegistrationRequest> validateSectorIdentifierUri(DynamicClientRegistrationRequest request) {
    //if sector_identifier_uri is provided, then retrieve content and validate redirect_uris among this list.
    if(request.getSectorIdentifierUri()!=null && request.getSectorIdentifierUri().isPresent()) {

        URI uri;
        try {
            //throw exception if uri mal formated
            uri = formatUrl(request.getSectorIdentifierUri().get());
        } catch (InvalidClientMetadataException err) {
            return Single.error(new InvalidClientMetadataException("sector_identifier_uri: "+err.getMessage()));
        }

        if(!uri.getScheme().equalsIgnoreCase("https")) {
            return Single.error(new InvalidClientMetadataException("Scheme must be https for sector_identifier_uri : "+request.getSectorIdentifierUri().get()));
        }

        return client.getAbs(uri.toString())
                .rxSend()
                .map(HttpResponse::bodyAsString)
                .map(JsonArray::new)
                .onErrorResumeNext(Single.error(new InvalidClientMetadataException("Unable to parse sector_identifier_uri : "+ uri.toString())))
                .flatMapPublisher(Flowable::fromIterable)
                .cast(String.class)
                .collect(HashSet::new,(set, value)->set.add(value))
                .flatMap(allowedRedirectUris -> Observable.fromIterable(request.getRedirectUris().get())
                        .filter(redirectUri -> !allowedRedirectUris.contains(redirectUri))
                        .collect(ArrayList<String>::new, (list, missingRedirectUri)-> list.add(missingRedirectUri))
                        .flatMap(missing -> {
                            if(!missing.isEmpty()) {
                                return Single.error(
                                        new InvalidRedirectUriException("redirect uris are not allowed according to sector_identifier_uri: "+
                                                String.join(" ",missing)
                                        )
                                );
                            } else {
                                return Single.just(request);
                            }
                        })
                );
    }
    return Single.just(request);
}
 

/**
 * Get a period object specifying a periodId.
 * If the periodId does not exist in the database, it is inserted.
 *
 * @param periodId Period id.
 * @throws IllegalArgumentException if the periodId does not match any period
 *
 * @return {@code Single} with the generated period.
 */
public Single<Period> getPeriodForPeriodId(@NonNull String periodId) throws IllegalArgumentException {
    return Single.just(blockingGetPeriodForPeriodId(periodId));
}