RXJava Android 进行多个不同长度的调用

Posted 2023-03-22

【中文标题】RXJava Android 进行多个不同长度的调用

【英文标题】：RXJava Android make multiple calls with different lengths

【发布时间】：2021-10-07 06:33:23

【问题描述】：

我需要从服务器获取相机ID，然后通过ID获取此相机的事件，最后通过ID获取图像。 然后，所有这些都需要组合成一个 UI 帖子 Camera name - image url - it events 例如，我们可能有 5 个摄像头名称、30 个事件 ID（一个摄像头名称对应 6 个）和 30 个 url。 我尝试使用 zip 来组合 Observable api 调用（分别）获取相机名称、事件 ID 和图像 url。但据我了解，由于 zip 无法正确匹配 5 个名称和 30 个 ID，因此它无法正常工作 告诉我什么可以用来做这个？

 GetApiMethods getApiMethods = NetworkService.getInstance().createService(GetApiMethods.class);
    Observable<Camera> cameras = getCamera();
    Observable<Event> events = getEvents();
    Observable<Post> combined = Observable.zip(cameras, events, (camera, event) -> new Post(camera.getDisplayName(),"null",event.getType()));


    public LiveData<String> CameraRequests() 
        combined.subscribeOn(Schedulers.io())
                .observeOn(androidSchedulers.mainThread())
                .concatMap(Observable::fromArray)
                .subscribe(new Observer<Post>() 

                    @Override
                    public void onSubscribe(@NonNull Disposable d) 

                    

                    @Override
                    public void onNext(@NonNull Post post) 
                    

                    @Override
                    public void onError(@NonNull Throwable e) 

                    

                    @Override
                    public void onComplete() 

                    
                );

        return liveData;
    

    public Observable<Event> getEvents() 
        return Observable.create(allEvents -> getApiMethods.getCameras()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .concatMap(Observable::fromIterable)
                .concatMap(Observable::fromIterable)
                .concatMap(camera -> getApiMethods.getEvents(camera.getAccessPoint().replace("hosts", "")))
                .concatMap(Observable::fromIterable)
                .map(Event::getId)
                .subscribe(new Observer<String>() 
                    @Override
                    public void onSubscribe(@NonNull Disposable d) 

                    

                    @Override
                    public void onNext(@NonNull String event) 
                    

                    @Override
                    public void onError(@NonNull Throwable e) 
                        Log.d("RXJAVA1", String.valueOf(e));
                    

                    @Override
                    public void onComplete() 

                    
                )
        );

    



    public Observable<Camera> getCamera() 

        return Observable.create(cameras1 -> getApiMethods.getCameras()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .concatMap(Observable::fromIterable)
                .map(Camera::getDisplayName)
                .subscribe(new Observer<String>() 
                    @Override
                    public void onSubscribe(@NonNull Disposable d) 

                    

                    @Override
                    public void onNext(@NonNull String s) 

                    

                    @Override
                    public void onError(@NonNull Throwable e) 
                        Log.d("RXJAVA", String.valueOf(e));
                    

                    @Override
                    public void onComplete() 
                        

                    
                )

        );

    

【问题讨论】：

@MikkelT 谢谢！但是我对 Livedata 没有任何问题，只有 RXJava

omg 对不起，我一定是回错帖了

你能添加你到目前为止尝试过的代码吗？

@DatPhamTat 你好！我已附上代码

【参考方案1】：

如果使用flatMap，对我来说似乎很期待：

Observable<Post> getPostsObservable() 
        return getCameras().flatMap(camera ->
                // for each camera, build observable to get events
                getEvents(camera.accessPoint).flatMap(event ->
                        // for each event, build observable to get imageUrl
                        getImageUrl(event.id).map(imageUrl ->
                                // combine camera, event and imageUrl in current rx-chain
                                new Post(camera, event, imageUrl)
                        )
                )
        );
    

您可能需要稍微调整您的代码，例如简化您的 getCameras 和 getEvents 方法。您可以查看下面模仿您的应用程序的演示，以更好地了解它的工作原理以及要返工的内容。如果调用 getPosts 方法，您应该在 logcat 中看到帖子正在正确创建和接收。

import android.util.Log;

import androidx.annotation.NonNull;

import java.util.ArrayList;
import java.util.List;

import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

