@Override
protected Future<OperationResult<Id>> processCreateDevice(final DeviceKey key, final Device device, final Span span) {

    final DeviceInformation value = new DeviceInformation();
    value.setTenantId(key.getTenantId());
    value.setDeviceId(key.getDeviceId());
    value.setRegistrationInformation(Json.encode(device));

    final CompletableFuture<OperationResult<Id>> f = this.managementCache
            .withFlags(Flag.FORCE_RETURN_VALUE)
            .putIfAbsentAsync(deviceKey(key), value)
            .thenApply(result -> {
                if (result == null) {
                    return OperationResult.ok(HTTP_CREATED,
                            Id.of(key.getDeviceId()),
                            Optional.empty(),
                            Optional.of(value.getVersion()));
                } else {
                    return OperationResult.empty(HTTP_CONFLICT);
                }
            });

    return MoreFutures.map(f);

}
 

private void lazyInit(KeycloakSession session) {
    if (codeCache == null) {
        synchronized (this) {
            if (codeCache == null) {
                InfinispanConnectionProvider connections = session.getProvider(InfinispanConnectionProvider.class);
                Cache cache = connections.getCache(InfinispanConnectionProvider.ACTION_TOKEN_CACHE);

                RemoteCache remoteCache = InfinispanUtil.getRemoteCache(cache);

                if (remoteCache != null) {
                    LOG.debugf("Having remote stores. Using remote cache '%s' for single-use cache of code", remoteCache.getName());
                    this.codeCache = () -> {
                        // Doing this way as flag is per invocation
                        return remoteCache.withFlags(Flag.FORCE_RETURN_VALUE);
                    };
                } else {
                    LOG.debugf("Not having remote stores. Using normal cache '%s' for single-use cache of code", cache.getName());
                    this.codeCache = () -> {
                        return cache;
                    };
                }
            }
        }
    }
}
 

private void lazyInit(KeycloakSession session) {
    if (tokenCache == null) {
        synchronized (this) {
            if (tokenCache == null) {
                InfinispanConnectionProvider connections = session.getProvider(InfinispanConnectionProvider.class);
                Cache cache = connections.getCache(InfinispanConnectionProvider.ACTION_TOKEN_CACHE);

                RemoteCache remoteCache = InfinispanUtil.getRemoteCache(cache);

                if (remoteCache != null) {
                    LOG.debugf("Having remote stores. Using remote cache '%s' for single-use cache of token", remoteCache.getName());
                    this.tokenCache = () -> {
                        // Doing this way as flag is per invocation
                        return remoteCache.withFlags(Flag.FORCE_RETURN_VALUE);
                    };
                } else {
                    LOG.debugf("Not having remote stores. Using normal cache '%s' for single-use cache of token", cache.getName());
                    this.tokenCache = () -> {
                        return cache;
                    };
                }
            }
        }
    }
}
 

@Override
protected org.infinispan.client.hotrod.RemoteCache<Object, Object> givenAConnectedCache() {
    final Configuration configuration = mock(Configuration.class);
    @SuppressWarnings("unchecked")
    final org.infinispan.client.hotrod.RemoteCache<Object, Object> result = mock(org.infinispan.client.hotrod.RemoteCache.class);
    when(remoteCacheManager.getCache(anyString(), anyBoolean())).thenReturn(result);
    when(remoteCacheManager.getConfiguration()).thenReturn(configuration);
    when(configuration.forceReturnValues()).thenReturn(false);
    when(result.withFlags(Flag.FORCE_RETURN_VALUE)).thenReturn(result);
    return result;
}
 

private <K, V extends SessionEntity> void runOnRemoteCache(TopologyInfo topology, RemoteCache<K, SessionEntityWrapper<V>> remoteCache, long maxIdleMs, K key, SessionUpdateTask<V> task, SessionEntityWrapper<V> sessionWrapper) {
    final V session = sessionWrapper.getEntity();
    SessionUpdateTask.CacheOperation operation = task.getOperation(session);

    switch (operation) {
        case REMOVE:
            remoteCache.remove(key);
            break;
        case ADD:
            remoteCache.put(key, sessionWrapper.forTransport(), task.getLifespanMs(), TimeUnit.MILLISECONDS, maxIdleMs, TimeUnit.MILLISECONDS);
            break;
        case ADD_IF_ABSENT:
            SessionEntityWrapper<V> existing = remoteCache
                    .withFlags(Flag.FORCE_RETURN_VALUE)
                    .putIfAbsent(key, sessionWrapper.forTransport(), -1, TimeUnit.MILLISECONDS, maxIdleMs, TimeUnit.MILLISECONDS);
            if (existing != null) {
                logger.debugf("Existing entity in remote cache for key: %s . Will update it", key);

                replace(topology, remoteCache, task.getLifespanMs(), maxIdleMs, key, task);
            }
            break;
        case REPLACE:
            replace(topology, remoteCache, task.getLifespanMs(), maxIdleMs, key, task);
            break;
        default:
            throw new IllegalStateException("Unsupported state " +  operation);
    }
}
 

public static int getClusterStartupTime(Cache<String, Integer> cache, String cacheKey, EntryInfo wrapper, int myThreadId) {
        Integer startupTime = myThreadId==1 ? Integer.parseInt(cacheKey.substring(4)) : Integer.parseInt(cacheKey.substring(4)) * 2;

        // Concurrency doesn't work correctly with this
        //Integer existingClusterStartTime = (Integer) cache.putIfAbsent(cacheKey, startupTime);

        // Concurrency works fine with this
        RemoteCache remoteCache = cache.getAdvancedCache().getComponentRegistry().getComponent(PersistenceManager.class).getStores(RemoteStore.class).iterator().next().getRemoteCache();

        Integer existingClusterStartTime = null;
        for (int i=0 ; i<10 ; i++) {
            try {
                existingClusterStartTime = (Integer) remoteCache.withFlags(Flag.FORCE_RETURN_VALUE).putIfAbsent(cacheKey, startupTime);
                break;
            } catch (HotRodClientException ce) {
                if (i == 9) {
                    throw ce;
                    //break;
                } else {
                    wrapper.exceptions.incrementAndGet();
                    System.err.println("Exception: i=" + i + " for key: " + cacheKey + " and myThreadId: " + myThreadId);
                }
            }
        }

        if (existingClusterStartTime == null
//                || startupTime.equals(remoteCache.get(cacheKey))
                ) {
            wrapper.successfulInitializations.incrementAndGet();
            return startupTime;
        } else {
            wrapper.failedInitializations.incrementAndGet();
            return existingClusterStartTime;
        }
    }
 

@Override
BasicCache<String, Serializable> getCache() {
    // Flags are per-invocation!
    return remoteCache.withFlags(Flag.FORCE_RETURN_VALUE);
}