代码之家 › 专栏 › 技术社区 › Hunt

从WebService存储数据获取后返回RXJava中的订阅服务器

rx-java2 realm android

1

Hunt · 技术社区 · 8 年前

现在,问题是当我成功地获取数据并将其存储在数据库中时,我想通知我的活动。如果发生了一些错误,那么我想在UI上显示它。

public Flowable<Response> getFitnessData() {

        Request request = new Request();
        request.setAccess_token("d80fa6bd6f78cc704104d61146c599bc94b82ca225349ee68762fc6c70d2dcf0");

        Flowable<Response> fitnessFlowable = new WebRequest()
                                            .getRemoteClient()
                                            .create(FitnessApi.class)
                                            .getFitnessData("5b238abb4d3590001d9b94a8",request.toMap());


         fitnessFlowable.subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .takeUntil(response->response.getSummary().getNext()!=null)

                .subscribe(new Subscriber<Response>() {
                    @Override
                    public void onSubscribe(Subscription s) {

                        s.request(Long.MAX_VALUE);
                    }

                    @Override
                    public void onNext(Response response) {

                        Log.e(TAG, "onNext" );

                        if(response !=null){

                            if(response.getFitness()!=null && response.getFitness().size()!=0){

                                Realm realm = Realm.getDefaultInstance();
                                realm.executeTransactionAsync(new Realm.Transaction() {
                                    @Override
                                    public void execute(Realm realm) {

                                        realm.copyToRealmOrUpdate(response.getFitness());

                                    }
                                }, new Realm.Transaction.OnSuccess() {
                                    @Override
                                    public void onSuccess() {

                                        Log.i(TAG, "onSuccess , fitness data saved");

                                    }
                                }, new Realm.Transaction.OnError() {
                                    @Override
                                    public void onError(Throwable error) {
                                        Log.i(TAG, "onError , fitness data failed to save"+error.getMessage() );
                                    }
                                });
                            }else{

                                Log.i(TAG, "onError , no fitness data available" );


                            }

                        }else{
                            Log.i(TAG, "onError , response is null" );

                        }
                    }

                    @Override
                    public void onError(Throwable t) {


                        Log.e(TAG, "onError" +t.getMessage());
                    }

                    @Override
                    public void onComplete() {

                        Log.e(TAG, "onComplete");
                    }
                });;

            return null;

    }

更新

RxBus 在UI上传播事件和错误

public class RxBus {

    private static final RxBus INSTANCE = new RxBus();

    private RxBus(){}
    private PublishSubject<Object> bus = PublishSubject.create();

    public static RxBus getInstance() {
        return INSTANCE;
    }


    public void send(Object o) {
        bus.onNext(o);
    }

    public void error(Throwable e){bus.onError(e);}

    public Observable<Object> toObservable() {
        return bus;
    }
}

 FitnessRepo fitnessRepo = new FitnessRepo();
        fitnessRepo.getFitnessData();
        RxBus.getInstance().toObservable().subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Object o) {

                if(o instanceof RealmList ){

                    RealmList<Fitness> realmList = (RealmList<Fitness>) o;
                    Log.e(TAG,"Fitness data size "+realmList.size());

                }
            }

            @Override
            public void onError(Throwable e) {

                Log.e(TAG,e.getMessage()+ "");

                if (e instanceof HttpException) {
                    ResponseBody body = ((HttpException) e).response().errorBody();


                    try {

                        Response response=  LoganSquare.parse(body.byteStream(),Response.class);

                        if(response.getErrors() !=null)
                            if(response.getErrors().size() > 0)
                                Log.e(TAG, "Error "+response.getErrors().get(0).getErrors());
                    } catch (IOException t) {
                        t.printStackTrace();
                    }

                }
            }

            @Override
            public void onComplete() {

            }
        });

