Quarkus - 反应上下文中 ServerRequestFilter 中的 Keycloak 客户端

Question

我将 Quarkus 与 quarkus-resteasy-reactive 和 quarkus-keycloak-admin-client-reactive 扩展一起使用。我正在构建一个 ServerRequestFilter 来查询 keycloak 的用户属性，但即使我使用的是反应式客户端，我在调用它的方法时也会得到 BlockingNotAllowedException。

这是过滤器：

public class Filters {
    @Inject
    Keycloak keycloak;

    @ServerRequestFilter
    public Uni<Response> filter(ContainerRequestContext requestContext) {
        return Uni.createFrom().item(() -> 
                keycloak.realm("my-realm")
                        .users()
                        .search("user-that-i-get-from-context")
                        .stream()
                        .findFirst()
                        .orElseThrow()
                        .firstAttribute("the-attribute")).map(attr -> {
                            if (attr.equals("some-value")){
                                return null;
                            }
                            return Response.status(403).build();
                        });
    }
}

我尝试将 runSubscriptionOn(Infrastructure.getDefaultWorkerPool()) 添加到 Uni，但错误仍然存​​在。

使请求阻塞不是一种选择，因为我希望我的端点是反应性的。

我怀疑问题出在过滤器的上下文和Keycloak 客户端的注入中，但我一直无法查明问题所在

Answer 1

不幸的是，KeycloakClient 使用经典的 ResteasyClient 阻塞而不是 ResteasyReactiveClient。

我像这样使用 Vertx 解决了它：

import io.vertx.mutiny.core.Vertx;

public class Filters {

    @Inject Keycloak keycloak;
    @Inject Vertx vertx;

    @ServerRequestFilter
    public Uni<Response> filter(ContainerRequestContext requestContext) {
        return vertx.getOrCreateContext().executeBlocking(
            Uni.createFrom().emitter(emitter -> {
                var attr = keycloak.realm("my-realm")
                        .users()
                        .search("user-that-i-get-from-context")
                        .stream()
                        .findFirst()
                        .orElseThrow()
                        .firstAttribute("the-attribute");

                if ("some-value".equals(attr)) {
                    emitter.complete(null);
                } else {
                    emitter.complete(Response.status(403).build());
                }
            }));
    }
}

Answer 2

问题是，尽管quarkus-keycloak-admin-client-reactive 在引擎盖下使用了 RESTEasy Reactive 并执行非阻塞 API，但 Keycloak API 并未公开 Mutiny API 变体。 你试图用 Mutiny 做的事情是行不通的。

当前唯一的解决方案是通过返回 Response 而不是 Uni<Response> 使您的过滤器成为阻塞过滤器

Answer 3

@Jean Merelis 的回答有效，感谢他让我走上了正确的道路。只想补充一点，我也发现 runSubscriptionOn(Infrastructure.getDefaultWorkerPool()) 也有效，但前提是我执行 Uni.createFrom().item() 的供应商变体，否则 mutiny 会将内部项目视为已完成的 uni，并且不会将上下文从事件循环切换到执行程序线程。当我测试它时，我使用的是非供应商变体：

Uni.createFrom().item(keycloak.realm("my-realm")...)
                .runSubscriptionOn(Infrastructure.getDefaultWorkerPool())

所以我在问题中发布的代码将适用于.runSubscriptionOn(Infrastructure.getDefaultWorkerPool())

Answer 4

我有一个类似的问题，我使用 Uni.createFrom().completionStage(future) 解决了如下问题：

我的用例是将 Keycloak 用户资源和表示形式作为 Uni pipleine 的一部分，然后使用该用户数据在 pipleine 中进一步更新 Keycloak 用户。

下面返回的自定义 KeycloakUser 类只是存储资源和表示供以后使用。

获取 Keycloak 用户数据的简化命令式函数：

private KeycloakUser doGetKeycloakUser(String username){

  // Get our realm resource
  realmResource = keycloak.realm("quarkus");

  // Get the Users
  UsersResource usersResource = realmResource.users();

  // Get the specific user by username
  List<UserRepresentation> user = usersResource.search(username);

  // Get that User's representation
  UserRepresentation userRepresentation = user.get(0);

  // Get the user resource via the user id
  UserResource userResource = usersResource.get(userRepresentation.getId());

  // To update a Keycloak user we need both the representation and the resource.
  // My KeycloakUser class just stores them for later use downstream in the Uni pipeline.
  KeycloakUser kcUser = new KeycloakUser(userRepresentation, userResource);

  return kcUser;
}

然后我将该函数称为我的 Uni 链的一部分，如下所示：

  public Uni<KeycloakUser> getKeycloakUser(String username){

    // FYI: about this block:
    // While it is true you are using the quarkus-keycloak-admin-client-reactive
    // ie a *reactive* client,
    // the org.keycloak.admin.client.Keycloak itself is not reactive.
    // So, when you call it as part of a Uni pipeline,
    // it blocks the IO thread and throws a BlockingNotAllowedException.
    // See: https://github.com/quarkusio/quarkus/issues/29786
    // So, here we wrap the blocking function doGetKeycloakUser() in a CompletableFuture
    // and run it as an async operation, then when it completes we
    // turn that completionStage into a Uni.

    CompletableFuture<KeycloakUser> future = CompletableFuture.supplyAsync(() -> {
      return doGetKeycloakUser(username);
    });

    return Uni.createFrom().completionStage(future);
  }

也许不是最好的解决方案，但它确实有效。

在你的测试中，你需要使用awaitItem()而不是assertCompleted()：

Uni<String> uni = adminClient.getKeycloakUser(username)
  .chain(kcUser -> adminClient.setAttributeToUser(kcUser, "mykey", "myvalue"))
  .onItem()
  . ...

UniAssertSubscriber<String> subscriber = uni.subscribe().withSubscriber(UniAssertSubscriber.create());

subscriber.awaitItem().assertItem("Was set");