Axonframework，如何将 MessageDispatchInterceptor 与反应式存储库一起使用

Question

我已阅读基于集合的consistency validation blog，我想通过调度拦截器进行验证。我按照示例进行操作，但我使用反应式存储库，它对我来说并不真正有用。我已经尝试过阻止和不阻止。有块它会抛出错误，但没有块它不会执行任何东西。这是我的代码。

class SubnetCommandInterceptor : MessageDispatchInterceptor<CommandMessage<*>> {

  @Autowired
  private lateinit var privateNetworkRepository: PrivateNetworkRepository

  override fun handle(messages: List<CommandMessage<*>?>): BiFunction<Int, CommandMessage<*>, CommandMessage<*>> {
    return BiFunction<Int, CommandMessage<*>, CommandMessage<*>> { index: Int?, command: CommandMessage<*> ->
      if (CreateSubnetCommand::class.simpleName == (command.payloadType.simpleName)){
        val interceptCommand = command.payload as CreateSubnetCommand
        privateNetworkRepository
          .findById(interceptCommand.privateNetworkId)
          // ..some validation logic here ex.
          // .filter { network -> network.isSubnetOverlap() }
          .switchIfEmpty(Mono.error(IllegalArgumentException("Requested subnet is overlap with the previous subnet.")))
          // .block() also doesn't work here it throws error
         // block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-
      }
      command
    }
  }
}

Answer 1

实际上并不推荐订阅消息调度程序中的反应式存储库，并且可能会导致奇怪的行为，因为底层 ThreadLocal（由 Axox 使用）is not adapted to be used in reactive programing

相反，请查看 Axon's Reactive Extension 和反应式拦截器部分。

例如你可能会做什么：

reactiveCommandGateway.registerDispatchInterceptor(
        cmdMono -> cmdMono.flatMap(cmd->privateNetworkRepository
      .findById(cmd.privateNetworkId))
.switchIfEmpty(
Mono.error(IllegalArgumentException("Requested subnet is overlap with the previous subnet."))
.then(cmdMono)));