RxJava阀门用例

Question

RxJava中是否有一个操作员、一个外部库或一种我不知道的方法来创建一个可流动/可观察的，接收控制数据排放的函数，比如阀门？

我有一个巨大的json文件需要处理，但我必须得到文件的一部分，一个实体列表，处理它，然后得到另一部分，我尝试使用windows（），buffer（），但我传递给Flowable的双函数。generate（）在收到第一个列表但尚未完成处理后继续执行。我也试过流动变形金刚。我是胡。阿卡诺克。rxjava3。但它只是在处理列表的flatMap（）函数之前堆积项目

private Flowable<T> flowable(InputStream inputStream) {

    return Flowable.generate(() -> jsonFactory.createParser(new GZIPInputStream(inputStream)), (jsonParser, emitter) -> {

        final var token = jsonParser.nextToken();

        if (token == null) {
            emitter.onComplete();
        }

        if (JsonToken.START_ARRAY.equals(token) || JsonToken.END_ARRAY.equals(token)) {
            return jsonParser;
        }

        if (JsonToken.START_OBJECT.equals(token)) {
            emitter.onNext(reader.readValue(jsonParser));
        }

        return jsonParser;
    }, JsonParser::close);
}

编辑：我需要控制项目的释放，以避免内存和处理数据的函数过载，因为该函数读取和写入数据库，处理也需要按顺序进行。处理数据的函数不完全是我的，它是用RxJava编写的，我希望使用Rx。

我成功地解决了这个问题，但如果有其他方法，请告诉我：

public static <T> Flowable<T> flowable(InputStream inputStream, JsonFactory jsonFactory, ObjectReader reader, Supplier<Boolean> booleanSupplier) {
    return Flowable.generate(() -> jsonFactory.createParser(new GZIPInputStream(inputStream)), (jsonParser, emitter) -> {

        if (booleanSupplier.get()) {
            final var token = jsonParser.nextToken();

            if (token == null) {
                emitter.onComplete();
            }

            if (JsonToken.START_ARRAY.equals(token) || JsonToken.END_ARRAY.equals(token)) {
                return jsonParser;
            }

            if (JsonToken.START_OBJECT.equals(token)) {
                emitter.onNext(reader.readValue(jsonParser));
            }

        }
        
        return jsonParser;
    }, JsonParser::close);
}

Edit2：这是我目前使用函数的方式之一

public Flowable<List<T>> paging(Function<List<T>, Single<List<T>>> function) {
    final var atomicInteger = new AtomicInteger(0);
    final var atomicBoolean = new AtomicBoolean(true);

    return flowable(inputStream, jsonFactory, reader, atomicBoolean::get)
            .buffer(pageSize)
            .flatMapSingle(list -> {

                final var counter = atomicInteger.addAndGet(1);

                if (counter == numberOfPages) {
                    atomicBoolean.set(false);
                }

                return function.apply(list)
                        .doFinally(() -> {
                            if (atomicInteger.get() == numberOfPages) {
                                atomicInteger.set(0);
                                atomicBoolean.set(true);
                            }
                        });
            });
}

Answer 1

就这样解决了

 public static Flowable<Object> flowable(JsonParser jsonParser, ObjectReader reader, PublishProcessor<Boolean> valve) {
    return Flowable.defer(() -> {
        final var token = jsonParser.nextToken();

        if (token == null) {

            return Completable.fromAction(jsonParser::close)
                    .doOnError(Throwable::printStackTrace)
                    .onErrorComplete()
                    .andThen(Flowable.empty());
        }


        if (JsonToken.START_OBJECT.equals(token)) {
            final var value = reader.readValue(jsonParser);
            final var just = Flowable.just(value).compose(FlowableTransformers.valve(valve, true));
            return Flowable.concat(just, flowable(jsonParser, reader, valve));
        }


        return flowable(jsonParser, reader, valve);
    });
}