io.reactivex.Observable#combineLatest ( )源码实例Demo

private Observable<TransactionFactory<? extends Transaction>> resolveTransactionFactory(
    SecretLockTransaction transaction, ReceiptSource expectedReceiptSource) {
    Observable<Statement> statementObservable = getStatement(transaction);
    Observable<Address> resolvedAddress = getResolvedAddress(transaction,
        transaction.getRecipient(), statementObservable, expectedReceiptSource);
    Observable<Mosaic> resolvedMosaic = getResolvedMosaic(transaction, transaction.getMosaic(),
        statementObservable, expectedReceiptSource
    );
    return Observable.combineLatest(resolvedAddress, resolvedMosaic,
        (address, mosaic) -> SecretLockTransactionFactory
            .create(transaction.getNetworkType(),
                mosaic,
                transaction.getDuration(),
                transaction.getHashAlgorithm(),
                transaction.getSecret(),
                address));
}
 

private Observable<TransactionFactory<? extends Transaction>> resolveTransactionFactory(
    TransferTransaction transaction, ReceiptSource expectedReceiptSource) {
    Observable<Statement> statementObservable = getStatement(transaction);
    Observable<List<Mosaic>> resolvedMosaics = Observable
        .fromIterable(transaction.getMosaics()).flatMap(
            m -> getResolvedMosaic(transaction, m, statementObservable, expectedReceiptSource))
        .toList().toObservable();

    Observable<Address> resolvedRecipient = getResolvedAddress(transaction,
        transaction.getRecipient(),
        statementObservable, expectedReceiptSource);

    BiFunction<Address, List<Mosaic>, TransferTransactionFactory> mergeFunction = (address, mosaics) ->
        TransferTransactionFactory
            .create(transaction.getNetworkType(), address, mosaics, transaction.getMessage());
    return Observable.combineLatest(resolvedRecipient, resolvedMosaics, mergeFunction);
}
 

private Observable<TransactionFactory<? extends Transaction>> resolveTransactionFactory(
    MosaicGlobalRestrictionTransaction transaction, ReceiptSource expectedReceiptSource) {
    Observable<Statement> statementObservable = getStatement(transaction);
    Observable<MosaicId> resolvedMosaicId = getResolvedMosaicId(transaction,
        transaction.getMosaicId(),
        statementObservable, expectedReceiptSource);

    Observable<MosaicId> resolvedReferenceMosaicId = getResolvedMosaicId(transaction,
        transaction.getReferenceMosaicId(),
        statementObservable, expectedReceiptSource);

    return Observable.combineLatest(resolvedMosaicId, resolvedReferenceMosaicId,
        (mosaicId, referenceMosaicId) ->
        {
            MosaicGlobalRestrictionTransactionFactory factory = MosaicGlobalRestrictionTransactionFactory
                .create(transaction.getNetworkType(), mosaicId,
                    transaction.getRestrictionKey(), transaction.getNewRestrictionValue(),
                    transaction.getNewRestrictionType());
            if (referenceMosaicId != null) {
                factory.referenceMosaicId(referenceMosaicId);
            }
            return factory.previousRestrictionValue(transaction.getPreviousRestrictionValue())
                .previousRestrictionType(transaction.getPreviousRestrictionType());
        });
}
 

private Observable<TransactionFactory<? extends Transaction>> resolveTransactionFactory(
    MosaicAddressRestrictionTransaction transaction,
    ReceiptSource expectedReceiptSource) {
    Observable<Statement> statementObservable = getStatement(transaction);
    Observable<MosaicId> resolvedMosaicId = getResolvedMosaicId(transaction,
        transaction.getMosaicId(),
        statementObservable, expectedReceiptSource);

    Observable<Address> resolvedTargetAddress = Observable
        .just(transaction.getTargetAddress())
        .flatMap(m -> getResolvedAddress(transaction, m, statementObservable,
            expectedReceiptSource));

    BiFunction<? super MosaicId, ? super Address, MosaicAddressRestrictionTransactionFactory> mapper = (mosaicId, targetAddress) ->
        MosaicAddressRestrictionTransactionFactory
            .create(transaction.getNetworkType(), mosaicId,
                transaction.getRestrictionKey(), targetAddress,
                transaction.getNewRestrictionValue())
            .previousRestrictionValue(transaction.getPreviousRestrictionValue());
    return Observable.combineLatest(resolvedMosaicId, resolvedTargetAddress, mapper);
}
 

