org.keycloak.cluster.infinispan.InfinispanNotificationsManager クラス

org.keycloak.cluster.infinispan.InfinispanNotificationsManager 連携図

クラス
class	CacheEntryListener
class	HotRodListener

静的公開メンバ関数
static InfinispanNotificationsManager	create (KeycloakSession session, Cache< String, Serializable > workCache, String myAddress, String mySite, Set< RemoteStore > remoteStores)

限定公開メンバ関数
	InfinispanNotificationsManager (Cache< String, Serializable > workCache, RemoteCache workRemoteCache, String myAddress, String mySite, ExecutorService listenersExecutor)

静的限定公開変数類
static final Logger	logger = Logger.getLogger(InfinispanNotificationsManager.class)

関数
void	registerListener (String taskKey, ClusterListener task)
TaskCallback	registerTaskCallback (String taskKey, TaskCallback callback)
void	notify (String taskKey, ClusterEvent event, boolean ignoreSender, ClusterProvider.DCNotify dcNotify)
void	taskFinished (String taskKey, boolean success)

非公開メンバ関数
void	eventReceived (String key, Serializable obj)

非公開変数類
final ConcurrentMultivaluedHashMap< String, ClusterListener >	listeners = new ConcurrentMultivaluedHashMap<>()
final ConcurrentMap< String, TaskCallback >	taskCallbacks = new ConcurrentHashMap<>()
final Cache< String, Serializable >	workCache
final RemoteCache	workRemoteCache
final String	myAddress
final String	mySite
final ExecutorService	listenersExecutor

詳解

Impl for sending infinispan messages across cluster and listening to them

著者

Marek Posolda

構築子と解体子

InfinispanNotificationsManager()

org.keycloak.cluster.infinispan.InfinispanNotificationsManager.InfinispanNotificationsManager ( Cache< String, Serializable >  workCache, RemoteCache  workRemoteCache, String  myAddress, String  mySite, ExecutorService  listenersExecutor  )

inlineprotected

                                                                                                                                                                                     {
        this.workCache = workCache;
        this.workRemoteCache = workRemoteCache;
        this.myAddress = myAddress;
        this.mySite = mySite;
        this.listenersExecutor = listenersExecutor;
    }

関数詳解

create()

static InfinispanNotificationsManager org.keycloak.cluster.infinispan.InfinispanNotificationsManager.create ( KeycloakSession  session, Cache< String, Serializable >  workCache, String  myAddress, String  mySite, Set< RemoteStore >  remoteStores  )

inlinestatic

                                                                                                                                                                                        {
        RemoteCache workRemoteCache = null;

        if (!remoteStores.isEmpty()) {
            RemoteStore remoteStore = remoteStores.iterator().next();
            workRemoteCache = remoteStore.getRemoteCache();

            if (mySite == null) {
                throw new IllegalStateException("Multiple datacenters available, but site name is not configured! Check your configuration");
            }
        }

        ExecutorService listenersExecutor = workRemoteCache==null ? null : session.getProvider(ExecutorsProvider.class).getExecutor("work-cache-event-listener");
        InfinispanNotificationsManager manager = new InfinispanNotificationsManager(workCache, workRemoteCache, myAddress, mySite, listenersExecutor);

        // We need CacheEntryListener for communication within current DC
        workCache.addListener(manager.new CacheEntryListener());
        logger.debugf("Added listener for infinispan cache: %s", workCache.getName());

        // Added listener for remoteCache to notify other DCs
        if (workRemoteCache != null) {
            workRemoteCache.addClientListener(manager.new HotRodListener(workRemoteCache));
            logger.debugf("Added listener for HotRod remoteStore cache: %s", workRemoteCache.getName());
        }

        return manager;
    }

eventReceived()

void org.keycloak.cluster.infinispan.InfinispanNotificationsManager.eventReceived ( String  key, Serializable  obj  )

inlineprivate

                                                             {
        if (!(obj instanceof WrapperClusterEvent)) {
            if (obj == null) {
                logger.warnf("Event object wasn't available in remote cache after event was received. Event key: %s", key);
            }
            return;
        }

        WrapperClusterEvent event = (WrapperClusterEvent) obj;

        if (event.isIgnoreSender()) {
            if (this.myAddress.equals(event.getSender())) {
                return;
            }
        }

        if (event.isIgnoreSenderSite()) {
            if (this.mySite == null || this.mySite.equals(event.getSenderSite())) {
                return;
            }
        }

        String eventKey = event.getEventKey();

        if (logger.isTraceEnabled()) {
            logger.tracef("Received event: %s", event);
        }

        ClusterEvent wrappedEvent = event.getDelegateEvent();

        List<ClusterListener> myListeners = listeners.get(eventKey);
        if (myListeners != null) {
            for (ClusterListener listener : myListeners) {
                listener.eventReceived(wrappedEvent);
            }
        }
    }

