我试图找到一种方法来并行执行请求并在每个可观察对象完成时处理它们。尽管当所有 observables 都给出响应时一切都在工作，但我没有看到在一切都完成后处理所有错误的方法。
这是 zip 运算符的示例，它基本上并行执行 2 个请求：
Observable.zip(
getObservable1()
.onErrorResumeNext { errorThrowable: Throwable ->
Observable.error(ErrorEntity(Type.ONE, errorThrowable))
}.subscribeOn(Schedulers.io()),
getObservable2()
.onErrorResumeNext { errorThrowable: Throwable ->
Observable.error(ErrorEntity(Type.TWO, errorThrowable))
}.subscribeOn(Schedulers.io()),
BiFunction { value1: String, value2: String ->
return@BiFunction value1 + value2
})
//execute requests should be on io() thread
.subscribeOn(Schedulers.io())
//there are other tasks inside subscriber that need io() thread
.observeOn(AndroidSchedulers.mainThread())
.subscribe(
{ result ->
Snackbar.make(view, "Replace with your own action " + result, Snackbar.LENGTH_LONG)
.setAction("Action", null).show()
},
{ error ->
Log.d("TAG", "Error …我目前正在审查 Android 应用程序的更改，这些更改将应用程序升级为使用 Java 8。
在审查期间，我注意到以下代码被加下划线作为错误：
List<MappableTask> tasks = Observable.fromIterable(taskMapHoldingPresenter.getMappableTasks())
.filter(isMappableTask)
.toList()
.blockingGet();
现在显示以下内容：
Incompatible Types.
Required: List<MappableTask>
Found: List<capture<? extends MappableTask>>
该getMappableTasks()方法在接口中定义如下：
List<? extends MappableTask> getMappableTasks();
请注意，这MappableTask是由各种模型类实现的接口。
怎么是List<MappableTask> list = taskMapHoldingPresenter.getMappableTasks();无效的？
此代码按预期构建。如上所述，在升级应用程序以使用 Java 8 之前未显示此错误。为什么会出现此问题？
我正在尝试将图像上传到 Firebase 存储，我想通过 RxJava2 实现这一目标。这是我的代码..
public override fun onActivityResult(requestCode: Int, resultCode: Int, data: Intent) {
if (requestCode == CropImage.CROP_IMAGE_ACTIVITY_REQUEST_CODE) {
val result = CropImage.getActivityResult(data)
if (resultCode == Activity.RESULT_OK) {
val resultUri = result.uri
var actualImageFile = File(resultUri.path)
dialogs = SpotsDialog(this, "upload")
imageCompressor = Compressor(this)
var image_bitmap = imageCompressor
?.setMaxWidth(200)
?.setMaxHeight(200)
?.setQuality(75)
?.compressToBitmap(actualImageFile)
profile_image?.setImageBitmap(image_bitmap)
dialogs?.show()
try{
FirebaseStorage.getInstance().reference.child("profile_images").child(FirebaseAuth.getInstance().currentUser?.uid + ".jpg").putFile(resultUri)
.addOnCompleteListener { task: Task<UploadTask.TaskSnapshot> ->
if (task.isSuccessful) {
showMessage("image uploaded")
val baos = ByteArrayOutputStream()
image_bitmap?.compress(Bitmap.CompressFormat.JPEG, 100, baos)
FirebaseStorage.getInstance().reference.child("profile_images").child("thumbs_images").child(FirebaseAuth.getInstance().currentUser?.uid + ".jpg").putBytes(baos.toByteArray())
.addOnCompleteListener { …我有两种方法。让它看到第一个的模型。
open class CommentModel {
var postid: String? = null
var ownerid: String? = null
var created: Date? = null
var message: String? = null
constructor() {
}
constructor(postid: String?, ownerid: String?, message: String?, created: Date?) {
this.ownerid = ownerid
this.created = created
this.postid = postid
this.message = message
}
}
在这个模型中。我有ownerid。我需要开始一个新的调用来获取所有者的 UserModel。
所以：
commentRepository.getPostCommentsById(postId)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(
{ commentModel ->
// it = the owner of comment.
userRepository.getUserDetailsByUid(commentModel.ownerid!!)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(
{ userModel ->
val comment = CommentWithOwnerModel(commentModel,usermodel)
view.loadComment(comment) …我有一个关于 RxJava 的一般性问题。我在很多地方看到说完全不建议在一个管道步骤中修改对象。您应该创建一个新的并将其传递下去，或doOnNext()用于副作用。
例如
Observable.fromIterable(users)
.map(user -> {
user.name =//do something with name);
user.age = // do something with age);
return user;
}).subscribe(user -> {
});
如您所见，用户对象正在管道步骤中被修改。但是这段代码运行良好，没有任何问题。但是为什么它被认为是一种不好的做法呢？任何人都可以更好地解释为什么对象在整个流中应该是不可变的？
我正在使用 Play Services Auth api Phone，到目前为止我有以下内容
fun startSmsListener() {
val client = SmsRetriever.getClient(applicationContext /* context */);
val task = client.startSmsRetriever();
task.addOnSuccessListener(object : OnSuccessListener<Void> {
override fun onSuccess(p0: Void?) {
//do somethin
}
})
task.addOnFailureListener(object : OnFailureListener {
override fun onFailure(p0: Exception) {
//Handle error
}
})
}
现在我想把它放在一个 SmsManager 类中，并将它转换成一个 Single/Observable，这样我就可以在我的视图模型中以一种反应方式处理它。我怎样才能做到这一点？
到目前为止，我有这个：
var single = Single.create(SingleOnSubscribe<Void> { e ->
val task = client.startSmsRetriever()
task.addOnSuccessListener {
e.onSuccess(it)
}
task.addOnFailureListener {
e.onError(it)
}
})
但我不确定这段代码是否正确，是否有遗漏的东西，比如在处理后删除监听器。
有什么帮助吗？
我正在使用 rxjava 2 并尝试使用 rxbus 传递值
接收总线代码
public class SeasonTabSelectorBus {
private static SeasonTabSelectorBus instance;
private PublishSubject<Object> subject = PublishSubject.create();
public static SeasonTabSelectorBus instanceOf() {
if (instance == null) {
instance = new SeasonTabSelectorBus();
}
return instance;
}
public void setTab(Object object) {
try {
subject.onNext(object);
subject.onComplete();
} catch (Exception e) {
e.printStackTrace();
}
}
public Observable<Object> getSelectedTab() {
return subject;
}
}
我将值设置为
SeasonTabSelectorBus.instanceOf().setTab(20);
这是我订阅的代码
SeasonTabSelectorBus.instanceOf().getSelectedTab().subscribe(new Observer<Object>(){
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Object o) …大家好，我在传递参数时遇到了一些问题。我有 2 个 API 调用，如果设备没有互联网连接，我需要将调用传递回另一个类，因为每个 api 都有不同的 pojo，android 给我警告不匹配类型。这是我的代码
override fun initRetrofitCarList(call: Observable<Response<MyCarModel>>) {
compositeDisposable.add(call
.map{
it.body()
}
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe({
if (it.status == Constants.Result.SUCCESS) {
mPresenter.successGetVehicleList(it.vehicle, "api")
} else {
mPresenter.failedGetVehicleList(it.message)
}
}, {
mPresenter.onNoConnection(call) <- this here will give WARNING because of mismatch type
}))
}
override fun initRetrofitBikeList(call: Observable<Response<MyBikeModel>>) {
compositeDisposable.add(call
.map{
it.body()
}
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe({
if (it.status == Constants.Result.SUCCESS) {
mPresenter.successGetVehicleList(it.vehicle, "api")
} else {
mPresenter.failedGetVehicleList(it.message)
}
}, {
mPresenter.onNoConnection(call) <- this here will give …我想使用 RxJava2 在 RecyclerView 项目上设置一个监听器。项目是复选框。我想分别听每个项目。所以我得到一个错误内部类 ViewHolder 的构造函数只能用包含类的接收器调用
返回 TraceAdapter.ViewHolder(view)
class TraceAdapter(private var checkList: List<TraceViewModelRow> = listOf()) :
RecyclerView.Adapter<TraceAdapter.ViewHolder>() {
private val publishSubject = PublishSubject.create<Event>()
val events: Observable<Event> = publishSubject
override fun onCreateViewHolder(parent: ViewGroup, viewType: Int): TraceAdapter.ViewHolder {
val view = LayoutInflater.from(parent.context).inflate(R.layout.item_trace_task, parent, false)
return TraceAdapter.ViewHolder(view)
}
override fun getItemCount(): Int = checkList.size
override fun onBindViewHolder(holder: TraceAdapter.ViewHolder, position: Int) {
holder.bindTraceList(checkList[position])
}
override fun onViewRecycled(holder: ViewHolder) {
super.onViewRecycled(holder)
}
inner class ViewHolder(override val containerView: View) : RecyclerView.ViewHolder(containerView), LayoutContainer {
fun bindTraceList(trace: …我有一个带有 Room 数据库的简单 Android 应用程序，我正在尝试@Insert使用 RxJava对查询做出反应，但我无法正确链接调用。
这是我调用插入的视图模型方法：
fun insertTopic(): Single<Long> {
val topic = Topic(null, topicText.value!!, difficulty.value!!, false)
return Single.create<Long> { Observable.just(topicDao.insert(topic)) }
}
这是我的活动中触发保存操作的代码：
disposable.add(RxView.clicks(button_save)
.flatMapSingle {
viewModel.insertTopic()
.subscribeOn(Schedulers.io())
}.observeOn(AndroidSchedulers.mainThread())
.doOnError { Toast.makeText(this, "Error inserting topic", Toast.LENGTH_SHORT).show() }
.subscribe { id ->
// NOT INVOKED
hideKeyboard()
Toast.makeText(this, "Topic inserted. ID: $id", Toast.LENGTH_SHORT).show()
this.finish
})
当我单击按钮时，实体被保存但没有调用任何订阅代码（没有显示吐司）。有人可以向我指出我做错了什么吗？我对 RX java 相当陌生。
rx-java2 ×10
android ×8
kotlin ×6
rx-java ×4
rx-android ×2
android-room ×1
firebase ×1
generics ×1
java ×1
java-8 ×1
rx-binding ×1