public void getFitnessData() {


        Request request = new Request();
        request.setAccess_token("d80fa6bd6f78cc704104d61146c599bc94b82ca225349ee68762fc6c70d2dcf0");
        request.setEnd_date("2018-07-01T00:00:00");
        Flowable<Response> fitnessFlowable = new WebRequest()
                .getRemoteClient()
                .create(FitnessApi.class)
                .getFitnessData("5b238abb4d3590001d9b94a8",request.toMap());


         fitnessFlowable.subscribeOn(Schedulers.io())
                .takeUntil(response->response.getSummary().getNext()!=null)
                .doOnNext((response) -> {
                    if(response ==null || response.getFitness() == null || response.getFitness().isEmpty()) {


                        Log.e(TAG, " Error ");
                        return;
                    }

                    RxBus.getInstance().send(response.getFitness());

                    try(Realm r = Realm.getDefaultInstance()) {
                        r.executeTransaction((realm) -> {
                            realm.copyToRealmOrUpdate(response.getFitness());
                        });
                    }
                }).subscribe(item ->{


                 },
                 error ->{

                     RxBus.getInstance().error(error);


                 });
    }

2 回复 | 直到 8 年前

1

EpicPandaForce Jigar Joshi 8 年前

我有好消息要告诉你!您可以删除几乎所有的代码,并因此使其总体上更好!

public void fetchFitnessData() {

    Request request = new Request();
    request.setAccess_token("d80fa6bd6f78cc704104d61146c599bc94b82ca225349ee68762fc6c70d2dcf0");

    Flowable<Response> fitnessFlowable = new WebRequest()
                                        .getRemoteClient()
                                        .create(FitnessApi.class)
                                        .getFitnessData("5b238abb4d3590001d9b94a8",request.toMap());


     fitnessFlowable.subscribeOn(Schedulers.io())
            .takeUntil(response->response.getSummary().getNext()!=null)
            .doOnNext((response) -> {
                    if(response ==null || response.getFitness() == null || response.getFitness().isEmpty()) return;

                    try(Realm r = Realm.getDefaultInstance()) {
                        r.executeTransaction((realm) -> {
                            realm.insertOrUpdate(response.getFitness());
                        });
                    }
                }
            }).subscribe();
}

void 因此,从该方法中发出内容的方法是使用 PublishSubject EventBus

private PublishSubject<Object> fitnessResults;
public Observable<Object> observeFitnessResults() {
    return fitnessResults;
}

public static class Success {
    public Success(List<Fitness> data) {
        this.data = data;
    }

    public List<Fitness> data;
}

public static class Failure {
    public Failure(Exception exception) {
        this.exception = exception;
    }

    public Exception exception;
}

public void fetchFitnessData() {
    ...
        fitnessResults.onNext(new Success(data));
    } catch(Exception e) {
        fitnessResults.onNext(new Failure(e));

errors = observeFitnessResults().ofType(Error.class);
success = observeFitnessResults().ofType(Success.class);

2

0

Yogesh Madaan 8 年前

实现这一点有不同的方法。我永远不会在生命周期范围之外独自处理订阅,因为它会造成内存泄漏的可能性。在您的例子中,成功和失败似乎都与UI绑定在一起,所以您可以简单地做到这一点。

public Completable fetchFitnessData() {

Request request = new Request();
request.setAccess_token("d80fa6bd6f78cc704104d61146c599bc94b82ca225349ee68762fc6c70d2dcf0");

Flowable<Response> fitnessFlowable = new WebRequest()
                                    .getRemoteClient()
                                    .create(FitnessApi.class)
                                    .getFitnessData("5b238abb4d3590001d9b94a8",request.toMap());


 return fitnessFlowable.subscribeOn(Schedulers.io())
        .takeUntil(response->response.getSummary().getNext()!=null)
        .doOnNext((response) -> {
                if(response ==null || response.getFitness() == null || response.getFitness().isEmpty()) return;

                try(Realm r = Realm.getDefaultInstance()) {
                    r.executeTransaction((realm) -> {
                        realm.insertOrUpdate(response.getFitness());
                    });
                }
            }
        }).ignoreElements();

在UI级别,您只需处理成功和失败的订阅。如果您需要成功,模型可以用单个或可流动的来替换completeable。

fetchFitnessData.subscrible(Functions.EMPTY_ACTION, Timber::d);