将异步调用转换为RxJava，如何在处理完所有项目后手动完成？

Question

我有一个自定义类，用于将单个文件或多个文件上传到Firebase存储：

public class FirebaseUploader {
    private Context appContext;
    private FirebaseStorage storage;

    public FirebaseUploader(Context appContext) {
        this.appContext = appContext;
        this.storage = FirebaseStorage.getInstance();
    }

    // Uploading a single file
    public Observable<String> send(Uri file) {
        return toObservable(Observable.just(file));
    }

    // Uploading multiple files
    public Observable<String> send(List<Uri> files) {
        return toObservable(Observable.fromIterable(files));
    }

    private Observable<String> toObservable(Observable<Uri> observable) {
        return observable.flatMap(this::upload);
    }

    private Observable<String> upload(Uri file) {
        return Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                StorageReference branch = storage.getReference().child("images");
                StorageReference leaf = branch.child(generateUniqueId(file));
                UploadTask uploadTask = leaf.putFile(file);

                uploadTask.continueWithTask(new Continuation<UploadTask.TaskSnapshot, Task<Uri>>() {
                    @Override
                    public Task<Uri> then(@NonNull Task<UploadTask.TaskSnapshot> task) throws Exception {
                        if (!task.isSuccessful()) emitter.onError(task.getException());
                        return leaf.getDownloadUrl();
                    }
                }).addOnCompleteListener(new OnCompleteListener<Uri>() {
                    @Override
                    public void onComplete(@NonNull Task<Uri> task) {
                        if (task.isSuccessful()) {
                            Uri downloadUri = task.getResult();
                            String url = downloadUri.toString();
                            emitter.onNext(url);
                            //emitter.onComplete();
                        } else {
                            // Handle failures
                            // ...
                        }
                    }
                });
            }
        });
    }

    private String generateUniqueId(Uri file) {
        String fileType = appContext.getContentResolver().getType(file);
        String extension = (fileType.equals("image/jpeg")) ? ".jpg" : ".png";
        String fileName = UUID.randomUUID().toString();
        return fileName + extension;
    }
}

实际上，我花了几天时间才想出这个解决方案，因为我最初不知道如何将Firebase异步调用转换为RxJava。然后我被困了很长一段时间，想知道为什么可观察的没有完成，直到我最终意识到我在调用emitter.onNext和emitter.onError，但是没有<--plhd-2/>完成在我的代码中的任何地方！

所以发射器。onComplete位于OnCompleteListener内。但它被注释掉了——这不是该打电话的地方。我一直在寻找一种调用的方法。只有在处理完所有文件后，才能在发射器上执行onComplete（）。由于我对RxJava的经验有限，我看到的唯一方法是首先计算要上传的文件数量，然后在达到该数量时手动完成序列。难道没有更好的办法吗？




Answer 1

从upload方法返回的可观察对象只负责一件事：上传Uri参数指定的单个文件。该任务在UploadTask完成时完成，因此此时应调用onComplete。

flatMap操作员负责将所有这些结果收集到一个可观察的中。上传所有文件后，它将向下游发出onComplete，这正是您想要的。