- iOS/Objective-C 元类和类别
- objective-c - -1001 错误,当 NSURLSession 通过 httpproxy 和/etc/hosts
- java - 使用网络类获取 url 地址
- ios - 推送通知中不播放声音
我有一个方法可以将多个文件同时上传到云存储。它看起来像这样:
List<String> files = Arrays.asList("file0", "file1", "file2");
Observable.from(files)
.flatMap(file -> uploadFile(file)
.flatMap(done -> notifyFinished(file)))
.subscribe(this::onNext, this::onError, this::onCompleted);
private Observable<Boolean> uploadFile(String file) {
Timber.d("Uploading: " + file);
return Observable.just(true).delay(6, TimeUnit.SECONDS);
}
private Observable<Boolean> notifyFinished(String file) {
Timber.d("Notify finished: " + file);
return Observable.just(true).delay(3, TimeUnit.SECONDS);
}
这个的输出是:
06-09 02:10:04.779 D: Uploading: file0
06-09 02:10:04.780 D: Uploading: file1
06-09 02:10:04.781 D: Uploading: file2
06-09 02:10:10.782 D: Notify finished: file1
06-09 02:10:10.782 D: Notify finished: file0
06-09 02:10:10.783 D: Notify finished: file2
06-09 02:10:13.784 D: onNext
06-09 02:10:13.786 D: onNext
06-09 02:10:13.786 D: onNext
06-09 02:10:13.787 D: onCompleted
我想让它按顺序工作,例如:
1) Uploading: file0
2) Notify finished: file0
3) onNext
4) Uploading: file1
5) Notify finished: file1
6) onNext
...
是否可以用 Rx 做这样的事情?
编辑
将第一个 flatMap
替换为 concatMap
完成了这项工作。我以为我知道这些运算符之间的区别,但这个例子只是表明我一无所知......现在输出是:
06-09 02:15:00.581 D: Uploading: file0
06-09 02:15:06.584 D: Notify finished: file0
06-09 02:15:09.586 D: onNext
06-09 02:15:09.587 D: Uploading: file1
06-09 02:15:15.590 D: Notify finished: file1
06-09 02:15:18.593 D: onNext
06-09 02:15:18.595 D: Uploading: file2
06-09 02:15:24.598 D: Notify finished: file2
06-09 02:15:27.599 D: onNext
06-09 02:15:27.601 D: onCompleted
最佳答案
如果您想要“有序”连续顺序,只需使用 concatMap()
而不是 flatMap()
关于java - RxJava - 按顺序上传文件 - 在调用 onNext 时发出下一个项目,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/37713988/
我有两个问题: 'bind(onNext:' 和 'subscribe(onNext:' 有什么区别? struct Info { var index: Int? var data:
在一个使用 RxJava 2 的 Android 项目中,我在初始 Activity 的 onCreate 中创建了一个像这样的 Flowable: Flowable.create(new MyFlo
我有一个函数可以从网络服务中获取一些信息: const rxjs = require('rxjs'); const request = require('request'); const moment
所以我有一个Observable。我想知道什么时候 每个 下游订阅者 onNext 调用将要进行,这既是为了测量每个下游订阅者花费的时间量,也是为了背压目的。 Observable.lift 让我完成
我是 Rx 的新手,我真的很感激在错误处理方面的一些帮助。我有以下代码,它基本上是预先输入的: var observable = Rx.Observable.fromEvent(searchField
有没有办法延迟某些值的onNext?,例如: let subject = new Rx.BehaviorSubject({value: 100}) subject.onNext({ value: 12
下面我有向两个 API 端点发出请求的代码。我正在寻找一种让 onNext() 函数仅触发一次的方法。现在,如果您运行下面的代码,您可以看到它 logs fire 1,然后是 fire 2。我只想让它
我有一个发出长值的可观察值。一旦我收到一个>0的长值,那么我只需要中断发出并点击onComplete。有没有比我所做的黑客方法更好的方法来做到这一点 long id = -1l; methodA()
我正在尝试使用 RxJava 创建模型的代理,该代理允许其他人订阅模型中的更改。 这是我的代理的简化版本: class MyModelProxy { private static MyMode
我正在用 Retrofit 调用一个 url,我不关心结果。无论如何要忽略 onNext 方法吗? 目前我正在使用: getRetrofit().create(UserAPI.class)
我正在使用 RxAndroid 库来处理使用订阅者/可观察模式的项目列表。我的问题是,当一个项目被处理时,有一个进度条需要更新。但是在处理了 16 个项目之后,似乎 observable 没有调用 o
我有一个创建新 Observable 的用例: Observable.create(new Observable.OnSubscribe() { @Override public void
PublishSubject 非常适合在观察者和可观察对象之间架起一座桥梁: Observable observable = Observable.just("string"); Pu
有没有更简洁的方法来执行以下操作? 我有一个 Android observable 可以消除请求。 onNext 它调用第二个可观察对象。 {// when creating the android
我想在调用 onNext 后 10 秒发出一个随机值。所以 onNext 将被调用 12345,10 秒后 onNext 应该被调用以获得一个随机数。使用 RxJava 实现此目的的最佳方法是什么?提
每隔一段时间我都会用特定的查询检索推文。这些推文必须传递给计算和操作这些推文的服务。所以这些服务订阅了我的出版商。所以 publisher.hasSubscribers() 返回 true。但是提交或
我正在尝试在我的 http 代码中使用一些 reactiveX。为此,我创建了一个主题,并尝试调用 onNext 函数,但我收到一条错误消息,指出 subject.onNext 不是一个函数。 htm
我有这个简单的 retrofit2 api 接口(interface),其中包含 interface Api { @GET(BuildConfig.END_POINT) Observable
有人可以告诉我这个插件是如何工作的 http://bootstraptour.com/api/ ?我有一个步骤,我想重定向到动态网址。所以我编写了 onNext 回调,如下所示的代码。重定向工作正常,
我在 GRPC 文档中找不到任何与此相关的内容。 GRPC 是否期望我的 StreamObserver.onNext() 实现是非阻塞的?如果 GRPC 确实阻塞(例如拒绝新请求、对新请求排队等),会
我是一名优秀的程序员,十分优秀!