Spring WebFlux：反应式 MongoDB

Question

我是 Spring Reactor 的新手，所以我想重构这个简单的 spring 数据（在 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);
}

没有中间变量和 cmets 的缩短版本：

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)
}