0

我的应用程序中有一个用例，用于从服务器下载两个 zip 文件。为此，我一直在使用改造+rxjava（创建了两个单独的改造服务）。对于并行执行，我一直在新线程中订阅改造服务，然后使用 zip 运算符将其组合起来。它工作正常。但后来我将地图运算符添加到这两个服务以进行解压缩操作，但它没有执行在地图运算符中编写的代码，并且控件直接传递给 zip 操作。我不知道如何解决这个问题，我是反应世界的新手。

到目前为止我已经尝试过什么

    Observable<Response<ResponseBody>> dFileObservable = dbDownloadApi.downloadDealerData(WebServiceConstants.ACTION_DEALER_DATA,
            params.getDealerNumber(),params.getUserId(),params.getClientId(), params.getSessionId()).subscribeOn(Schedulers.newThread());
    dFileObservable.map(new Function<Response<ResponseBody>, String>() {
        @Override
        public String apply(Response<ResponseBody> responseBody) throws Exception {
            String header = responseBody.headers().get("Content-Disposition");
            String filename = header.replace("attachment; filename=", "");
            String downloadFolderPath = fileManager.makeAndGetDownloadFolderPath();
            String dealerZipPath = fileManager.makeFolder(downloadFolderPath, StrConstants.DEALER_FOLDER_NAME);
            fileManager.writeDownloadedFileToDisk(dealerZipPath,filename, responseBody.body().source());
            String dealerFilePath = dealerZipPath+File.separator+filename;
            unzipUtility.unzip(dealerFilePath, fileManager.makeAndGetDownloadFolderPath()+File.separator+ StrConstants.GENERAL_FOLDER_NAME);
            return dealerFilePath;
        }
    });

   Observable<Response<ResponseBody>> generalFileObservable = dbDownloadApi.downloadGeneralData(WebServiceConstants.ACTION_GENERAL_DATA,
            params.getDealerNumber(),params.getUserId(),params.getClientId(), params.getSessionId()).subscribeOn(Schedulers.newThread());;
    generalFileObservable.map(new Function<Response<ResponseBody>, String>() {
        @Override
        public String apply(Response<ResponseBody> responseBody) throws Exception {
            String header = responseBody.headers().get("Content-Disposition");
            String filename = header.replace("attachment; filename=", "");
            String downloadFolderPath = fileManager.makeAndGetDownloadFolderPath();
            String generalZipPath = fileManager.makeFolder(downloadFolderPath, StrConstants.GENERAL_FOLDER_NAME);
            fileManager.writeDownloadedFileToDisk(generalZipPath,filename, responseBody.body().source());
            String generalFilePath = generalZipPath+File.separator+filename;
            unzipUtility.unzip(generalFilePath, fileManager.makeAndGetDownloadFolderPath()+File.separator+ StrConstants.GENERAL_FOLDER_NAME);
            return generalFilePath;
        }
    });

   Observable<String> zipped = Observable.zip(dealerFileObservable, generalFileObservable, new BiFunction<Response<ResponseBody>, Response<ResponseBody>, String>() {
        @Override
        public String apply(Response<ResponseBody> responseBodyResponse, Response<ResponseBody> responseBodyResponse2) throws Exception {
            System.out.println("zipped yess");
            return null;
        }
    }).observeOn(Schedulers.io());

    zipped.subscribe(getObserver());

和 getObserver() 函数

    private Observer<String> getObserver(){

    return new Observer<String>() {
        @Override
        public void onSubscribe(Disposable d) {

        }

        @Override
        public void onNext(String value) {

            System.out.println("------------total time-----------");
            System.out.println("result value-->"+value);
        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onComplete() {

        }
    };
}

当代码执行时，控件被转移到 zip 运算符中的 apply() 函数，并且两个 observable 中的 map 运算符都没有被执行。

还有一个问题

我正在合并/压缩两个 observables，传递给操作员的类型是 Response<"ResponseBody">。实际上我需要那里下载的文件路径（字符串类型），为此我该怎么办？

**

更新了@Yaroslav Stavnichiy 描述的解决方案，现在它可以工作了

**

    Observable<String> deObservable =  dbDownloadApi.downloaddData(WebServiceConstants.ACTION_DATA,
            params.getNumber(),params.getId(),params.getCtId(), params.getSessionId())
            .flatMap(new Function<Response<ResponseBody>, ObservableSource<String>>() {
                @Override
                public ObservableSource<String> apply(Response<ResponseBody> responseBody) throws Exception {
                    String zipPath = fileManager.processDownloadedFile(StrConstants.FOLDER_NAME,
                            StrConstants.FILE_NAME,responseBody.body().source());
                    return Observable.just(zipPath);
                }
            }).map(new Function<String, String>() {
                @Override
                public String apply(String filePath) throws Exception {
                    String unzipDestinationPath = fileManager.makeAndGetDownloadFolderPath()+
                            File.separator+ StrConstants.FOLDER_NAME;
                    unzipUtility.unzip(filePath, unzipDestinationPath);
                    return unzipDestinationPath;
                }
            }).subscribeOn(Schedulers.newThread());
4

1 回答 1

2

您实际上正在做的是：

Observable a = ...;
Observable b = ...;
a.map(...);
b.map(...);
Observable.zip(a, b).subscribe(f);

map()（以及所有其他 rx-operators）不会改变 source。它返回新的 observable，您可以在进一步的计算中使用它。在您的代码中，您忽略了那些返回的对象。您正在压缩原始可观察对象，而不是映射的，这就是为什么不调用映射器函数的原因。

我认为您想要执行以下操作：

Observable a = ... .map(...);
Observable b = ... .map(...);
Observable.zip(a, b).subscribe(f);
于 2017-01-11T11:03:41.423 回答