private Observable<TransactionFactory<? extends Transaction>> resolveTransactionFactory(
    AccountMosaicRestrictionTransaction transaction,
    ReceiptSource expectedReceiptSource) {
    Observable<Statement> statementObservable = getStatement(transaction);
    Observable<List<UnresolvedMosaicId>> unresolvedAdditions = getResolvedMosaicIds(transaction,
        transaction.getRestrictionAdditions(), statementObservable, expectedReceiptSource);

    Observable<List<UnresolvedMosaicId>> unresolvedDeletions = getResolvedMosaicIds(transaction,
        transaction.getRestrictionDeletions(), statementObservable, expectedReceiptSource);

    BiFunction<List<UnresolvedMosaicId>, List<UnresolvedMosaicId>, TransactionFactory<AccountMosaicRestrictionTransaction>> mapper =
        (additions, deletions) -> AccountMosaicRestrictionTransactionFactory
            .create(transaction.getNetworkType(), transaction.getRestrictionFlags(), additions,
                deletions);
    return Observable.combineLatest(unresolvedAdditions, unresolvedDeletions, mapper);
}
 

private Observable<TransactionFactory<? extends Transaction>> resolveTransactionFactory(
    AccountAddressRestrictionTransaction transaction,
    ReceiptSource expectedReceiptSource) {
    Observable<Statement> statementObservable = getStatement(transaction);
    Observable<List<UnresolvedAddress>> unresolvedAdditions = getResolvedAddresses(transaction,
        transaction.getRestrictionAdditions(), statementObservable, expectedReceiptSource);

    Observable<List<UnresolvedAddress>> unresolvedDeletions = getResolvedAddresses(transaction,
        transaction.getRestrictionDeletions(), statementObservable, expectedReceiptSource);

    BiFunction<List<UnresolvedAddress>, List<UnresolvedAddress>, AccountAddressRestrictionTransactionFactory> mapper =
        (additions, deletions) -> AccountAddressRestrictionTransactionFactory
            .create(transaction.getNetworkType(), transaction.getRestrictionFlags(), additions,
                deletions);
    return Observable.combineLatest(unresolvedAdditions, unresolvedDeletions, mapper);
}
 

public RxCommand<String> captchaCommand() {
    if (captchaCommand == null) {
        Observable<Boolean> enabled = Observable.combineLatest(
                phoneNumberValid,
                countdownCommand().executing(),
                (valid, executing) -> valid && !executing);

        captchaCommand = RxCommand.create(enabled, o -> {
            String phone = phoneNumber.value().toString();
            Timber.i("fetch captcha with %s", phone);
            Observable<String> fetchCode =  fetchCaptcha(phone);
            Observable<String> countdown =  Observable.defer(() -> countdownCommand().execute(null).ignoreElements().toObservable()) ;
            return Observable.concat(fetchCode, countdown);
        });
    }
    return captchaCommand;
}
 

private Observable<ProductDetail> getProductWithShoppingCartInfo(int productId) {
  List<Observable<?>> observables =
      Arrays.asList(backendApi.getProduct(productId), shoppingCart.itemsInShoppingCart());

  return Observable.combineLatest(observables, objects -> {
    Product product = (Product) objects[0];
    List<Product> productsInShoppingCart = (List<Product>) objects[1];
    boolean inShoppingCart = false;
    for (Product p : productsInShoppingCart) {
      if (p.getId() == productId) {
        inShoppingCart = true;
        break;
      }
    }

    return new ProductDetail(product, inShoppingCart);
  });
}
 

