Java Project Reactor/WebFlux token service

Question

I'm trying to achieve a thread-safe reactive token service. The point is that all subscribers must wait until the token is received or updated (when expired). It works, but I want a second opinion about it's thread safety, as I'm not really sure how it should be done with reactive stack. The token lives 1 hour, but I update it 5000 ms before expiry.

@Service
@Slf4j
public class TokenService {
 // time before real token expires
 // token expiry time calculated like this:
 // time = expiry_time - UPDATE_TOKEN_BEFORE
 // thus the token will be updated a bit earlier than it expires
 private static final int UPDATE_TOKEN_BEFORE = 5000;
 private final WebClient webClient;
 private final TokenConfiguration config;
 @SuppressWarnings("FieldCanBeLocal")
 private volatile String token;
 private volatile long expiresMillis;
 private volatile boolean hasToken = false;
 private volatile boolean loading = false;
 private final Sinks.Many<String> sink;
 public TokenService(WebClient webClient, TokenConfiguration config) {
 this.webClient = webClient;
 this.config = config;
 this.sink = Sinks.unsafe().many().replay().latest();
 }
 /**
 * Get the cached token or request a new one from the auth service.
 * Makes subscribers wait until the first one gets the token from the auth service, then supplies them with
 * already retrieved token.
 *
 * @return {@link Mono} of token string.
 */
 public Mono<String> getToken() {
 if (!hasToken || System.currentTimeMillis() >= expiresMillis) {
 if (!loading) {
 // flag that we're processing the token
 this.loading = true;
 return getTokenMono()
 // update token & expiresMillis locally
 .map(this::updateToken)
 .map(token -> {
 final Sinks.EmitResult result = sink.tryEmitNext(token);
 // throw EmissionException if emitting fails
 result.orThrow();
 return token;
 })
 // finally, flag we're done processing
 .doOnSuccess(i -> this.loading = false)
 // if something goes wrong, flag that we should try again
 .doOnError(i -> {
 this.hasToken = false;
 this.loading = false;
 });
 }
 }
 // emit when the token gets ready
 return sink.asFlux().filter(token -> !token.isEmpty()).next();
 }
 /**
 * Invalidate the token.
 */
 public void invalidateToken() {
 this.hasToken = false;
 }
 /**
 * Retrieve the token from the auth service.
 *
 * @return Mono of {@link TokenResponse}.
 */
 private Mono<TokenResponse> getTokenMono() {
 log.info("Retrieving the access token from upstream");
 return webClient.post()
 .uri(config.getTokenUrl())
 .body(BodyInserters.fromFormData("grant_type", "credentials")
 .with("client_id", config.getClientId())
 .with("client_secret", config.getClientSecret()))
 .retrieve()
 .bodyToMono(TokenResponse.class);
 }
 /**
 * Update the token in this class and return it.
 *
 * @param response see {@link TokenResponse}
 * @return the token from {@code response}.
 */
 private String updateToken(TokenResponse response) {
 // make the token expire a few seconds before it is set to expire
 log.info("New token: {}", response);
 this.expiresMillis = (response.getExpiresIn() * 1000) + System.currentTimeMillis() - UPDATE_TOKEN_BEFORE;
 this.token = "Bearer " + response.getAccessToken();
 this.hasToken = true;
 return this.token;
 }
}