Correct pattern to insert or update existed entity with spring data reactive

Question

I implement an application with spring data (mongo).

• It gets data from external service
• Store it to DB
• If item does not exist I'd like to insert it to db
• If item exists but has same md5 hash with external item - do nothing
• If item exists in DB and md5 hashes are not equal I'd like to update it

I have the code

public Mono<Void> processItems() {  
  return externalService.getItems() //this returns Flux<Item>
         .flatMap(this::createOrUpdateItem)
         .flatMap(itemRepository::save)
         .then() 
}

private Mono<Item> createOrUpdateItem(Item item) {  
  return itemRepository.findById(item.getId) //this returns Mono<Item>
          .flatMap(itemFromDb -> updateItem(itemFromDb, item))
          .defaultIfEmpty(item) 
}

private Mono<Item> updateItem(Item itemFromDb, Item item){
  if(itemFromDb.getMd5.equals(item.getMd5)){
     return Mono.just(itemFromDb);
  }

  itemFromDb.setName(item.getName)
            .setDescription(item.getDescription); 

  return Mono.just(itemFromDb);
}

How does it work:

• If item does not exist it stored in db
• If item exists and has same md5 hash updateItem returns existed in db and version in DB increases
• If item exists and md5 hashes are not equal updateItem set feilds and returns existed in db and version in DB increases

The problem is: when md5 hash are equal I don't want to cal DB. Because it increases version of the item in db but there were no changes. How to properly implement this algorithm?

I can use @EqualsAndHashCode(exclude = "id") but not sure if it is a right way

Answer 1

The only missing ingredient in your example is to continue with an empty stream if nothing needs to be done.

This solution is actually similar to yours, but has a clear separation of concerns.

The first flatMap only loads existing data. The second one contains only the business logic to decide what to do. It's up to you to follow these established principles (Single Responsibility Principle, Integration Operation Segregation Principle etc.).

public Mono<Void> processItems() {  
  return externalService.getItems() //this returns Flux<Item>
         .flatMap(this::loadExistingItem)
         .flatMap(this::setupOperation)
         .flatMap(this::saveItem)
         .then() 
}

private Mono<List<Item>> loadExistingItem(Item item) {  
  return itemRepository.findById(item.getId)
           .map(fromDb -> Arrays.asList(item, fromDb))
           .defaultIfEmpty(() -> Arrays.asList(item, null));
}

private Mono<Item> setupOperation(List<Item> items) {
  Item newItem = items.get(0);
  Item existingItem = items.get(1);

  if (existingItem == null) {
    return Mono.just(newItem);
  }

  if(newItem.getMd5().equals(existingItem.getMd5())){
    return Mono.empty();
  }

  existingItem.setName(newItem.getName)
  existingItem.setDescription(newItem.getDescription);

  return Mono.just(existingItem);
}