安卓改装并行文件下载问题

Question

我的应用程序中有一个用例，用于从服务器下载两个zip文件。为此，我一直在使用reforming+rxjava(创建了两个独立的改装服务)。对于并行I执行，已经在新线程中订阅了改造服务，然后使用zip操作符将其组合。它运行良好。但后来我在这两个服务中都添加了map操作符进行解压缩操作，但它没有执行map操作符中编写的代码，控件直接传递给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运算符都没有被执行。

还有一个问题

我正在合并/压缩这两个可观测值，并且传递给运算符的类型是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());

Answer 1

您正在有效地做的是：

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

map()(以及所有其他rx运算符)不会使源发生突变。它返回新的可观测值，您可以在进一步的计算中使用。在代码中，您将忽略那些返回的对象。您正在压缩原始的可观测值，而不是映射的可观测量，这就是为什么映射器函数不会被调用的原因。

我想你想做以下事情：

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