1

Буду благодарен за любые подсказки и варианты как это поправить.

Задача: выдавать результаты вычислений асинхронно, по готовности. Пример составлен для теста из рабочего кода, его можно использовать в IDE.

И как приведено в примере - работает и в рабочем примере заполняет ячейки лэйаута рандомно, но иногда один или два результата выпадают и иногда выполнение заканчивается ошибкой.

Думаю, что использовать subject.onNext() напрямую некорректно, т.к. нельзя потом проверить subject.hasComplete(), который не возвращает в таком случае true. Проблема правильно подписать "subject" на все "observable", чтобы код работал асинхронно без ошибок.

Используется RxJava3. Gradle:

 dependencies {
    implementation 'io.reactivex.rxjava3:rxandroid:3.0.0'
    implementation 'io.reactivex.rxjava3:rxjava:3.0.4'
}

Комментарии "TODO" в коде.

 private void run() {
    
    Log.d("APP", "INIT");

    PublishSubject<String> subject = exec(1000000);
    subject.observeOn(AndroidSchedulers.mainThread())
        .subscribe(unit -> {
            Log.d("STRING RESULT = ", unit);
            if (subject.hasComplete()) {
                //TODO: this condition should work if we use observable.subscribe(subject), not subject.onNext
                Log.d("SUBJECT", "COMPLETED");
            }
        }, Throwable::printStackTrace);
}

private int insertValue = 1000000;

public PublishSubject<String> exec(int inputNumber) {
    PublishSubject<String> subject = PublishSubject.create();

    getListObservable(inputNumber)
        .flatMap(resultList -> getOperationsObservable()
        .flatMap(operationElem -> getResultListObservable(resultList)
            .map(listElem ->
                                    calculate(operationElem, listElem)
//TODO: should be smth like this, i.e. calculate.subscribe(subject)
//                        .subscribe(subject)
                                        .subscribe(subject::onNext)

                        )
                )
        ).subscribe();

    return subject;
}


private Observable<ArrayList<List<Integer>>> getListObservable(int inputNumber) {
    return Observable.fromCallable(() -> {

        ArrayList<List<Integer>> list = new ArrayList<>();

        Integer[] populatedArray = new Integer[inputNumber];
        Arrays.fill(populatedArray, insertValue);

        list.add(new ArrayList<>(Arrays.asList(populatedArray)));
        list.add(new LinkedList<>(Arrays.asList(populatedArray)));
        list.add(new CopyOnWriteArrayList<>(Arrays.asList(populatedArray)));

        return list;
    }).subscribeOn(Schedulers.computation());
}


private Observable<String> calculate(ListOperationName operationName, List<Integer> list) {
    return Observable.fromCallable(() -> {
        try {
            TimeUnit.MILLISECONDS.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        double start = getTime();

        synchronized (list) {

            switch (operationName) {
                case ADD_FIRST:
                    list.add(0, insertValue);
                    break;
                case ADD_MID:
                    list.add(list.size() / 2, insertValue);
                    break;
                case ADD_LAST:
                    list.add(insertValue);
                    break;
                case SEARCH:
                    //noinspection unused
                    boolean contains = list.contains(insertValue);
                    break;
                case RM_FIRST:
                    list.remove(0);
                    break;
                case RM_MID:
                    list.remove(list.size() / 2);
                    break;
                case RM_LAST:
                    list.remove(list.size() - 1);
                    break;
            }
        }
        return Double.toString(getTime() - start);
    }).subscribeOn(Schedulers.newThread());
}

private Observable<ListOperationName> getOperationsObservable() {
    return Observable.fromArray(ListOperationName.values());
}

private Observable<List<Integer>> getResultListObservable(ArrayList<List<Integer>> list) {
    return Observable.fromIterable(list);
}

private double getTime() {
    return System.nanoTime();
}

public enum ListOperationName {
    ADD_FIRST,
    ADD_MID,
    ADD_LAST,
    SEARCH,
    RM_FIRST,
    RM_MID,
    RM_LAST;
}

1 ответ 1

0

А зачем вообще этот subject? Попробуй:

public Observable<String> exec(int inputNumber) {
return getListObservable(inputNumber)
    .flatMap(resultList -> getOperationsObservable()
        .flatMap(operationElem -> getResultListObservable(resultList)
            .map(listElem -> calculate(operationElem, listElem))
        )
    );
}

и затем:

private void run() {

Log.d("APP", "INIT");

exec(1000000)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(unit -> {
        Log.d("STRING RESULT = ", unit);
        if (subject.hasComplete()) {
            //TODO: this condition should work if we use observable.subscribe(subject), not subject.onNext
            Log.d("SUBJECT", "COMPLETED");
        }
    }, Throwable::printStackTrace);
}

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.