สตรีมผลลัพธ์ REST ที่สังเกตได้แบบกำหนดเองจากรายการลิงก์

ฉันกำลังใช้ไลบรารีที่ใช้คำขอ REST ของตัวเอง ใช้ 'ลิงก์ถาวร' และออกคำขอ REST เพื่อดึงเนื้อหา (ฉันจัดเตรียมการเยาะเย้ยในซอร์สโค้ดสำหรับการทดสอบ) ฉันมีรายการลิงก์ถาวร และต้องการกำหนดเวลาคำขอสำหรับลิงก์แต่ละรายการและสร้างสตรีมผลลัพธ์เนื้อหา ทั้งหมดควรเป็นแบบอะซิงโครนัส และเมื่อผลลัพธ์ทั้งหมดเสร็จสิ้น ฉันต้องการปล่อยรายการผลลัพธ์เหล่านั้น

ฉันกำลังพยายามบรรลุเป้าหมายนี้โดยใช้ RxJava นี่คือสิ่งที่ฉันมีตอนนี้ (ขออภัยที่ไม่ได้ใช้ lambdas ฉันแค่อยากทำความคุ้นเคยกับชื่อคลาส RxJava):

public class Main {

    public static void main(String[] args) {
        int count = 10;
        List<String> permalinks = new ArrayList<>(count);
        for (int i = 1; i <= count; ++i) {
            permalinks.add("permalink_" + i);
        }

        ContentManager cm = new ContentManager();

        Observable
                .create(new Observable.OnSubscribe<Content>() {

                    int gotCount = 0;

                    @Override
                    public void call(Subscriber<? super Content> subscriber) {
                        for (String permalink : permalinks) { // 1. is iterating here the correct way?
                            if (!subscriber.isUnsubscribed()) { // 2. how often and where should I check isUnsubscribed?
                                cm.getBasicContentByPermalink(permalink, new RestCallback() {

                                    @Override
                                    public void onSuccess(Content content) {
                                        if (!subscriber.isUnsubscribed()) {
                                            subscriber.onNext(content);
                                            completeIfFinished(); // 3. if guarded by isUnsubscribed, onComplete might never be called
                                        }
                                    }

                                    @Override
                                    public void onFailure(int code, String message) {
                                        if (!subscriber.isUnsubscribed()) {
                                            subscriber.onNext(null); // 4. is this OK or is there some other way to mark a failure?
                                            completeIfFinished();
                                        }
                                    }

                                    private void completeIfFinished() {
                                        ++gotCount;
                                        if (gotCount == permalinks.size()) { // 5. how to know that the last request is done? am I supposed to implement such custom logic?
                                            subscriber.onCompleted();
                                        }
                                    }
                                });
                            }
                        }
                    }
                })
                .toList()
                .subscribe(new Action1<List<Content>>() {

                    @Override
                    public void call(List<Content> contents) {
                        System.out.println("list count: " + contents.size());
                        System.out.println("results: ");
                        contents.stream().map(content -> content != null ? content.basic : null).forEach(System.out::println);
                    }
                });

        System.out.println("finishing main");
    }
}

// library mocks

class Content {

    String basic;

    String extended;

    public Content(String basic, String extended) {
        this.basic = basic;
        this.extended = extended;
    }
}

interface RestCallback {

    void onSuccess(Content content);

    void onFailure(int code, String message);
}

class ContentManager {

    private final Random random = new Random();

    public void getBasicContentByPermalink(String permalink, RestCallback callback) {
        // just to simulate network latency and unordered results
        new Thread() {

            @Override
            public void run() {
                try {
                    Thread.sleep(random.nextInt(1000) + 200);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                if (random.nextInt(100) < 95) {
                    // 95% of the time we succeed
                    callback.onSuccess(new Content(permalink + "_basic", null));
                } else {
                    callback.onFailure(-1, permalink + "_basic_failure");
                }
            }
        }.start();
    }
}

ใช้งานได้นิดหน่อย แต่ฉันไม่แน่ใจว่าฉันทำสิ่งที่ถูกต้องหรือไม่ โปรดดูบรรทัดที่ 1-5:

1. เมื่อสร้าง Observable จากรายการ ฉันควรจะวนซ้ำรายการด้วยตัวเอง หรือมีวิธีอื่นที่ดีกว่านี้หรือไม่ ตัวอย่างเช่นมี Observable.from(Iterable) แต่ฉันไม่คิดว่าจะใช้งานได้ใช่ไหม
2. ฉันกำลังตรวจสอบ isUnsubscribed ก่อนที่จะส่งคำขอ REST และในตัวจัดการผลลัพธ์ทั้งสองด้วย (สำเร็จ/ล้มเหลว) นี่คือวิธีที่ควรจะใช้ใช่ไหม?
3. ฉันใช้ตรรกะบางอย่างเพื่อเรียก onComplete เมื่อคำขอทั้งหมดกลับมา ไม่ว่าจะสำเร็จหรือล้มเหลว (ดูคำถามที่ 5) แต่เนื่องจากพวกเขาได้รับการปกป้องโดยการเรียก isUnsubscribed จึงอาจเกิดขึ้นได้ว่าการโทร onComplete จะไม่ถูกเรียกเลย กระแสรู้ได้อย่างไรว่าควรจะเสร็จสิ้น? จะเสร็จสิ้นก่อนกำหนดหรือไม่เมื่อสมาชิกยกเลิกการสมัครและฉันไม่ต้องคิดเกี่ยวกับเรื่องนี้? มีข้อแม้หรือไม่? ตัวอย่างเช่น จะเกิดอะไรขึ้นหากผู้สมัครสมาชิกยกเลิกการสมัครรับข้อมูลในขณะที่ปล่อยผลลัพธ์เนื้อหาออกมา ในการทดสอบของฉัน รายการผลลัพธ์ทั้งหมดจะไม่ถูกปล่อยออกมา แต่ฉันคาดหวังว่ารายการผลลัพธ์ทั้งหมดจนถึงจุดนั้น
4. ในเมธอด onFailure ฉันส่งค่า null ด้วย onNext(null) เพื่อทำเครื่องหมายความล้มเหลว ฉันทำเพราะท้ายที่สุดแล้ว ฉันจะมี zip 2 สตรีม และจะปล่อยอินสแตนซ์ของคลาสที่กำหนดเองเฉพาะเมื่อค่า zip ทั้งสองไม่เป็นค่าว่าง นี่เป็นวิธีที่ถูกต้องหรือไม่?
5. ดังที่ได้กล่าวไปแล้ว ฉันมีตรรกะที่กำหนดเองเพื่อตรวจสอบว่าการสตรีมเสร็จสิ้นหรือไม่ สิ่งที่ฉันทำที่นี่คือนับผลลัพธ์ REST และเมื่อประมวลผลได้มากเท่าที่มีลิงก์ถาวรในรายการ ก็เป็นอันเสร็จสิ้น สิ่งนี้จำเป็นหรือไม่? นี่เป็นวิธีที่ถูกต้องหรือไม่?