{
Logger log = LoggerFactory.getLogger(StorageStrategy.class.getCanonicalName());
- default Flux<ChatRoomInfo> write(ChatHomeService chatHomeService)
+ default Mono<Void> write(ChatHomeService chatHomeService)
{
- return writeChatRoomInfo(
- chatHomeService
- .getChatRoomInfo()
- .flatMap(chatRoomInfo -> writeChatRoomData(
- chatRoomInfo.getId(),
- chatHomeService
- .getChatRoomData(chatRoomInfo.getId())
- .flatMapMany(chatRoomData -> chatRoomData.getMessages())
- )
- .then(Mono.just(chatRoomInfo))
- .doOnSuccess(emittedChatRoomInfo -> log.info("Stored {}", chatRoomInfo))
- .doOnError(throwable -> log.error("Could not store {}: {}", chatRoomInfo, throwable)))
- )
- .doOnComplete(() -> log.info("Stored {}", chatHomeService))
- .doOnError(throwable -> log.error("Could not store {}: {}", chatHomeService, throwable));
+ return writeChatRoomInfo(chatHomeService.getChatRoomInfo())
+ .flatMap(chatRoomInfo -> writeChatRoomData(
+ chatRoomInfo.getId(),
+ chatHomeService
+ .getChatRoomData(chatRoomInfo.getId())
+ .flatMapMany(chatRoomData -> chatRoomData.getMessages())
+ )
+ .count()
+ .doOnSuccess(count -> log.info("Stored {} messages for {}", count, chatRoomInfo))
+ .doOnError(throwable -> log.error("Could not store {}: {}", chatRoomInfo, throwable)))
+ .count()
+ .doOnSuccess(count -> log.info("Stored {} chat-rooms for {}", count, chatHomeService))
+ .doOnError(throwable -> log.error("Could not store {}: {}", chatHomeService, throwable))
+ .then();
}
Flux<ChatRoomInfo> writeChatRoomInfo(Flux<ChatRoomInfo> chatRoomInfoFlux);