gistfile1.txt

    private final Subject<CurrentJob, CurrentJob> jobSubject = PublishSubject.create();
    private final Observable<CurrentJob> jobObservable = jobSubject
            .flatMap(job -> api call etc...)
            .doOnNext(job -> job.toReadEvents().forEach(this::processJobReadEvent));
    ...
    startDocument() {
        jobObservable.subscribe();
    }
    
    endJob() {
        jobSubject.onNext(currentJob);
    }
    
    endDocument() {
        jobSubject.onComplete();
        jobObservable.??? // wait for the observable to complete.
    }