java类rx.functions.Func2的实例源码

private Func1<TemperatureModel, Observable<TemperatureModel>> requestTemperature() {

        final Func2<String, Heating.TemperatureScale, Heating> request = Heating::requestTemperature;

        final Func2<OpenSession, TemperatureModel, TemperatureModel> handler = (openSession, temperature) -> {
            Heating.handleTemperature(
                value -> temperature.setValue(String.valueOf(value)),
                () -> temperature.setValue(null))
            .call(openSession);
            return temperature;
        };

        return temperature -> commonService.findClient(temperature.getGatewayUuid())
            .send(request.call(temperature.getWhere(), preferenceService.getDefaultTemperatureScale()))
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .map(openSession -> handler.call(openSession, temperature))
            .onErrorReturn(throwable -> {
                log.warn("temperature={} | failing request={}", temperature.getUuid(), Heating.requestTemperature(temperature.getWhere()));
                // unreadable temperature
                return temperature;
            });

    }

private Func1<SoundModel, Observable<SoundModel>> requestSound(
    Func3<String, SoundSystem.Type, SoundSystem.Source, SoundSystem> request,
    Func2<OpenSession, SoundModel, SoundModel> handler) {

    return sound -> commonService.findClient(sound.getGatewayUuid())
        .send(request.call(sound.getWhere(), sound.getSoundSystemType(), sound.getSoundSystemSource()))
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .map(openSession -> handler.call(openSession, sound))
        .onErrorReturn(throwable -> {
            log.warn("sound={} | failing request={}", sound.getUuid(),
                request.call(sound.getWhere(), sound.getSoundSystemType(), sound.getSoundSystemSource()).getValue());
            // unreadable status
            return sound;
        });
}

public static <T> boolean elementsEqual(List<T> list1, List<T> list2, Func2<T, T, Boolean> comparator) {
    if (list1 == null || list2 == null) {
        return false;
    }

    if (list1.size() != list2.size()) {
        return false;
    }

    for (int i = 0; i < list1.size(); i++) {
        if (!comparator.call(list1.get(i), list2.get(i))) {
            return false;
        }
    }

    return true;
}

@Provides
@PerActivity
Observable<Integer> provideNotificationObservable(GameApi mGameApi,
                                                  ForumApi mForumApi) {
    return Observable.zip(mGameApi.queryPmList(""), mForumApi.getMessageList("", 1),
            new Func2<PmData, MessageData, Integer>() {
                @Override
                public Integer call(PmData pmData, MessageData messageData) {
                    int size = 0;
                    if (pmData != null) {
                        if (pmData.is_login == 0) {
                            return null;
                        }
                        for (Pm pm : pmData.result.data) {
                            if (!TextUtils.isEmpty(pm.unread) && pm.unread.equals("1")) {
                                size++;
                            }
                        }
                    }
                    if (messageData != null && messageData.status == 200) {
                        size += messageData.result.list.size();
                    }
                    return size;
                }
            }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
}

public Observable<List<Program>> syncMetaData() {
    Set<ProgramType> programTypes = new HashSet<>();
    programTypes.add(ProgramType.WITHOUT_REGISTRATION);

    return Observable.zip(userOrganisationUnitInteractor.pull(),
            userProgramInteractor.pull(ProgramFields.DESCENDANTS, programTypes),
            new Func2<List<OrganisationUnit>, List<Program>, List<Program>>() {
                @Override
                public List<Program> call(List<OrganisationUnit> units, List<Program> programs) {
                    if (syncDateWrapper != null) {
                        syncDateWrapper.setLastSyncedNow();
                    }
                    return programs;
                }
            });
}

private void findUsersWhoLovesBoth() {
    // here we are using zip operator to combine both request
    Observable.zip(getCricketFansObservable(), getFootballFansObservable(),
            new Func2<List<User>, List<User>, List<User>>() {
                @Override
                public List<User> call(List<User> cricketFans,
                                       List<User> footballFans) {
                    List<User> userWhoLovesBoth =
                            filterUserWhoLovesBoth(cricketFans, footballFans);
                    return userWhoLovesBoth;
                }
            }
    ).subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<List<User>>() {
                @Override
                public void onCompleted() {
                    // do anything onComplete
                }

                @Override
                public void onError(Throwable e) {
                    // handle error
                }

                @Override
                public void onNext(List<User> users) {
                    // do anything with user who loves both
                    Log.d(TAG, "userList size : " + users.size());
                    for (User user : users) {
                        Log.d(TAG, "id : " + user.id);
                        Log.d(TAG, "firstname : " + user.firstname);
                        Log.d(TAG, "lastname : " + user.lastname);
                    }
                }
            });
}

private void findUsersWhoLovesBoth() {
    // here we are using zip operator to combine both request
    Observable.zip(getCricketFansObservable(), getFootballFansObservable(),
            new Func2<List<User>, List<User>, List<User>>() {
                @Override
                public List<User> call(List<User> cricketFans,
                                       List<User> footballFans) {
                    List<User> userWhoLovesBoth =
                            filterUserWhoLovesBoth(cricketFans, footballFans);
                    return userWhoLovesBoth;
                }
            }
    ).subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<List<User>>() {
                @Override
                public void onCompleted() {
                    // do anything onComplete
                }

                @Override
                public void onError(Throwable e) {
                    // handle error
                }

                @Override
                public void onNext(List<User> users) {
                    // do anything with user who loves both
                    Log.d(TAG, "userList size : " + users.size());
                    for (User user : users) {
                        Log.d(TAG, "id : " + user.id);
                        Log.d(TAG, "firstname : " + user.firstname);
                        Log.d(TAG, "lastname : " + user.lastname);
                    }
                }
            });
}

public void runCode() {

        Observable obs1 = Observable.interval(1, TimeUnit.SECONDS);
        Observable obs2 = Observable.range(0,3);
        Plan0 plan0 = JoinObservable.from(obs1).and(obs2).then(new Func2() {
            @Override
            public Object call(Object o1, Object o2) {
                return o2;
            }
        });
        JoinObservable.when(plan0).toObservable()
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onCompleted() {
                        println("onCompleted");
                    }

                    @Override
                    public void onError(Throwable e) {
                        println("onError");
                    }

                    @Override
                    public void onNext(Integer integer) {
                        println("onNext:"+integer);
                    }

                });


    }

public void runCode() {

        Observable.just(1,2,3,4).reduce(new Func2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer i1, Integer i2) {
                return i1*i2;
            }
        }) .subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                println("onCompleted");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Integer integer) {
                println("onNext:"+integer);
            }
        });



    }

@Override
public void start() {
    Utils.debugLog(TAG, "Preparing swap webserver.");
    sendBroadcast(SwapService.EXTRA_STARTING);

    if (FDroidApp.ipAddressString == null) {
        Log.e(TAG, "Not starting swap webserver, because we don't seem to be connected to a network.");
        setConnected(false);
    }

    Single.zip(
            Single.create(getWebServerTask()),
            Single.create(getBonjourTask()),
            new Func2<Boolean, Boolean, Boolean>() {
                @Override
                public Boolean call(Boolean webServerTask, Boolean bonjourServiceTask) {
                    return bonjourServiceTask && webServerTask;
                }
            })
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.newThread())
            .subscribe(new Action1<Boolean>() {
                    @Override
                    public void call(Boolean success) {
                        setConnected(success);
                    }
                },
                new Action1<Throwable>() {
                    @Override
                    public void call(Throwable throwable) {
                        setConnected(false);
                    }
                });
}