public RxCommand<Boolean> loginCommand() {
    if (loginCommand == null) {
        Observable<Boolean> loginInputValid = Observable.combineLatest(
                captchaValid,
                phoneNumberValid,
                (captchaValid, phoneValid) -> captchaValid && phoneValid);

        loginCommand = RxCommand.create(loginInputValid, o -> {
            String phone = this.phoneNumber.value().toString();
            String captcha = this.captcha.value().toString();
            return login(phone, captcha);
        });
    }
    return loginCommand;
}
 

public Observable<Integer> getErrorResourceId() {
	return Observable.combineLatest(
		playerEventHandler.getErrorResourceId(),
		playerEventHandler.isIdle(),
		(errorResourceId, isIdle) -> isIdle ? errorResourceId : -1);
}
 

@Override
public <U, R> ReactiveSeq<R> zipLatest(Publisher<? extends U> other, BiFunction<? super T, ? super U, ? extends R> zipper) {
    Observable<R> obs = Observable.combineLatest(observable, Observables.observable(other), (a, b) -> zipper.apply((T)a, (U)b));
    return observable(obs);
}
 

@Override
public EditTriggersPresenter initialize() {
    ObjectUtils.requireNotNull(selection, "selection");
    super.initialize();
    getView().displayStatus(selection);

    ProviderFacade.QueryBuilder<Trigger> triggersQuery = providerFacade.query(Trigger.class);
    ProviderFacade.QueryBuilder<Cluster> clusterQuery = providerFacade.query(Cluster.class);
    ProviderFacade.QueryBuilder<Vm> vmQuery = providerFacade.query(Vm.class); // just vms for now

    if (selection.isOneAccount()) {
        triggersQuery.where(Trigger.ACCOUNT_ID, selection.getAccountId());
        clusterQuery.where(Cluster.ACCOUNT_ID, selection.getAccountId());
        vmQuery.where(Vm.ACCOUNT_ID, selection.getAccountId());
    }

    if (selection.isCluster()) {
        triggersQuery.where(Trigger.CLUSTER_ID, selection.getClusterId());
        clusterQuery.where(Cluster.ID, selection.getClusterId());
        vmQuery.where(Vm.CLUSTER_ID, selection.getClusterId());
    }

    if (entityId != null) {
        triggersQuery.where(Trigger.TARGET_ID, entityId);
        vmQuery.where(Vm.ID, entityId);
    }

    final Observable<List<Trigger>> triggersObservable = triggersQuery.asObservable();
    final Observable<List<Cluster>> clusterObservable = clusterQuery.asObservable();
    final Observable<List<Vm>> vmObservable = vmQuery.asObservable();

    getDisposables().add(Observable.combineLatest(rxStore.ALL_ACCOUNTS.startWith(AllAccounts.NO_ACCOUNTS), clusterObservable,
            triggersObservable, vmObservable, Wrapper::new)
            .subscribeOn(Schedulers.computation())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(wrapper -> getView().showTriggers(wrapper.asViewTriggers(selection, entityId))));

    final Observable<TriggerAndVm> titleInfo;
    if (entityId != null) {
        titleInfo = Observable.combineLatest(
                vmObservable.filter(vms -> !vms.isEmpty()).map(vms -> vms.iterator().next()),
                selectedTrigger,
                TriggerAndVm::new);
    } else {
        titleInfo = selectedTrigger.map(selected -> new TriggerAndVm(null, selected));
    }

    getDisposables().add(titleInfo
            .subscribeOn(Schedulers.computation())
            .distinctUntilChanged()
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(wrapper -> {
                if (wrapper.selectedTrigger.isSelected()) {
                    getView().displayTitle(wrapper.selectedTrigger.trigger.toString());
                } else {
                    String show = resources.getTriggers();
                    if (wrapper.entity != null) {
                        show += " " + ((OVirtAccountNamedEntity.class.isAssignableFrom(wrapper.entity.getClass())) ?
                                wrapper.entity.getName() : entityId);
                    }
                    getView().displayTitle(show);
                }
            }));

    return this;
}