如何将许多AsyncTasks转换为Android中的Rx Observables？

Question

我正在使用Facebook Graph API来查找喜欢我的帖子的用户。所有的逻辑实现了几个步骤：

1. 查找做出请求API中的AsyncTask
2. 转换所有帖子这个AT到Rx.Observable
3. 地图观测量GraphResponse到列表（Post是POJO）
4. 呼叫FlatMap运算符和它的body调用方法，它迭代每个帖子并创建asyncTask，然后将其转换为Observable并放入Array。
5. 在一个Observable中合并观察点阵列。
6. 其映射到GraphResponse喜欢简介
7. 订阅和渲染Likers列表
8. 利润！

我在步骤＃4-5有一些小问题。请看存储库中的方法'喜欢'。在评论我写的问题

提示：我使用的MVP +清洁架构库（数据层）和交互器（经营层）

class FacebookRepository { 

    private val facebook = Facebook.instance() 
    private val gson = GsonBuilder().create() 

    fun posts(): Observable<GraphResponse>? { 

     return RxDecorator<GraphResponse>().decorate(Observable.defer({ 
      val request = GraphRequest(
        facebook.token, 
        "/me/posts", 
        null, 
        HttpMethod.GET, 
        GraphRequest.Callback { /* handle the result */ } 
      ) 

      Observable.just(request.executeAndWait()) 
     })) 

    } 

    fun setFaceBookAccessToken(currentAccessToken: AccessToken?) { 
     facebook.token = currentAccessToken 
    } 

    fun logout() { 
     facebook.logout() 
    } 

    fun token(): String? { 
     return facebook.token?.token 
    } 

    fun likes(posts: List<Post>?): Observable<List<Profile>> { 

     Log.d("observables:posts", posts.toString()) 

     val p = iterateObservables(posts) // STOP HERE and WAIT to complete this method. 

     // Then p is composite - merge and return 
     return Observable.merge(p).map { 
      Log.d("merge:posts", it.toString()) 
      val profiles = gson.fromJson<List<Profile>>(
        it.jsonObject["data"].toString(), 
        object : TypeToken<List<Profile>>() {}.type 
      ) 

      return@map profiles 
     } 
    } 

    private fun iterateObservables(posts: List<Post>?): MutableList<Observable<GraphResponse>>? { 
     val observables: MutableList<Observable<GraphResponse>>? = null 

     Log.d("iterateObs:posts", posts.toString()) 
     Log.d("posts_not_null", (posts != null).toString()) 
     Log.d("posts.size", posts?.size.toString()) 

     if (posts != null) { 
      for (post in posts) { 

       Log.d("iterateObs:post", post.toString()) 

       val request = GraphRequest(
          AccessToken.getCurrentAccessToken(), 
          "/${post.id}/likes", 
          null, 
          HttpMethod.GET, 
          GraphRequest.Callback { Log.d("fb:post:id", "${post.id}: ${it.jsonObject}") 
         }).executeAsync() 

       Log.d("obs:after:post", observables.toString()) 
      } 
     } 

     return observables 
    } 
} 

这是交互器

class FacebookInteractor { 

    private val callbackManager = com.facebook.CallbackManager.Factory.create() 
    private val repository = FacebookRepository() 

    fun facebookAuth(view: IMainView) { 
     LoginManager 
       .getInstance() 
       .logInWithReadPermissions(
         view.getContext() as MainActivity, 
         Arrays.asList("user_friends", "user_likes", "user_posts", "public_profile", "email") 
       ) 
    } 

    fun onFacebookLoginResult(requestCode: Int, resultCode: Int, data: Intent) { 
     callbackManager.onActivityResult(requestCode, resultCode, data) 
    } 

    fun facebookAccessTokenChanged(oldAccessToken: AccessToken?, currentAccessToken: AccessToken?) { 
     if(oldAccessToken?.token != currentAccessToken?.token) { 
      repository.setFaceBookAccessToken(currentAccessToken) 
     } 
    } 

    fun likes(): Observable<List<Profile>>? { 
     return repository.posts()?.map { 
      val gson = GsonBuilder().create() 
      val posts = gson.fromJson<List<Post>>(
        it.jsonObject["data"].toString(), 
        object : TypeToken<List<Post>>() {}.getType() 
      ) 

      return@map posts 
     }?.flatMap { 
      return@flatMap repository.likes(it) 
     } 
    } 

    fun logout() { 
     repository.logout() 
    } 

    fun isLogined(): Boolean { 
     return repository.token() != null 
    } 

} 

我使用Kotlin作为开发语言。

Answer 1

谢谢大家！），但答案是另一种情况）的问题是在逻辑上，而不是实现）的Facebook SDK具有RequestBatch和未来或回采线程没有必要）

package com.github.scrobot.likes_listener.data.facebook 

import android.util.Log 
import com.facebook.* 
import com.github.scrobot.likes_listener.data.facebook.models.Facebook 
import com.github.scrobot.likes_listener.data.facebook.models.Post 
import com.github.scrobot.likes_listener.data.facebook.models.Profile 
import com.google.gson.GsonBuilder 
import com.google.gson.reflect.TypeToken 
import rus.pifpaf.client.util.rx.RxDecorator 
import rx.Observable 
import java.util.* 


/** 
* Created by aleksejskrobot on 07.12.16. 
*/ 
class FacebookRepository { 

    private val facebook = Facebook.instance() 
    private val gson = GsonBuilder().create() 

    fun posts(): Observable<GraphResponse>? { 

     return RxDecorator<GraphResponse>().decorate(Observable.defer({ 
      val request = GraphRequest(
        facebook.token, 
        "/me/posts", 
        null, 
        HttpMethod.GET, 
        GraphRequest.Callback { /* handle the result */ } 
      ) 

      Observable.just(request.executeAndWait()) 
     })) 

    } 

    fun setFaceBookAccessToken(currentAccessToken: AccessToken?) { 
     facebook.token = currentAccessToken 
    } 

    fun logout() { 
     facebook.logout() 
    } 

    fun token(): String? { 
     return facebook.token?.token 
    } 

    fun likes(posts: List<Post>?): Observable<List<Profile>> { 

     val batch = GraphRequestBatch() 

     posts?.mapTo(batch) { post -> 
      GraphRequest(
        facebook.token, 
        "/${post.id}/likes", 
        null, 
        HttpMethod.GET, 
        GraphRequest.Callback { Log.d("fb:post:id", "${post.id}: ${it.jsonObject}") } 
      ) 
     } 

     return RxDecorator<List<GraphResponse>>().decorate(rx.Observable.defer({ 
      rx.Observable.just(batch.executeAndWait()) 
     })).map { 
      Log.d("batchResponse:it", it.toString()) 

      val list = ArrayList<Profile>() 

      for (i in it) { 
       list.addAll(gson.fromJson<List<Profile>>(
         i.jsonObject["data"].toString(), 
         object : TypeToken<List<Profile>>() {}.type 
       )) 
      } 

      list.filter { 
       list.contains(it) 
      } 

      return@map list 
     } 
    } 
}