public class RxDemo 
    // your Camera class
    public class Camera 
        private final String accessPoint;

        public Camera(String accessPoint) 
            this.accessPoint = accessPoint;
        

        public String getAccessPoint() 
            return accessPoint;
        
    

    // your Event class
    public class Event 
        private final String id;

        public Event(String id) 
            this.id = id;
        

        public String getId() 
            return id;
        
    

    // your Post class
    public class Post 
        private final Camera camera;
        private final Event event;
        private final String imageUrl;

        public Post(Camera camera, Event event, String imageUrl) 
            this.camera = camera;
            this.event = event;
            this.imageUrl = imageUrl;
        

        public Camera getCamera() 
            return camera;
        

        public Event getEvent() 
            return event;
        

        public String getImageUrl() 
            return imageUrl;
        

        @NonNull
        @Override
        public String toString() 
            return "camera: " + camera.accessPoint + "; event: " + event.id + "; imageUrl: " + imageUrl;
        
    

    // mocked getApiMethods.getCameras(), return 5 mocked cameras
    private Observable<List<Camera>> apiGetCameras() 
        List<Camera> cameras = new ArrayList<>();
        cameras.add(new Camera("cam_1"));
        cameras.add(new Camera("cam_2"));
        cameras.add(new Camera("cam_3"));
        cameras.add(new Camera("cam_4"));
        cameras.add(new Camera("cam_5"));
        return Observable.just(cameras);
    

    // mocked getApiMethods.getEvents, return 6 mocked events with cameraId postFix
    private Observable<List<Event>> apiGetEvents(String cameraId) 
        List<Event> events = new ArrayList<>();
        events.add(new Event("event_1_" + cameraId));
        events.add(new Event("event_2_" + cameraId));
        events.add(new Event("event_3_" + cameraId));
        events.add(new Event("event_4_" + cameraId));
        events.add(new Event("event_5_" + cameraId));
        events.add(new Event("event_6_" + cameraId));
        return Observable.just(events);
    

    // mocked getApiMethods.getImageUrl
    private Observable<String> apiGetImageUrl(String eventId) 
        return Observable.just("http://my.domain.com/" + eventId + ".png");
    

    public Observable<Camera> getCameras() 
        return apiGetCameras().concatMapIterable(cam -> cam);
    

    public Observable<Event> getEvents(String cameraId) 
        return apiGetEvents(cameraId).concatMapIterable(event -> event);
    

    // build observable that emits posts
    private Observable<Post> getPostsObservable() 
        return getCameras().flatMap(camera ->
                // for each camera, build observable to get events
                getEvents(camera.accessPoint).flatMap(event ->
                        // for each event, build observable to get imageUrl
                        apiGetImageUrl(event.id).map(imageUrl ->
                                // combine camera, event and imageUrl in current rx-chain
                                new Post(camera, event, imageUrl)
                        )
                )
        );
    

    // call this in your activity/fragment/viewModel to get the posts
    public void getPosts() 
        getPostsObservable()
                .subscribeOn(Schedulers.io())
                .subscribe(new Observer<Post>() 
                    @Override
                    public void onSubscribe(@NonNull Disposable d) 

                    

                    @Override
                    public void onNext(@NonNull Post post) 
                        Log.d("AAAA", post.toString());
                    

                    @Override
                    public void onError(@NonNull Throwable e) 

                    

                    @Override
                    public void onComplete() 

                    
                );
    

【讨论】：

非常感谢！这完全解决了我的问题。如果可能的话，你能告诉我如何根据 event.getID 添加到从服务器下载的照片吗？ （没有代码，只是理论）

不确定您的下载逻辑是什么样的，但在getPosts 内的onNext 回调中，您拥有发布所需的所有信息，包括imageUrl，因此您可以在那里执行下载.假设您的下载函数名为public void downloadImage(String imageUrl)，那么您将在onNext 回调中调用它，使用post.imageUrl 作为参数。

如果我的评论没有回答您关于图像下载的问题，那么可能会使用代码（您尝试过的内容/您认为可行的内容）和进一步的解释创建一个新问题。如果我能更好地理解您的问题，我会更容易为您提供帮助。

谢谢！新问题***.com/questions/68619492/…

抱歉再次询问，但return cameras.concatMap(camera -> getArchive().concatMap( archive -> getEvents(archive.getAccessPoint().replace("hosts", "")).flatMap(event -> getdetectorImage(archive.getAccessPoint().replace("hosts", ""), event.getTimestamp(), event.getId()).map(image -> new Post(camera.getDisplayName(), image.byteStream(), event.getType())) ))); 无法正常工作，摄像头名称显示为随机显示。我做错了什么？