Save Multiple Records using Web Flux and Mongo DB

Question

I'm working on a project which uses Spring web Flux and mongo DB and I'm very new to reactive programming and WebFlux.

I have scenario of saving into 3 collections using one service. For each collection im generating id using a sequence and then save them. I have FieldMaster which have List on them and every Field Info has List . I need to save FieldMaster, FileInfo and FieldOption. Below is the Code i'm using. The code works only when i'm running on debugging mode, otherwise it get blocked on below line

Integer field_seq_id = Integer.parseInt(sequencesCollection.getNextSequence(FIELDINFO).block().getSeqValue());

Here is the full code

public Mono< FieldMaster > createMasterData(Mono< FieldMaster > fieldmaster) {

    return fieldmaster.flatMap(fm -> {
        return sequencesCollection.getNextSequence(FIELDMASTER).flatMap(seqVal -> {
            LOGGER.info("Generated Sequence value :" + seqVal.getSeqValue());
            fm.setId(Integer.parseInt(seqVal.getSeqValue()));
            List<FieldInfo> fieldInfo = fm.getFieldInfo();
            fieldInfo.forEach(field -> {
                // saving Field Goes Here
                Integer field_seq_id = Integer.parseInt(sequencesCollection.getNextSequence(FIELDINFO).block().getSeqValue()); // stops execution at this line
                LOGGER.info("Generated Sequence value  Field Sequence:" + field_seq_id);
                field.setId(field_seq_id);
                field.setMasterFieldRefId(fm.getId());
                mongoTemplate.save(field).block();
                LOGGER.info("Field Details Saved");
                List<FieldOption> fieldOption = field.getFieldOptions();
                fieldOption.forEach(option -> {
                    // saving Field Option Goes Here
                    Integer opt_seq_id = Integer.parseInt(sequencesCollection.getNextSequence(FIELDOPTION).block().getSeqValue());
                    LOGGER.info("Generated Sequence value Options Sequence:" + opt_seq_id);
                    option.setId(opt_seq_id);
                    option.setFieldRefId(field_seq_id);
                    mongoTemplate.save(option).log().block();
                    LOGGER.info("Field Option Details Saved");
                });
            });
            return mongoTemplate.save(fm).log();
        });
    });

}

Ricard Kollcaku Ricard Kollcaku · Accepted Answer · 2019-09-26T14:04:51

First in reactive programming is not good to use .block because you turn nonblocking code to blocking. if you want to get from a stream and save in 3 streams you can do like that. There are many different ways to do that for performance purpose but it depends of the amount of data. here you have a sample using simple data and using concat operator but there are even zip and merge. it depends from your needs.

  public void run(String... args) throws Exception {
    Flux<Integer> dbData = Flux.range(0, 10);

    dbData.flatMap(integer -> Flux.concat(saveAllInFirstCollection(integer), saveAllInSecondCollection(integer), saveAllInThirdCollection(integer))).subscribe();
}

Flux<Integer> saveAllInFirstCollection(Integer integer) {
    System.out.println(integer);
    //process and save in collection
    return Flux.just(integer);
}

Flux<Integer> saveAllInSecondCollection(Integer integer) {
    System.out.println(integer);
    //process and save in collection
    return Flux.just(integer);
}

Flux<Integer> saveAllInThirdCollection(Integer integer) {
    System.out.println(integer);
    //process and save in collection
    return Flux.just(integer);
}

Save Multiple Records using Web Flux and Mongo DB

1 Answers