android - RxJava、OkHttp、Okio、Android 中的文件下载器

Question

我正在尝试将此工作文件下载代码转换为 Reactive。但是由于我对 RxJava 知之甚少，卡住了。你能帮我把它变成 Reactive 吗？

public void downloadFile(MessageComponent media) {
        Request request = new Request.Builder()
                .url(media.getMediaUrl())
                .build();

        Call call = http_client.newCall(request);
        call.enqueue(new Callback() {
            @Override
            public void onFailure(Request request, IOException e) {
                Log.e(TAG, "Failed to execute " + request, e);
            }

            @Override
            public void onResponse(Response response) throws IOException {
                if (!response.isSuccessful()) {
                    throw new IOException("Unexpected code " + response);
                }
                String mimeType = MimeTypeMap.getFileExtensionFromUrl(media.getMediaUrl());
                File file = new File(helper.getTmpFolder() + "/" + helper.generateUniqueName() + "test." + mimeType);
                BufferedSink sink = Okio.buffer(Okio.sink(file));
                sink.writeAll(response.body().source());
                sink.close();
                Log.d(TAG, "downloadFileFromServer done: " + media.getMediaUrl());
            }
        });
    } 

这是我到目前为止所写的，它没有得到任何结果或错误:

public void downloadFile(MessageComponent media){
   Observable<String> downloadObservable = Observable.create(
            sub -> {
                Request request = new Request.Builder()
                        .url(media.getMediaUrl())
                        .build();
                Response response = null;
                try {
                    response = http_client.newCall(request).execute();
                    if (!response.isSuccessful()) new IOException();
                } catch (IOException e) {
                    e.printStackTrace();
                }

                sub.onNext(response.toString());
            }
    );

    Subscriber<String> mySubscriber = new Subscriber<String>() {
        @Override
        public void onNext(String responseString) {
            Log.d(TAG, "works: " + responseString);
        }

        @Override
        public void onCompleted() {
        }

        @Override
        public void onError(Throwable e) {
            Log.e(TAG, e.getMessage(), e);
        }
    };
    downloadObservable
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(mySubscriber);
    mySubscriber.unsubscribe();

}

Answer 1

您的代码有一些错误，这可能解释了预期的行为不是您得到的。

可观察到的契约(Contract)错误

Reactive Extensions(RxJava 是它的一个实现)基于这个契约:你可以在 onNext 上收到多次通知，然后，你将在错误或完成时收到一次(或永远不会......)通知。

onNext* (onComplete | onError)?

因此，您的 Observable 代码可以重写为这样，以发出您的流出错或已完成的事实。

Observable<String> downloadObservable = Observable.create(
        sub -> {
            Request request = new Request.Builder()
                    .url(media.getMediaUrl())
                    .build();
            Response response = null;
                response = http_client.newCall(request).execute();
                if (response.isSuccessful()) {
                    sub.onNext(response.toString());
                    sub.onCompleted();
                } else {
                    sub.onError(new IOException());
                }
        }
);

提前退订

您在订阅后立即取消订阅，因此您的 Observable 可能没有时间执行。

downloadObservable
        .subscribeOn(Schedulers.newThread())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(mySubscriber);
mySubscriber.unsubscribe();

如果 Observable 完成，它将取消订阅。所以在这种情况下不必退订。