notify()

void org.keycloak.cluster.infinispan.InfinispanNotificationsManager.notify ( String  taskKey, ClusterEvent  event, boolean  ignoreSender, ClusterProvider.DCNotify  dcNotify  )

inlinepackage

                                                                                                             {
        WrapperClusterEvent wrappedEvent = new WrapperClusterEvent();
        wrappedEvent.setEventKey(taskKey);
        wrappedEvent.setDelegateEvent(event);
        wrappedEvent.setIgnoreSender(ignoreSender);
        wrappedEvent.setIgnoreSenderSite(dcNotify == ClusterProvider.DCNotify.ALL_BUT_LOCAL_DC);
        wrappedEvent.setSender(myAddress);
        wrappedEvent.setSenderSite(mySite);

        String eventKey = UUID.randomUUID().toString();

        if (logger.isTraceEnabled()) {
            logger.tracef("Sending event with key %s: %s", eventKey, event);
        }

        if (dcNotify == ClusterProvider.DCNotify.LOCAL_DC_ONLY || workRemoteCache == null) {
            // Just put it to workCache, but skip notifying remoteCache
            workCache.getAdvancedCache().withFlags(Flag.IGNORE_RETURN_VALUES, Flag.SKIP_CACHE_STORE)
                    .put(eventKey, wrappedEvent, 120, TimeUnit.SECONDS);
        } else {
            // Add directly to remoteCache. Will notify remote listeners on all nodes in all DCs
            Retry.executeWithBackoff((int iteration) -> {
                try {
                    workRemoteCache.put(eventKey, wrappedEvent, 120, TimeUnit.SECONDS);
                } catch (HotRodClientException re) {
                if (logger.isDebugEnabled()) {
                    logger.debugf(re, "Failed sending notification to remote cache '%s'. Key: '%s', iteration '%s'. Will try to retry the task",
                            workRemoteCache.getName(), eventKey, iteration);
                }

                // Rethrow the exception. Retry will take care of handle the exception and eventually retry the operation.
                throw re;
            }

        }, 10, 10);

        }
    }

registerListener()

void org.keycloak.cluster.infinispan.InfinispanNotificationsManager.registerListener ( String  taskKey, ClusterListener  task  )

inlinepackage

                                                                {
        listeners.add(taskKey, task);
    }

registerTaskCallback()

TaskCallback org.keycloak.cluster.infinispan.InfinispanNotificationsManager.registerTaskCallback ( String  taskKey, TaskCallback  callback  )

inlinepackage

                                                                             {
        TaskCallback existing = taskCallbacks.putIfAbsent(taskKey, callback);

        if (existing != null) {
            return existing;
        } else {
            return callback;
        }
    }

taskFinished()

void org.keycloak.cluster.infinispan.InfinispanNotificationsManager.taskFinished ( String  taskKey, boolean  success  )

inlinepackage

                                                       {
        TaskCallback callback = taskCallbacks.remove(taskKey);

        if (callback != null) {
            if (logger.isDebugEnabled()) {
                logger.debugf("Finished task '%s' with '%b'", taskKey, success);
            }
            callback.setSuccess(success);
            callback.getTaskCompletedLatch().countDown();
        }

    }

メンバ詳解

listeners

final ConcurrentMultivaluedHashMap<String, ClusterListener> org.keycloak.cluster.infinispan.InfinispanNotificationsManager.listeners = new ConcurrentMultivaluedHashMap<>()

private

listenersExecutor

final ExecutorService org.keycloak.cluster.infinispan.InfinispanNotificationsManager.listenersExecutor

private

logger

final Logger org.keycloak.cluster.infinispan.InfinispanNotificationsManager.logger = Logger.getLogger(InfinispanNotificationsManager.class)

staticprotected

myAddress

final String org.keycloak.cluster.infinispan.InfinispanNotificationsManager.myAddress

private

mySite

final String org.keycloak.cluster.infinispan.InfinispanNotificationsManager.mySite

private

taskCallbacks

final ConcurrentMap<String, TaskCallback> org.keycloak.cluster.infinispan.InfinispanNotificationsManager.taskCallbacks = new ConcurrentHashMap<>()

private

workCache

final Cache<String, Serializable> org.keycloak.cluster.infinispan.InfinispanNotificationsManager.workCache

private

workRemoteCache

final RemoteCache org.keycloak.cluster.infinispan.InfinispanNotificationsManager.workRemoteCache

private

---

このクラス詳解は次のファイルから抽出されました:

• D:/AppData/doxygen/keycloak/src/keycloak/src/main/java/org/keycloak/cluster/infinispan/InfinispanNotificationsManager.java