AtomicReference for Multi Stream

[stefanwendelmann]

Hey together 😊,

i am new to reactive programming and new to mutiny so pls be gentle.

First i would like to thank the mutiny team for the good work and @cescoffier for the good presentations on yt and so on.

I am working on a API build on Quarkus (2.7.1) where i strictly go the reactive route.

What i need to do is to write a scheduled method wich reads all UserSettings for the DB and check Keycloak if there was any change to the Users Email, then write the Email back to the mongo.

Since there is no Support for co-existing quarkus-keycloak-admin-client and quarkus-rest-client-reactive (see quarkusio/quarkus#20505 ) i needed to use the bare Keycloak Admin REST API with the quarkus-rest-client-reactive.

My Problem is, that i first need to get a Service Access Token (SAT) and then use this token for all requests on the User Details.
Problem: But the token gets recreated for each request, which is imho not that performat.

My approach was to use .onFailure(). if the token is not valid, recreate it and try it again, because the failure of the 401 lead to it.

My Code:

SettingsService.java

@ApplicationScoped
public class SettingsService
{
  private static final Logger LOG = Logger.getLogger(SettingsService.class);

  @Inject
  ReactiveMongoClient mongoClient;

  @Inject
  KeycloakAdminService keycloakAdminService;

  private ReactiveMongoCollection<UserSetting> getCollection()
  {
    return mongoClient
      .getDatabase("notification")
      .getCollection("userSettings", UserSetting.class);
  }

  public Multi<UserSetting> getAll(){
    return getCollection().find();
  }

  @Scheduled(every = "1h")
  public void updateUserInfosFromKeycloak(){
    getAll()
      .onItem().transformToUni( userSetting ->
        Uni.createFrom().item(userSetting.getKeycloakId())
          .onItem().transformToUni(uid -> keycloakAdminService.getUserInfo(uid))
      ).merge().onItem().transformToUni(userRepresentation -> getCollection().updateOne(
         Filters.eq("_id",userRepresentation.getId()),
         Updates.set("email",userRepresentation.getEmail())))
      .merge().
      subscribe().with(LOG::debug);
  }
}

KEycloakAdminService.java

@ApplicationScoped
public class KeycloakAdminService
{
  private static final Logger LOG = Logger.getLogger(KeycloakAdminService.class);

  @ConfigProperty(name = "keycloak.realm")
  String realm;
  @ConfigProperty(name = "keycloak.client")
  String client;
  @ConfigProperty(name = "keycloak.client.secret")
  String clientSecret;

  @RestClient
  KeycloakAdminRestClient keycloakAdminRestClient;

  private static AtomicReference<String> token = new AtomicReference<>();

  private Uni<String> login()
  {
    return Uni.createFrom().voidItem()
      .onItem().transformToUni(unused ->
        keycloakAdminRestClient.getSAT(realm, client, clientSecret, "client_credentials"))
      .onItem().invoke(() -> LOG.debug("Create a fresh Keycloak SAT"))
      .onItem().transform(tkn ->
      {
        token.set("Bearer " + tkn.getAccess_token());
        return token.get();
      });
  }

  public Uni<UserRepresentation> getUserInfo(String id)
  {
    // TODO: every call Refrehes the Token
    return keycloakAdminRestClient.getUser(token.get(), realm, id)
      .onFailure().recoverWithUni(() ->
        login().onItem().transformToUni(refreshedToken ->
        {
          LOG.debug("Refreshed the Token");
          return keycloakAdminRestClient.getUser(refreshedToken, realm, id);
        })
      );
  }
}

[jponge]

Does your code work? (you did not mention that).

Quick remark:

.onItem().transformToUni( userSetting ->
        Uni.createFrom().item(userSetting.getKeycloakId())
          .onItem().transformToUni(uid -> keycloakAdminService.getUserInfo(uid))

Why do you create an intermediate Uni? You should be able to do something more direct like:

.onItem().transformToUni(userSettings -> keycloakAdminService.getUserInfo(userSettings.getKeycloakId())

Note that I have no idea if the Keycloack APIs you are using are blocking or not. If they are blocking you might use emitOn/runSubscriptionOn to offload and not block.

[stefanwendelmann]

Hey thanks for your anwser.

Thanks for the hint with the intermediate Uni i will change that.
Bad habits.

And yes it works, but is calling the login() method for every user. (there should be one Access Token used by all iterations, and if its not valid anymore reCreate it with login()).

Nothing specific to the KeycloakAPI, its just a RestEASY Reactive Client returning a Uni<T>

KeycloakAdminRestClient.java

@Produces(MediaType.APPLICATION_JSON)
@Consumes(MediaType.APPLICATION_JSON)
@RegisterRestClient(configKey = "keycloak-api")
public interface KeycloakAdminRestClient
{

  @POST
  @Path("/realms/{realm}/protocol/openid-connect/token")
  @Consumes(MediaType.APPLICATION_FORM_URLENCODED)
  public Uni<SimpleTokenRepresentation> getUAT(@PathParam("realm") String realm,
                                               @FormParam("username") String username,
                                               @FormParam("password") String password,
                                               @FormParam("client_id") String clientId,
                                               @FormParam("client_secret") String clientSecret,
                                               @FormParam("grant_type") String grantType
  );

  @POST
  @Path("/realms/{realm}/protocol/openid-connect/token")
  @Consumes(MediaType.APPLICATION_FORM_URLENCODED)
  public Uni<SimpleTokenRepresentation> getSAT(@PathParam("realm") String realm,
                                               @FormParam("client_id") String clientId,
                                               @FormParam("client_secret") String clientSecret,
                                               @FormParam("grant_type") String grantType
  );

  @GET
  @Path("/admin/realms/{realm}/users/{id}")
  public Uni<UserRepresentation> getUser(@HeaderParam("Authorization") String auth,
                                         @PathParam("realm") String realm,
                                         @PathParam("id") String id);

}