Convenient complexity for Stream API usage?

Question

I have JSON document which is a result of parsing for a bunch of files:

{
  "offer": {
    "clientName": "Tom",
    "insuranceCompany": "INSURANCE",
    "address": "GAMLE BONDALSVEGEN 53",
    "renewalDate": "22.12.2018",
    "startDate": "22.12.2017",
    "too_old": false,
    "products": [
      {
        "productType": "TRAVEL",
        "objectName": "Reiseforsikring - Holen, Tom Andre",
        "name": null,
        "value": null,
        "isExclude": false,
        "monthPrice": null,
        "yearPrice": 1637,
        "properties": {}
      }
    ]
  },
  "documents": [
    {
      "clientName": "Tom",
      "insuranceCompany": "INSURANCE",
      "fileName": "insurance_tom.pdf",
      "address": "GAMLE BONDALSVEGEN 53",
      "renewalDate": "22.12.2019",
      "startDate": "22.12.2018",
      "issuedDate": "20.11.2018",
      "policyNumber": "6497777",
      "products": [
        {
          "productType": "TRAVEL",
          "objectName": "Reiseforsikring - Holen, Tom Andre",
          "name": null,
          "value": null,
          "isExclude": false,
          "monthPrice": null,
          "yearPrice": 1921,
          "properties": {
            "TRAVEL_PRODUCT_NAME": "Reise Ekstra",
            "TRAVEL_DURATION_TYPE": "DAYS",
            "TRAVEL_TYPE": "FAMILY",
            "TRAVEL_DURATION": "70",
            "TRAVEL_INSURED_CLIENT_NAME": "Holen, Tom Andre, Familie"
          }
        },

I want to iterate throug all products from documents section and set missed properties to products from offer section.

Offer and documents at the same depth level at JSON.

Implementation for this with Stream API is following:

private void mergePropertiesToOffer(InsuranceDocumentsSession insuranceSession) {
    Validate.notNull(insuranceSession, "insurance session can't be null");
    if (insuranceSession.getOffer() == null) return;

    log.info("BEFORE_MERGE");
    // merge all properties by `objectName`
    Stream.of(insuranceSession).forEach(session -> session.getDocuments().stream()
            .filter(Objects::nonNull)
            .flatMap(doc -> doc.getProducts().stream())
            .filter(Objects::nonNull)
            .filter(docProduct -> StringUtils.isNotEmpty(docProduct.getObjectName()))
            .filter(docProduct -> MapUtils.isNotEmpty(docProduct.getProperties()))
            .forEach(docProduct -> Stream.of(session.getOffer())
                    .flatMap(offer -> offer.getProducts().stream())
                    .filter(Objects::nonNull)
                    .filter(offerProduct -> MapUtils.isEmpty(offerProduct.getProperties()))
                    .filter(offerProduct -> StringUtils.isNotEmpty(offerProduct.getObjectName()))
                    .filter(offerProduct -> offerProduct.getObjectName().equals(docProduct.getObjectName()))
                    .forEach(offerProduct -> {
                        try {
                            ObjectMapper mapper = new ObjectMapper().enable(SerializationFeature.INDENT_OUTPUT);
                            log.info("BEFORE_PRODUCT: {}", mapper.writeValueAsString(offerProduct));
                            offerProduct.setProperties(docProduct.getProperties());
                            log.info("UPDATED_PRODUCT: {}", mapper.writeValueAsString(offerProduct));
                        } catch (JsonProcessingException e) {
                            log.error("Error converting product to offer: {}", e.getCause());
                        }
                    })));
    log.info("AFTER_MERGE");
}

It works fine. However, implementing is much faster than maintaining in the future.

There two times I am using Stream.of() factory method for getting a stream for 2 entities at a different level. Also, flatMap() is used as much as possible, + all null checks.

And the question isn't this implementation too difficult?

Should it be refactored and divided into smaller parts? If yes how it should be with good programming principles?

SOLUTION:

Huge thanks to nullpointer answer.
Final solution is following:

Map<Integer, InsuranceProductDto> offerProductMap = session.getOffer().getProducts()
    .stream()
    .filter(this::validateOfferProduct)
    .collect(Collectors.toMap(InsuranceProductDto::getYearPrice, Function.identity(), (first, second) -> first));

Map<Integer, InsuranceProductDto> documentsProductMap = session.getDocuments()
    .stream()
    .flatMap(d -> d.getProducts().stream())
    .filter(this::validateDocumentProduct)
    .collect(Collectors.toMap(InsuranceProductDto::getYearPrice, Function.identity(), (first, second) -> first));

documentsProductMap.forEach((docPrice, docProduct) -> {
    if (offerProductMap.containsKey(docPrice)) {
        offerProductMap.compute(docPrice, (s, offerProduct) -> {
            setProductProperties(offerProduct, docProduct);
            return offerProduct;
        });
    }
}); 
// after finishing execution `offerProductMap` contains updated products

Answer 1

To start off with, you can create a common Predicates for those chained filters as

.filter(offerProduct -> MapUtils.isEmpty(offerProduct.getProperties()))
.filter(offerProduct -> StringUtils.isNotEmpty(offerProduct.getObjectName()))
.filter(offerProduct -> offerProduct.getObjectName().equals(docProduct.getObjectName()))

you can write a Predicate such that

Predicate<OfferProduct> offerProductSelection = offerProduct -> MapUtils.isEmpty(offerProduct.getProperties())
                                    && StringUtils.isNotEmpty(offerProduct.getObjectName())
                                    && offerProduct.getObjectName().equals(docProduct.getObjectName());

and then simply use that as a single filter

.filter(offerProductSelection);

By the way, you could have preferably moved that to a method returning boolean and then used that in the filter.

---

Not precise for the sake of data types and utility classes used, but for the sake of representation, you could do something like :

private void mergePropertiesToOffer(InsuranceDocumentsSession insuranceSession) {
    Validate.notNull(insuranceSession, "insurance session can't be null");
    if (insuranceSession.getOffer() == null) return;
    Map<String, InsuranceProductDto> offerProductMap = insuranceSession.getOffer().getProducts()
            .stream()
            .filter(this::validateOfferProduct)
            .collect(Collectors.toMap(InsuranceProductDto::getObjectName, Function.identity())); // assuming 'objectName' to be unique

    Map<String, InsuranceProductDto> documentsProductMap = insuranceSession.getDocuments()
            .stream()
            .filter(Objects::nonNull)
            .flatMap(d -> d.getProducts().stream())
            .filter(this::validateDocumentProduct)
            .collect(Collectors.toMap(InsuranceProductDto::getObjectName, Function.identity())); // assuming 'objectName' to be unique

    Map<String, Product> productsToProcess = new HashMap<>(documentsProductMap);
    productsToProcess.forEach((k, v) -> {
        if (offerProductMap.containsKey(k)) {
            offerProductMap.compute(k, (s, product) -> {
                Objects.requireNonNull(product).setProperties(v.getProperties());
                return product;
            });
        }
    });

    // now the values of 'offerProductMap' is what you can set as an updated product list under offer
}


private boolean validateDocumentProduct(InsuranceProductDto product) {
    return Objects.nonNull(product)
            && MapUtils.isNotEmpty(product.getProperties())
            && StringUtils.isNotEmpty(product.getObjectName());
}

private boolean validateOfferProduct(InsuranceProductDto offerProduct) {
    return Objects.nonNull(offerProduct)
            && MapUtils.isEmpty(offerProduct.getProperties())
            && StringUtils.isNotEmpty(offerProduct.getObjectName());
}

---

Edit: Based on the comment,

objectName can be the same for a bunch of products

you can update the code to use the merge function as:

Map<String, InsuranceProductDto> offerProductMap = insuranceSession.getOffer().getProducts()
        .stream()
        .filter(this::validateOfferProduct)
        .collect(Collectors.toMap(InsuranceProductDto::getObjectName, Function.identity(), 
                     (a,b) -> {// logic to merge and return value for same keys
                            }));

Answer 2

For each session, all offer products's properties will refer to the last qualified document product's properties, right ?

Because the inner stream will always evaluate to the same result independently of the current document product.

So, while rectifying this, i will suggest the following refactor :

final class ValueWriter
{
    private final static ObjectMapper mapper = new ObjectMapper();

    static
    {
        mapper.enable(SerializationFeature.INDENT_OUTPUT);
    }

    static String writeValue(final Object value) throws JsonProcessingException
    {
        return mapper.writeValueAsString(value);
    }
}

private Optional<Product> firstQualifiedDocumentProduct(final InsuranceDocumentsSession insuranceSession)
{
    return insuranceSession.getDocuments().stream()
        .filter(Objects::notNull)
        .map(Document::getProducts)
        .flatMap(Collection::stream)
        .filter(docProduct -> StringUtils.isNotEmpty(docProduct.getObjectName()))
        .filter(docProduct -> MapUtils.isNotEmpty(docProduct.getProperties()))
        .findFirst()
    ;
}

private void mergePropertiesToOffer(final InsuranceDocumentsSession insuranceSession)
{
    Validate.notNull(insuranceSession, "insurance session can't be null");

    if(insuranceSession.getOffer() == null) return;

    log.info("BEFORE_MERGE");

    final Optional<Product> qualifiedDocumentProduct = firstQualifiedDocumentProduct(insuranceSession);

    if (qualifiedDocumentProduct.isPresent())
    {
        insuranceSession.getOffer().getProducts().stream()
            .filter(Objects::nonNull)
            .filter(offerProduct -> MapUtils.isEmpty(offerProduct.getProperties()))
            .filter(offerProduct -> StringUtils.isNotEmpty(offerProduct.getObjectName()))
            .filter(offerProduct -> offerProduct.getObjectName().equals(qualifiedDocumentProduct.get().getObjectName()))
            .forEach(offerProduct ->
            {
                try
                {
                    log.info("BEFORE_PRODUCT: {}", ValueWriter.writeValueAsString(offerProduct));
                    offerProduct.setProperties(qualifiedDocumentProduct.get().getProperties());
                    log.info("BEFORE_PRODUCT: {}", ValueWriter.writeValueAsString(offerProduct));
                }
                catch (final JsonProcessingException e)
                {
                    log.error("Error converting product to offer: {}", e.getCause());
                }
            })
        ;
    }
}