Reactive Java Spring Boot with MongoDB

1
git clone git@github.com:mongodb-developer/mdb-spring-boot-reactive.git

1
@RestController
2
public class AccountController {
3
   //...
4
   @PostMapping("/account")
5
   public Mono<Account> createAccount(@RequestBody Account account) {
6
       return accountRepository.save(account);
7
   }
8
9
   @GetMapping("/account/{accountNum}")
10
   public Mono<Account> getAccount(@PathVariable String accountNum) {
11
       return accountRepository.findByAccountNum(accountNum).switchIfEmpty(Mono.error(new AccountNotFoundException()));
12
   }
13
   //...
14
}

1
public interface AccountRepository extends ReactiveMongoRepository<Account, String> {
2
  
3
   @Query("{accountNum:'?0'}")
4
   Mono<Account> findByAccountNum(String accountNum);
5
   //...
6
}

1
@RestController
2
public class AccountController {
3
   //...
4
   @PostMapping("/account/{accountNum}/debit")
5
   public Mono<Txn> debitAccount(@PathVariable String accountNum, @RequestBody Map<String, Object> requestBody) {
6
       //...
7
       txn.addEntry(new TxnEntry(accountNum, amount));
8
       return txnService.saveTransaction(txn).flatMap(txnService::executeTxn);
9
   }
10
11
   @PostMapping("/account/{accountNum}/credit")
12
   public Mono<Txn> creditAccount(@PathVariable String accountNum, @RequestBody Map<String, Object> requestBody) {
13
       //...
14
       txn.addEntry(new TxnEntry(accountNum, -amount));
15
       return txnService.saveTransaction(txn).flatMap(txnService::executeTxn);
16
   }
17
18
   @PostMapping("/account/{from}/transfer")
19
   public Mono<Txn> transfer(@PathVariable String from, @RequestBody TransferRequest transferRequest) {
20
       //...
21
       txn.addEntry(new TxnEntry(from, -amount));
22
       txn.addEntry(new TxnEntry(to, amount));
23
       //save pending transaction then execute
24
       return txnService.saveTransaction(txn).flatMap(txnService::executeTxn);
25
   }
26
   //...
27
}

1
@Service
2
public class TxnService {
3
   //...
4
   public Mono<Txn> saveTransaction(Txn txn) {
5
       return txnTemplate.save(txn);
6
   }
7
8
   public Mono<Txn> executeTxn(Txn txn) {
9
       return updateBalances(txn)
10
               .onErrorResume(DataIntegrityViolationException.class
11
                       /*lambda expression to handle error*/)
12
               .onErrorResume(AccountNotFoundException.class
13
                       /*lambda expression to handle error*/)
14
               .then(txnTemplate.findAndUpdateStatusById(txn.getId(), TxnStatus.SUCCESS));
15
   }
16
17
   public Flux<Long> updateBalances(Txn txn) {
18
       //read entries to update balances, concatMap maintains the sequence
19
       Flux<Long> updatedCounts = Flux.fromIterable(txn.getEntries()).concatMap(
20
               entry -> accountRepository.findAndIncrementBalanceByAccountNum(entry.getAccountNum(), entry.getAmount())
21
           );
22
       return updatedCounts.handle(/*...*/);
23
   }
24
}

1
public interface AccountRepository extends ReactiveMongoRepository<Account, String> {
2
   //...
3
   @Update("{'$inc':{'balance': ?1}}")
4
   Mono<Long> findAndIncrementBalanceByAccountNum(String accountNum, double increment);
5
}

1
@Service
2
public class TxnTemplate {
3
   //...
4
   public Mono<Txn> save(Txn txn) {
5
       return template.save(txn);
6
   }
7
8
   public Mono<Txn> findAndUpdateStatusById(String id, TxnStatus status) {
9
       Query query = query(where("_id").is(id));
10
       Update update = update("status", status);
11
       FindAndModifyOptions options = FindAndModifyOptions.options().returnNew(true);
12
       return template.findAndModify(query, update, options, Txn.class);
13
   }
14
   //...
15
}

1
@Configuration
2
public class ReactiveMongoConfig extends AbstractReactiveMongoConfiguration {
3
   //...
4
   @Bean
5
   ReactiveMongoTransactionManager transactionManager(ReactiveMongoDatabaseFactory dbFactory) {
6
       return new ReactiveMongoTransactionManager(dbFactory);
7
   }
8
}

1
@Service
2
public class TxnService {
3
   //In the actual code we are using constructor injection instead of @Autowired
4
   //Using @Autowired here to keep code snippet concise
5
   @Autowired
6
   private TransactionalOperator transactionalOperator;
7
   //...
8
   public Mono<Txn> executeTxn(Txn txn) {
9
       return updateBalances(txn)
10
               .onErrorResume(DataIntegrityViolationException.class
11
                       /*lambda expression to handle error*/)
12
               .onErrorResume(AccountNotFoundException.class
13
                       /*lambda expression to handle error*/)
14
               .then(txnTemplate.findAndUpdateStatusById(txn.getId(), TxnStatus.SUCCESS))
15
               .as(transactionalOperator::transactional);
16
   }
17
   //...
18
}

1
public class TxnService {
2
   //...
3
   @Transactional
4
   public Mono<Txn> executeTxn(Txn txn) {
5
       return updateBalances(txn)
6
               .onErrorResume(DataIntegrityViolationException.class
7
                       /*lambda expression to handle error*/)
8
               .onErrorResume(AccountNotFoundException.class
9
                       /*lambda expression to handle error*/)
10
               .then(txnTemplate.findAndUpdateStatusById(txn.getId(), TxnStatus.SUCCESS));
11
   }
12
   //...
13
}