Spring WebFlux: Reactive MongoDB

Question

我是Spring Reactor的新手，所以我想重构这个简单的spring data (在kotlin上)方法：

fun save(user: User): Mono<User> {
    if (findByEmail(user.email).block() != null) {
        throw UserAlreadyExistsException()
    }

    user.password = passwordEncoder.encode(user.password)
    return userRepository.save(user)
}

谢谢

Answer 1

像这样的东西应该是有效的：

  open fun save(req: ServerRequest): Mono<ServerResponse> {
    logger.info { "${req.method()} ${req.path()}" }
    return req.bodyToMono<User>().flatMap {
      // You might need to "work out" this if since I don't know what you are doing
      if (null != findByEmail(it.email).block()) {
        throw UserAlreadyExistsException()
      }
      it.password = passwordEncoder.encode(it.password)
      repository.save(it).flatMap {
        logger.debug { "Entity saved successfully! Result: $it" }
        ServerResponse.created(URI.create("${req.path()}/${it.id}")).build()
      }
    }
  }

注意，我使用的是MicroUtils/kotlin-logging。如果您不知道或不想要日志语句，请删除它们。

基本上，你需要“消费”(也称为订阅)首先访问ServerRequest中的内容，以便访问内容。

或者，您也可以有一个实际的流来处理该场景，而不是抛出异常；类似于：

open fun ...
  return ServerResponse.ok()
      // Keep doing stuff here...if something is wrong
      .switchIfEmpty(ServerResponse.notFound().build())
}

您可以将示例调整为您的User类型，以防您确实想要传递它而不是ServerRequest。

Answer 2

(如果Kotlin语法是错误的，如果我正在用Java风格做事情，请原谅我:o)

fun save(user: User): Mono<User> {
    //we'll prepare several helpful Monos and finally combine them.
    //as long as we don't subscribe to them, nothing happens.

    //first we want to short-circuit if the user is found (by email).
    //the mono below will onError in that case, or be empty
    Mono<User> failExistingUser = findByEmail(user.email)
        .map(u -> { throw new UserAlreadyExistsException(); });

    //later we'll need to encode the password. This is likely to be
    //a blocking call that takes some time, so we isolate that call
    //in a Mono that executes on the Elastic Scheduler. Note this
    //does not execute immediately, since it's not subscribed to yet...
    Mono<String> encodedPassword = Mono
        .fromCallable(() -> passwordEncoder.encode(user.password))
        .subscribeOn(Schedulers.elastic());

    //lastly the save part. We want to combine the original User with
    //the result of the encoded password.
    Mono<User> saveUser = user.toMono() //this is a Kotlin extension
        .and(encodedPassword, (u, p) -> {
            u.password = p;
            return u;
        })
        //Once this is done and the user has been updated, save it
        .flatMap(updatedUser -> userRepository.save(updatedUser));

   //saveUser above is now a Mono that represents the completion of
   //password encoding, user update and DB save.

   //what we return is a combination of our first and last Monos.
   //when something subscribes to this combination:
   // - if the user is found, the combination errors
   // - otherwise, it subscribes to saveUser, which triggers the rest of the process
   return failExistingUser.switchIfEmpty(saveUser);
}

没有中间变量和注释的简短版本：

fun save(user: User): Mono<User> {
    return findByEmail(u.email)
        .map(u -> { throw new UserAlreadyExistsException(); })
        .switchIfEmpty(user.toMono())
        .and(Mono.fromCallable(() -> passwordEncoder.encode(user.password))
                 .subscribeOn(Schedulers.elastic()),
             (u, p) -> {
                u.password = p;
                return u;
             })
        .flatMap(updatedUser -> userRepository.save(updatedUser));
}

Answer 3

您可以在Mono中使用hasElement()函数。看看Mono的扩展函数：

inline fun <T> Mono<T>.errorIfEmpty(crossinline onError: () -> Throwable): Mono<T> {
        return this.hasElement()
                .flatMap { if (it) this else Mono.error(onError()) }
}

inline fun <T> Mono<T>.errorIfNotEmpty(crossinline onError: (T) -> Throwable): Mono<T> {
    return this.hasElement()
            .flatMap { if (it) Mono.error(onError.invoke(this.block()!!)) else this }
}

switchIfEmpty的问题是它总是计算传递给参数的表达式--编写这样的代码总是会产生Foo对象：

mono.switchIfEmpty(Foo())

您可以编写自己的扩展来延迟计算表达式，并传入参数：

inline fun <T> Mono<T>.switchIfEmpty(crossinline default: () -> Mono<T>): Mono<T> {
    return this.hasElement()
            .flatMap { if (it) this else default() }
}

下面是另外两个扩展函数-您可以使用它们来检查密码是否正确：

inline fun <T> Mono<T>.errorIf(crossinline predicate: (T) -> Boolean, crossinline throwable: (T) -> Throwable): Mono<T> {
    return this.flatMap { if (predicate(it)) Mono.error(throwable(it)) else Mono.just(it) }
}

inline fun <T> Mono<T>.errorIfNot(crossinline predicate: (T) -> Boolean, crossinline throwable: (T) -> Throwable): Mono<T> {
    return this.errorIf(predicate = { !predicate(it) }, throwable = throwable)
}