RX java组合逻辑

Question

我是 reactiveX 的新手，所以这可能是一个愚蠢的问题。 我试图在一个方法调用中一个接一个地组合两个操作。

第 1 步是获取数据的一些详细信息并存储这些详细信息

第 2 步是根据这些详细信息获取数据。

public Observable<ApiResponse> getData(String id, String token)
{
    //step 1
    apiInterface.fetchDataDetails(id)
            .map(new Func1<Map<String, String>, Void>()
            {
                @Override
                public Void call(Map<String, String> details)
                {
                    if (details != null && !details.isEmpty() &&
                            details.containsKey("version") && details.containsKey("count"))
                    {
                        // save details
                    } else
                        throw new RuntimeException("Details not present");
                    return null;
                }
            });



    //step 2 if step 1 is complete i.e data details are saved 
    //then bring data

    int dataCount = 5; // the count saved from fetched details
    List<Observable> dataCalls = new ArrayList<>();
    for (int i = 0; i < dataCount; i++)
 dataCalls.add(apiInterface.fetchData(token,"http://blahblahblah/data/"+i));

    return Observable.zip(dataCalls, new FuncN<ApiResponse>()
    {
        @Override
        public ApiResponse call(Object... args)
        {
            // read all the data from all the observable, 
            //combine with some logic 
            // and then return just one ApiResponse.
            return new ApiResponse();
        }
     });
}

我无法理解的是，我如何结合第 1 步和第 2 步。 仅当成功完成第 1 步时才应执行第 2 步。

我想试试这个。

public Observable<ApiResponse> getData(String id, String token)
{
     return Observable.create(new Observable.OnSubscribe<ApiResponse>()
      {
        @Override
        public void call(Subscriber<? super ApiResponse> subscriber)
        {
            apiInterface.fetchDataDetails(id)
                        .map(new Func1<Map<String, String>, Void>()
                        {
                            @Override
                            public Void call(Map<String, String> details)
                            {
                                if (details != null && !details.isEmpty() &&
                                        details.containsKey("version") && 
                                        details.containsKey("count"))
                                {
                                    // save details
                                } else
                                    throw new RuntimeException("Details not 
                                                             present");
                                return null;
                            }
                        }).doOnCompleted(new Action0()
                            {
                                @Override
                                public void call()
                                {
                                    int dataCount = 5; // the count saved 
                                                      //from fetched details
                                    List<Observable> dataCalls = new 
                                                         ArrayList<>();
                                    for (int i = 0; i < dataCount; i++)
                                    dataCalls.add(apiInterface
                                                 .fetchData(token,
                                        "http://blahblahblah/data/"+i));

                                    return Observable.zip(dataCalls, new 
                                     FuncN<ApiResponse>()
                                    {
                                    @Override
                                    public ApiResponse call(Object... args)
                                    {
                                        // read all the data from all the 
                                       //observable, combine with some logic 
                                        // and then return just one 
                                        //ApiResponse.
                                        return new ApiResponse();
                                    }
                                    });
                                }
                            });
            }
       });
    }

但这样我将无法将我的数据发回，这感觉像是一种反模式。

Answer 1

我在这个问题中找到了答案

https://github.com/ReactiveX/RxJava/issues/2608