| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more contributor license |
| * agreements. See the NOTICE file distributed with this work for additional information regarding |
| * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance with the License. You may obtain a |
| * copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software distributed under the License |
| * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express |
| * or implied. See the License for the specific language governing permissions and limitations under |
| * the License. |
| */ |
| package org.apache.geode.cache.query.cq.internal; |
| |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentLinkedQueue; |
| |
| import org.apache.logging.log4j.Logger; |
| |
| import org.apache.geode.InvalidDeltaException; |
| import org.apache.geode.StatisticsFactory; |
| import org.apache.geode.SystemFailure; |
| import org.apache.geode.cache.Cache; |
| import org.apache.geode.cache.CacheEvent; |
| import org.apache.geode.cache.CacheLoaderException; |
| import org.apache.geode.cache.EntryEvent; |
| import org.apache.geode.cache.Operation; |
| import org.apache.geode.cache.RegionEvent; |
| import org.apache.geode.cache.TimeoutException; |
| import org.apache.geode.cache.client.Pool; |
| import org.apache.geode.cache.client.internal.GetEventValueOp; |
| import org.apache.geode.cache.client.internal.InternalPool; |
| import org.apache.geode.cache.client.internal.QueueManager; |
| import org.apache.geode.cache.client.internal.UserAttributes; |
| import org.apache.geode.cache.query.CqAttributes; |
| import org.apache.geode.cache.query.CqClosedException; |
| import org.apache.geode.cache.query.CqException; |
| import org.apache.geode.cache.query.CqExistsException; |
| import org.apache.geode.cache.query.CqListener; |
| import org.apache.geode.cache.query.CqQuery; |
| import org.apache.geode.cache.query.CqServiceStatistics; |
| import org.apache.geode.cache.query.CqStatusListener; |
| import org.apache.geode.cache.query.QueryException; |
| import org.apache.geode.cache.query.QueryInvalidException; |
| import org.apache.geode.cache.query.RegionNotFoundException; |
| import org.apache.geode.cache.query.SelectResults; |
| import org.apache.geode.cache.query.cq.internal.ops.ServerCQProxyImpl; |
| import org.apache.geode.cache.query.internal.CompiledSelect; |
| import org.apache.geode.cache.query.internal.CqQueryVsdStats; |
| import org.apache.geode.cache.query.internal.CqStateImpl; |
| import org.apache.geode.cache.query.internal.DefaultQuery; |
| import org.apache.geode.cache.query.internal.ExecutionContext; |
| import org.apache.geode.cache.query.internal.cq.ClientCQ; |
| import org.apache.geode.cache.query.internal.cq.CqService; |
| import org.apache.geode.cache.query.internal.cq.InternalCqQuery; |
| import org.apache.geode.cache.query.internal.cq.ServerCQ; |
| import org.apache.geode.distributed.internal.DistributionAdvisor.Profile; |
| import org.apache.geode.distributed.internal.DistributionConfig; |
| import org.apache.geode.internal.cache.CacheDistributionAdvisor.CacheProfile; |
| import org.apache.geode.internal.cache.EntryEventImpl; |
| import org.apache.geode.internal.cache.EventID; |
| import org.apache.geode.internal.cache.FilterProfile; |
| import org.apache.geode.internal.cache.FilterRoutingInfo; |
| import org.apache.geode.internal.cache.InternalCache; |
| import org.apache.geode.internal.cache.LocalRegion; |
| import org.apache.geode.internal.cache.tier.MessageType; |
| import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier; |
| import org.apache.geode.internal.cache.tier.sockets.CacheClientProxy; |
| import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID; |
| import org.apache.geode.internal.cache.tier.sockets.Part; |
| import org.apache.geode.internal.logging.LogService; |
| import org.apache.geode.internal.util.JavaWorkarounds; |
| |
| /** |
| * Implements the CqService functionality. |
| * |
| * @since GemFire 5.5 |
| */ |
| public class CqServiceImpl implements CqService { |
| private static final Logger logger = LogService.getLogger(); |
| |
| private static final Integer MESSAGE_TYPE_LOCAL_CREATE = MessageType.LOCAL_CREATE; |
| private static final Integer MESSAGE_TYPE_LOCAL_UPDATE = MessageType.LOCAL_UPDATE; |
| private static final Integer MESSAGE_TYPE_LOCAL_DESTROY = MessageType.LOCAL_DESTROY; |
| private static final Integer MESSAGE_TYPE_EXCEPTION = MessageType.EXCEPTION; |
| |
| /** |
| * System property to evaluate the query even though the initial results are not required when cq |
| * is executed using the execute() method. |
| */ |
| public static boolean EXECUTE_QUERY_DURING_INIT = Boolean.valueOf(System |
| .getProperty(DistributionConfig.GEMFIRE_PREFIX + "cq.EXECUTE_QUERY_DURING_INIT", "true")); |
| |
| private static final String CQ_NAME_PREFIX = "GfCq"; |
| |
| private final InternalCache cache; |
| |
| /** |
| * Manages cq pools to determine if a status of connect or disconnect needs to be sent out |
| */ |
| private final HashMap<String, Boolean> cqPoolsConnected = new HashMap<>(); |
| |
| /** |
| * Manages CQ objects. uses serverCqName as key and CqQueryImpl as value |
| * |
| * GuardedBy cqQueryMapLock |
| */ |
| private volatile HashMap<String, CqQueryImpl> cqQueryMap = new HashMap<>(); |
| |
| private final Object cqQueryMapLock = new Object(); |
| |
| private volatile boolean isRunning = false; |
| |
| /** |
| * Used by client when multiuser-authentication is true. |
| */ |
| private final HashMap<String, UserAttributes> cqNameToUserAttributesMap = new HashMap<>(); |
| |
| // Map to manage the similar CQs (having same query - performance optimization). |
| // With query as key and Set of CQs as values. |
| private final ConcurrentHashMap matchingCqMap; |
| |
| // CQ Service statistics |
| private final CqServiceStatisticsImpl cqServiceStats; |
| private final CqServiceVsdStats stats; |
| |
| // CQ identifier, also used in auto generated CQ names |
| private volatile long cqId = 1; |
| |
| /* This is to manage region to CQs map, client side book keeping. */ |
| private HashMap<String, ArrayList<String>> baseRegionToCqNameMap = new HashMap<>(); |
| |
| /** |
| * Access and modification to the contents of this map do not necessarily need to be lock |
| * protected. This is just used to optimize construction of a server side cq name. Missing values |
| * in this cache will mean a look up for a specific proxy id and cq name will miss and reconstruct |
| * the string before adding it back to the cache |
| */ |
| private static final ConcurrentHashMap<String, ConcurrentHashMap<ClientProxyMembershipID, String>> serverCqNameCache = |
| new ConcurrentHashMap<>(); |
| |
| /** |
| * Constructor. |
| * |
| * @param cache The cache used for the service |
| */ |
| public CqServiceImpl(final InternalCache cache) { |
| if (cache == null) { |
| throw new IllegalStateException("cache is null"); |
| } |
| cache.getCancelCriterion().checkCancelInProgress(null); |
| |
| this.cache = cache; |
| |
| // Initialize the Map which maintains the matching cqs. |
| this.matchingCqMap = new ConcurrentHashMap<String, HashSet<String>>(); |
| |
| // Initialize the VSD statistics |
| StatisticsFactory factory = this.cache.getDistributedSystem(); |
| this.stats = new CqServiceVsdStats(factory); |
| this.cqServiceStats = new CqServiceStatisticsImpl(this); |
| } |
| |
| /** |
| * Returns the cache associated with the cqService. |
| */ |
| public Cache getCache() { |
| return this.cache; |
| } |
| |
| public InternalCache getInternalCache() { |
| return this.cache; |
| } |
| |
| public CqServiceVsdStats stats() { |
| return this.stats; |
| } |
| |
| @Override |
| public synchronized ClientCQ newCq(String cqName, String queryString, CqAttributes cqAttributes, |
| InternalPool pool, boolean isDurable) |
| throws QueryInvalidException, CqExistsException, CqException { |
| if (queryString == null) { |
| throw new IllegalArgumentException( |
| String.format("Null argument %s", "queryString")); |
| |
| } else if (cqAttributes == null) { |
| throw new IllegalArgumentException( |
| String.format("Null argument %s", "cqAttribute")); |
| } |
| |
| if (isServer()) { |
| throw new IllegalStateException( |
| "client side newCq() method invocation on server."); |
| } |
| |
| // Check if the given cq already exists. |
| if (cqName != null && isCqExists(cqName)) { |
| throw new CqExistsException( |
| String.format("CQ with the given name already exists. CqName : %s", |
| cqName)); |
| } |
| |
| |
| ServerCQProxyImpl serverProxy = pool == null ? null : new ServerCQProxyImpl(pool); |
| ClientCQImpl cQuery = |
| new ClientCQImpl(this, cqName, queryString, cqAttributes, serverProxy, isDurable); |
| cQuery.updateCqCreateStats(); |
| |
| // cQuery.initCq(); |
| |
| // Check if query is valid. |
| cQuery.validateCq(); |
| |
| // Add cq into meta region. |
| // Check if Name needs to be generated. |
| if (cqName == null) { |
| // in the case of cqname internally generated, the CqExistsException needs |
| // to be taken care internally. |
| while (true) { |
| cQuery.setName(generateCqName()); |
| try { |
| addToCqMap(cQuery); |
| } catch (CqExistsException ex) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Got CqExistsException while intializing cq : {} Error : {}", |
| cQuery.getName(), ex.getMessage()); |
| } |
| continue; |
| } |
| break; |
| } |
| } else { |
| addToCqMap(cQuery); |
| } |
| |
| this.addToBaseRegionToCqNameMap(cQuery.getBaseRegionName(), cQuery.getServerCqName()); |
| |
| return cQuery; |
| } |
| |
| /** |
| * Executes the given CqQuery, if the CqQuery for that name is not there it registers the one and |
| * executes. This is called on the Server. |
| * |
| * @param manageEmptyRegions whether to update the 6.1 emptyRegions map held in the CCN |
| * @param regionDataPolicy the data policy of the region associated with the query. This is only |
| * needed if manageEmptyRegions is true. |
| * @param emptyRegionsMap map of empty regions. |
| * @throws IllegalStateException if this is called at client side. |
| */ |
| @Override |
| public synchronized ServerCQ executeCq(String cqName, String queryString, int cqState, |
| ClientProxyMembershipID clientProxyId, CacheClientNotifier ccn, boolean isDurable, |
| boolean manageEmptyRegions, int regionDataPolicy, Map emptyRegionsMap) |
| throws CqException, RegionNotFoundException, CqClosedException { |
| if (!isServer()) { |
| throw new IllegalStateException( |
| String.format("Server side executeCq method is called on client. CqName : %s", |
| cqName)); |
| } |
| |
| String serverCqName = constructServerCqName(cqName, clientProxyId); |
| ServerCQImpl cQuery; |
| |
| // If this CQ is not yet registered in Server, register CQ. |
| if (!isCqExists(serverCqName)) { |
| cQuery = new ServerCQImpl(this, cqName, queryString, isDurable, |
| constructServerCqName(cqName, clientProxyId)); |
| |
| try { |
| cQuery.registerCq(clientProxyId, ccn, cqState); |
| if (manageEmptyRegions) { // new in 6.1 |
| if (emptyRegionsMap != null && emptyRegionsMap.containsKey(cQuery.getBaseRegionName())) { |
| regionDataPolicy = 0; |
| } |
| |
| CacheClientProxy proxy = getCacheClientProxy(clientProxyId, ccn); |
| ccn.updateMapOfEmptyRegions( |
| proxy.getRegionsWithEmptyDataPolicy(), |
| cQuery.getBaseRegionName(), regionDataPolicy); |
| } |
| } catch (CqException cqe) { |
| logger.info("Exception while registering CQ on server. CqName : {}", |
| cQuery.getName()); |
| throw cqe; |
| } |
| |
| } else { |
| cQuery = (ServerCQImpl) getCq(serverCqName); |
| resumeCQ(cqState, cQuery); |
| } |
| |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("Successfully created CQ on the server. CqName : {}", cQuery.getName()); |
| } |
| return cQuery; |
| } |
| |
| public CacheClientProxy getCacheClientProxy(ClientProxyMembershipID clientProxyId, |
| CacheClientNotifier ccn) throws CqException { |
| CacheClientProxy proxy = ccn.getClientProxy(clientProxyId, true); |
| if (proxy == null) { |
| throw new CqException("No Cache Client Proxy found while executing CQ."); |
| } |
| return proxy; |
| } |
| |
| @Override |
| public void resumeCQ(int cqState, ServerCQ cQuery) { |
| // Initialize the state of CQ. |
| if (((CqStateImpl) cQuery.getState()).getState() != cqState) { |
| cQuery.setCqState(cqState); |
| // addToCqEventKeysMap(cQuery); |
| // Send state change info to peers. |
| cQuery.getCqBaseRegion().getFilterProfile().setCqState(cQuery); |
| } |
| // If we are going to set the state to running, we need to check to see if it matches any other |
| // cq |
| if (cqState == CqStateImpl.RUNNING) { |
| // Add to the matchedCqMap. |
| addToMatchingCqMap((CqQueryImpl) cQuery); |
| } |
| } |
| |
| /** |
| * Adds the given CQ and cqQuery object into the CQ map. |
| */ |
| void addToCqMap(CqQueryImpl cq) throws CqExistsException, CqException { |
| // On server side cqName will be server side cqName. |
| String sCqName = cq.getServerCqName(); |
| if (logger.isDebugEnabled()) { |
| logger.debug("Adding to CQ Repository. CqName : {} ServerCqName : {}", cq.getName(), sCqName); |
| } |
| HashMap<String, CqQueryImpl> cqMap = cqQueryMap; |
| if (cqMap.containsKey(sCqName)) { |
| throw new CqExistsException( |
| String.format("A CQ with the given name %s already exists.", |
| sCqName)); |
| } |
| synchronized (cqQueryMapLock) { |
| HashMap<String, CqQueryImpl> tmpCqQueryMap = new HashMap<>(cqQueryMap); |
| try { |
| tmpCqQueryMap.put(sCqName, cq); |
| } catch (Exception ex) { |
| String errMsg = |
| "Failed to store Continuous Query in the repository. CqName: %s %s"; |
| Object[] errMsgArgs = new Object[] {sCqName, ex.getLocalizedMessage()}; |
| String s = String.format(errMsg, errMsgArgs); |
| logger.error(s); |
| throw new CqException(s, ex); |
| } |
| UserAttributes attributes = UserAttributes.userAttributes.get(); |
| if (attributes != null) { |
| this.cqNameToUserAttributesMap.put(cq.getName(), attributes); |
| } |
| cqQueryMap = tmpCqQueryMap; |
| } |
| } |
| |
| /** |
| * Removes given CQ from the cqMap.. |
| */ |
| void removeCq(String cqName) { |
| // On server side cqName will be server side cqName. |
| synchronized (cqQueryMapLock) { |
| HashMap<String, CqQueryImpl> tmpCqQueryMap = new HashMap<>(cqQueryMap); |
| tmpCqQueryMap.remove(cqName); |
| this.cqNameToUserAttributesMap.remove(cqName); |
| cqQueryMap = tmpCqQueryMap; |
| } |
| } |
| |
| @Override |
| public CqQuery getClientCqFromServer(ClientProxyMembershipID clientProxyId, String clientCqName) { |
| // On server side cqName will be server side cqName. |
| HashMap<String, CqQueryImpl> cqMap = cqQueryMap; |
| return cqMap.get(this.constructServerCqName(clientCqName, clientProxyId)); |
| } |
| |
| @Override |
| public InternalCqQuery getCq(String cqName) { |
| // On server side cqName will be server side cqName. |
| return cqQueryMap.get(cqName); |
| } |
| |
| @Override |
| public Collection<? extends InternalCqQuery> getAllCqs() { |
| return cqQueryMap.values(); |
| } |
| |
| @Override |
| public Collection<? extends InternalCqQuery> getAllCqs(final String regionName) |
| throws CqException { |
| if (regionName == null) { |
| throw new IllegalArgumentException( |
| String.format("Null argument %s", "regionName")); |
| } |
| |
| String[] cqNames; |
| |
| synchronized (this.baseRegionToCqNameMap) { |
| ArrayList<String> cqs = this.baseRegionToCqNameMap.get(regionName); |
| if (cqs == null) { |
| return null; |
| } |
| cqNames = new String[cqs.size()]; |
| cqs.toArray(cqNames); |
| } |
| |
| ArrayList<InternalCqQuery> cQueryList = new ArrayList<>(); |
| for (int cqCnt = 0; cqCnt < cqNames.length; cqCnt++) { |
| InternalCqQuery cq = getCq(cqNames[cqCnt]); |
| if (cq != null) { |
| cQueryList.add(cq); |
| } |
| } |
| |
| return cQueryList; |
| } |
| |
| @Override |
| public synchronized void executeAllClientCqs() throws CqException { |
| executeCqs(this.getAllCqs()); |
| } |
| |
| @Override |
| public synchronized void executeAllRegionCqs(final String regionName) throws CqException { |
| executeCqs(getAllCqs(regionName)); |
| } |
| |
| @Override |
| public synchronized void executeCqs(Collection<? extends InternalCqQuery> cqs) |
| throws CqException { |
| if (cqs == null) { |
| return; |
| } |
| String cqName = null; |
| for (InternalCqQuery internalCq : cqs) { |
| CqQuery cq = internalCq; |
| if (!cq.isClosed() && cq.isStopped()) { |
| try { |
| cqName = cq.getName(); |
| cq.execute(); |
| } catch (QueryException | CqClosedException e) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Failed to execute the CQ, CqName : {} Error : {}", cqName, |
| e.getMessage()); |
| } |
| } |
| } |
| } |
| } |
| |
| @Override |
| public synchronized void stopAllClientCqs() throws CqException { |
| stopCqs(this.getAllCqs()); |
| } |
| |
| @Override |
| public synchronized void stopAllRegionCqs(final String regionName) throws CqException { |
| stopCqs(this.getAllCqs(regionName)); |
| } |
| |
| @Override |
| public synchronized void stopCqs(Collection<? extends InternalCqQuery> cqs) throws CqException { |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| if (isDebugEnabled) { |
| if (cqs == null) { |
| logger.debug("CqService.stopCqs cqs : null"); |
| } else { |
| logger.debug("CqService.stopCqs cqs : ({} queries)", cqs.size()); |
| } |
| } |
| |
| if (cqs == null) { |
| return; |
| } |
| |
| String cqName = null; |
| for (InternalCqQuery internalCqQuery : cqs) { |
| CqQuery cq = internalCqQuery; |
| if (!cq.isClosed() && cq.isRunning()) { |
| try { |
| cqName = cq.getName(); |
| cq.stop(); |
| } catch (QueryException | CqClosedException e) { |
| if (isDebugEnabled) { |
| logger.debug("Failed to stop the CQ, CqName : {} Error : {}", cqName, e.getMessage()); |
| } |
| } |
| } |
| } |
| } |
| |
| @Override |
| public void closeCqs(final String regionName) throws CqException { |
| Collection<? extends InternalCqQuery> cqs = this.getAllCqs(regionName); |
| if (cqs != null) { |
| String cqName = null; |
| for (InternalCqQuery cq : cqs) { |
| try { |
| cqName = cq.getName(); |
| |
| if (isServer()) { |
| // invoked on the server |
| cq.close(false); |
| } else { |
| // TODO: grid: if regionName has a pool check its keepAlive |
| boolean keepAlive = this.cache.keepDurableSubscriptionsAlive(); |
| if (cq.isDurable() && keepAlive) { |
| logger.warn("Not sending CQ close to the server as it is a durable CQ"); |
| cq.close(false); |
| } else { |
| cq.close(true); |
| } |
| } |
| |
| } catch (QueryException | CqClosedException e) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Failed to close the CQ, CqName : {} Error : {}", cqName, e.getMessage()); |
| } |
| } |
| } |
| } |
| } |
| |
| /** |
| * Called directly on server side. |
| */ |
| @Override |
| public void stopCq(String cqName, ClientProxyMembershipID clientId) throws CqException { |
| String serverCqName = cqName; |
| if (clientId != null) { |
| serverCqName = this.constructServerCqName(cqName, clientId); |
| removeFromCacheForServerToConstructedCQName(cqName, clientId); |
| } |
| |
| ServerCQImpl cQuery = null; |
| String errMsg = null; |
| Exception ex = null; |
| |
| try { |
| HashMap<String, CqQueryImpl> cqMap = cqQueryMap; |
| if (!cqMap.containsKey(serverCqName)) { |
| /* |
| * gregp 052808: We should silently fail here instead of throwing error. This is to deal |
| * with races in recovery |
| */ |
| return; |
| } |
| cQuery = (ServerCQImpl) getCq(serverCqName); |
| |
| } catch (CacheLoaderException e1) { |
| errMsg = "CQ not found in the cq meta region, CqName: %s"; |
| ex = e1; |
| } catch (TimeoutException e2) { |
| errMsg = "Timeout while trying to get CQ from meta region, CqName: %s"; |
| ex = e2; |
| } finally { |
| if (ex != null) { |
| String s = String.format(errMsg, cqName); |
| if (logger.isDebugEnabled()) { |
| logger.debug(s); |
| } |
| throw new CqException(s, ex); |
| } |
| } |
| |
| try { |
| if (!cQuery.isStopped()) { |
| cQuery.stop(); |
| } |
| } catch (CqClosedException cce) { |
| throw new CqException(cce.getMessage()); |
| } finally { |
| // If this CQ is stopped, disable caching event keys for this CQ. |
| // this.removeCQFromCaching(cQuery.getServerCqName()); |
| this.removeFromMatchingCqMap(cQuery); |
| } |
| // Send stop message to peers. |
| cQuery.getCqBaseRegion().getFilterProfile().stopCq(cQuery); |
| } |
| |
| @Override |
| public void closeCq(String cqName, ClientProxyMembershipID clientProxyId) throws CqException { |
| String serverCqName = cqName; |
| if (clientProxyId != null) { |
| serverCqName = this.constructServerCqName(cqName, clientProxyId); |
| removeFromCacheForServerToConstructedCQName(cqName, clientProxyId); |
| } |
| |
| ServerCQImpl cQuery = null; |
| String errMsg = null; |
| Exception ex = null; |
| |
| try { |
| HashMap<String, CqQueryImpl> cqMap = cqQueryMap; |
| if (!cqMap.containsKey(serverCqName)) { |
| /* |
| * gregp 052808: We should silently fail here instead of throwing error. This is to deal |
| * with races in recovery |
| */ |
| return; |
| } |
| cQuery = (ServerCQImpl) cqMap.get(serverCqName); |
| |
| } catch (CacheLoaderException e1) { |
| errMsg = "CQ not found in the cq meta region, CqName: %s"; |
| ex = e1; |
| } catch (TimeoutException e2) { |
| errMsg = "Timeout while trying to get CQ from meta region, CqName: %s"; |
| ex = e2; |
| } finally { |
| if (ex != null) { |
| String s = String.format(errMsg, cqName); |
| if (logger.isDebugEnabled()) { |
| logger.debug(s); |
| } |
| throw new CqException(s, ex); |
| } |
| } |
| |
| try { |
| cQuery.close(false); |
| |
| // Repository Region. |
| // If CQ event caching is enabled, remove this CQs event cache reference. |
| // removeCQFromCaching(serverCqName); |
| |
| // CqBaseRegion |
| try { |
| LocalRegion baseRegion = cQuery.getCqBaseRegion(); |
| if (baseRegion != null && !baseRegion.isDestroyed()) { |
| // Server specific clean up. |
| if (isServer()) { |
| FilterProfile fp = baseRegion.getFilterProfile(); |
| if (fp != null) { |
| fp.closeCq(cQuery); |
| } |
| CacheClientProxy clientProxy = |
| cQuery.getCacheClientNotifier().getClientProxy(clientProxyId); |
| clientProxy.decCqCount(); |
| if (clientProxy.hasNoCq()) { |
| this.stats.decClientsWithCqs(); |
| } |
| } |
| } |
| } catch (Exception e) { |
| // May be cache is being shutdown |
| if (logger.isDebugEnabled()) { |
| logger.debug("Failed to remove CQ from the base region. CqName : {}", cqName); |
| } |
| } |
| |
| if (isServer()) { |
| removeFromBaseRegionToCqNameMap(cQuery.getRegionName(), serverCqName); |
| } |
| |
| LocalRegion baseRegion = cQuery.getCqBaseRegion(); |
| if (baseRegion.getFilterProfile().getCqCount() <= 0) { |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "Should update the profile for this partitioned region {} for not requiring old value", |
| baseRegion); |
| } |
| } |
| } catch (CqClosedException cce) { |
| throw new CqException(cce.getMessage()); |
| } finally { |
| this.removeFromMatchingCqMap(cQuery); |
| } |
| } |
| |
| @Override |
| public void closeAllCqs(boolean clientInitiated) { |
| closeAllCqs(clientInitiated, getAllCqs()); |
| } |
| |
| /** |
| * Close all CQs executing in this VM, and release resources associated with executing CQs. |
| * CqQuerys created by other VMs are unaffected. |
| */ |
| private void closeAllCqs(boolean clientInitiated, Collection<? extends InternalCqQuery> cqs) { |
| closeAllCqs(clientInitiated, cqs, this.cache.keepDurableSubscriptionsAlive()); |
| } |
| |
| @Override |
| public void closeAllCqs(boolean clientInitiated, Collection<? extends InternalCqQuery> cqs, |
| boolean keepAlive) { |
| |
| if (cqs != null) { |
| String cqName = null; |
| if (logger.isDebugEnabled()) { |
| logger.debug("Closing all CQs, number of CQ to be closed : {}", cqs.size()); |
| } |
| for (InternalCqQuery cQuery : cqs) { |
| try { |
| cqName = cQuery.getName(); |
| |
| if (isServer()) { |
| cQuery.close(false); |
| } else { |
| if (clientInitiated) { |
| cQuery.close(true); |
| } else { |
| if (!isServer() && cQuery.isDurable() && keepAlive) { |
| logger.warn("Not sending CQ close to the server as it is a durable CQ"); |
| cQuery.close(false); |
| } else { |
| cQuery.close(true); |
| } |
| } |
| } |
| } catch (QueryException | CqClosedException e) { |
| if (!isRunning()) { |
| // Not cache shutdown |
| logger |
| .warn("Failed to close CQ %s %s", |
| cqName, e.getMessage()); |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug(e.getMessage(), e); |
| } |
| } |
| } |
| } |
| } |
| |
| @Override |
| public CqServiceStatistics getCqStatistics() { |
| return cqServiceStats; |
| } |
| |
| @Override |
| public void closeClientCqs(ClientProxyMembershipID clientProxyId) throws CqException { |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| if (isDebugEnabled) { |
| logger.debug("Closing Client CQs for the client: {}", clientProxyId); |
| } |
| List<ServerCQ> cqs = getAllClientCqs(clientProxyId); |
| for (ServerCQ cq : cqs) { |
| CqQueryImpl cQuery = (CqQueryImpl) cq; |
| try { |
| cQuery.close(false); |
| } catch (QueryException | CqClosedException e) { |
| if (isDebugEnabled) { |
| logger.debug("Failed to close the CQ, CqName : {} Error : {}", cQuery.getName(), |
| e.getMessage()); |
| } |
| } |
| } |
| } |
| |
| @Override |
| public List<ServerCQ> getAllClientCqs(ClientProxyMembershipID clientProxyId) { |
| Collection<? extends InternalCqQuery> cqs = getAllCqs(); |
| ArrayList<ServerCQ> clientCqs = new ArrayList<>(); |
| |
| for (InternalCqQuery cq : cqs) { |
| ServerCQImpl cQuery = (ServerCQImpl) cq; |
| ClientProxyMembershipID id = cQuery.getClientProxyId(); |
| if (id != null && id.equals(clientProxyId)) { |
| clientCqs.add(cQuery); |
| } |
| } |
| return clientCqs; |
| } |
| |
| @Override |
| public List<String> getAllDurableClientCqs(ClientProxyMembershipID clientProxyId) |
| throws CqException { |
| if (clientProxyId == null) { |
| throw new CqException( |
| String.format("Unable to retrieve durable CQs for client proxy id %s", clientProxyId)); |
| } |
| List<ServerCQ> cqs = getAllClientCqs(clientProxyId); |
| ArrayList<String> durableClientCqs = new ArrayList<>(); |
| |
| for (ServerCQ cq : cqs) { |
| ServerCQImpl cQuery = (ServerCQImpl) cq; |
| if (cQuery != null && cQuery.isDurable()) { |
| ClientProxyMembershipID id = cQuery.getClientProxyId(); |
| if (id != null && id.equals(clientProxyId)) { |
| durableClientCqs.add(cQuery.getName()); |
| } |
| } |
| } |
| return durableClientCqs; |
| } |
| |
| /** |
| * Server side method. Closes non-durable CQs for the given client proxy id. |
| */ |
| @Override |
| public void closeNonDurableClientCqs(ClientProxyMembershipID clientProxyId) throws CqException { |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| if (isDebugEnabled) { |
| logger.debug("Closing Client CQs for the client: {}", clientProxyId); |
| } |
| List<ServerCQ> cqs = getAllClientCqs(clientProxyId); |
| for (ServerCQ cq : cqs) { |
| ServerCQImpl cQuery = (ServerCQImpl) cq; |
| try { |
| if (!cQuery.isDurable()) { |
| cQuery.close(false); |
| } |
| } catch (QueryException | CqClosedException e) { |
| if (isDebugEnabled) { |
| logger.debug("Failed to close the CQ, CqName : {} Error : {}", cQuery.getName(), |
| e.getMessage()); |
| } |
| } |
| } |
| } |
| |
| /** |
| * Is the CQ service in a cache server environment |
| * |
| * @return true if cache server, false otherwise |
| */ |
| public boolean isServer() { |
| if (this.cache.getCacheServers().isEmpty()) { |
| return false; |
| } |
| return true; |
| } |
| |
| /** |
| * Cleans up the CqService. |
| */ |
| @Override |
| public void close() { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Closing CqService. {}", this); |
| } |
| // Close All the CQs. |
| // Need to take care when Clients are still connected... |
| closeAllCqs(false); |
| isRunning = false; |
| } |
| |
| @Override |
| public boolean isRunning() { |
| return this.isRunning; |
| } |
| |
| @Override |
| public void start() { |
| this.isRunning = true; |
| } |
| |
| /** |
| * @return Returns the serverCqName. |
| */ |
| @Override |
| public String constructServerCqName(String cqName, ClientProxyMembershipID clientProxyId) { |
| ConcurrentHashMap<ClientProxyMembershipID, String> cache = |
| JavaWorkarounds.computeIfAbsent(serverCqNameCache, cqName, |
| key -> new ConcurrentHashMap<>()); |
| |
| String cName = cache.get(clientProxyId); |
| if (null == cName) { |
| final StringBuilder sb = new StringBuilder(cqName).append("__"); |
| if (clientProxyId.isDurable()) { |
| sb.append(clientProxyId.getDurableId()); |
| } else { |
| sb.append(clientProxyId.getDSMembership()); |
| } |
| cName = sb.toString(); |
| cache.put(clientProxyId, cName); |
| } |
| |
| return cName; |
| } |
| |
| private void removeFromCacheForServerToConstructedCQName(final String cqName, |
| ClientProxyMembershipID clientProxyMembershipID) { |
| ConcurrentHashMap<ClientProxyMembershipID, String> cache = serverCqNameCache.get(cqName); |
| if (cache != null) { |
| cache.remove(clientProxyMembershipID); |
| if (cache.size() == 0) { |
| serverCqNameCache.remove(cqName); |
| } |
| } |
| } |
| |
| /** |
| * Checks if CQ with the given name already exists. |
| * |
| * @param cqName name of the CQ. |
| * |
| * @return true if exists else false. |
| */ |
| private synchronized boolean isCqExists(String cqName) { |
| HashMap<String, CqQueryImpl> cqMap = cqQueryMap; |
| return cqMap.containsKey(cqName); |
| } |
| |
| /** |
| * Generates a name for CQ. Checks if CQ with that name already exists if so generates a new |
| * cqName. |
| */ |
| private synchronized String generateCqName() { |
| while (true) { |
| String cqName = CQ_NAME_PREFIX + (cqId++); |
| if (!isCqExists(cqName)) { |
| return cqName; |
| } |
| } |
| } |
| |
| @Override |
| public void dispatchCqListeners(HashMap<String, Integer> cqs, int messageType, Object key, |
| Object value, byte[] delta, QueueManager qManager, EventID eventId) { |
| Object[] fullValue = new Object[1]; |
| Iterator<Map.Entry<String, Integer>> iter = cqs.entrySet().iterator(); |
| String cqName = null; |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| while (iter.hasNext()) { |
| try { |
| Map.Entry<String, Integer> entry = iter.next(); |
| cqName = entry.getKey(); |
| ClientCQImpl cQuery = (ClientCQImpl) this.getCq(cqName); |
| |
| if (cQuery == null || (!cQuery.isRunning() && cQuery.getQueuedEvents() == null)) { |
| if (isDebugEnabled) { |
| logger.debug("Unable to invoke CqListener, {}, CqName : {}", |
| ((cQuery == null) ? "CQ not found" : " CQ is Not running"), cqName); |
| } |
| continue; |
| } |
| |
| Integer cqOp = entry.getValue(); |
| |
| // If Region destroy event, close the cq. |
| if (cqOp.intValue() == MessageType.DESTROY_REGION) { |
| // The close will also invoke the listeners close(). |
| try { |
| cQuery.close(false); |
| } catch (Exception ex) { |
| // handle? |
| } |
| continue; |
| } |
| |
| // Construct CqEvent. |
| CqEventImpl cqEvent = new CqEventImpl(cQuery, getOperation(messageType), getOperation(cqOp), |
| key, value, delta, qManager, eventId); |
| |
| // Update statistics |
| cQuery.updateStats(cqEvent); |
| |
| // Check if CQ Event needs to be queued. |
| if (cQuery.getQueuedEvents() != null) { |
| synchronized (cQuery.queuedEventsSynchObject) { |
| // Get latest value. |
| ConcurrentLinkedQueue<CqEventImpl> queuedEvents = cQuery.getQueuedEvents(); |
| // Check to see, if its not set to null while waiting to get |
| // Synchronization lock. |
| if (queuedEvents != null) { |
| if (isDebugEnabled) { |
| logger.debug("Queueing event for key: {}", key); |
| } |
| cQuery.getVsdStats().incQueuedCqListenerEvents(); |
| queuedEvents.add(cqEvent); |
| continue; |
| } |
| } |
| } |
| |
| this.invokeListeners(cqName, cQuery, cqEvent, fullValue); |
| if (value == null) { |
| value = fullValue[0]; |
| } |
| |
| } // outer try |
| catch (Throwable t) { |
| logger.warn(String.format("Error processing CqListener for cq: %s", cqName), t); |
| |
| if (t instanceof VirtualMachineError) { |
| logger.warn(String.format("VirtualMachineError processing CqListener for cq: %s", |
| cqName), |
| t); |
| return; |
| } |
| } |
| } // iteration. |
| } |
| |
| void invokeListeners(String cqName, ClientCQImpl cQuery, CqEventImpl cqEvent) { |
| invokeListeners(cqName, cQuery, cqEvent, null); |
| } |
| |
| private void invokeListeners(String cqName, ClientCQImpl cQuery, CqEventImpl cqEvent, |
| Object[] fullValue) { |
| if (!cQuery.isRunning() || cQuery.getCqAttributes() == null) { |
| return; |
| } |
| // invoke CQ Listeners. |
| CqListener[] cqListeners = cQuery.getCqAttributes().getCqListeners(); |
| |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| if (isDebugEnabled) { |
| logger.debug("Invoking CQ listeners for {}, number of listeners : {} cqEvent : {}", cqName, |
| cqListeners.length, cqEvent); |
| } |
| |
| for (int lCnt = 0; lCnt < cqListeners.length; lCnt++) { |
| try { |
| // Check if the listener is not null, it could have been changed/reset |
| // by the CqAttributeMutator. |
| if (cqListeners[lCnt] != null) { |
| cQuery.getVsdStats().incNumCqListenerInvocations(); |
| try { |
| if (cqEvent.getThrowable() != null) { |
| cqListeners[lCnt].onError(cqEvent); |
| } else { |
| cqListeners[lCnt].onEvent(cqEvent); |
| } |
| } catch (InvalidDeltaException ide) { |
| if (isDebugEnabled) { |
| logger.debug("CqService.dispatchCqListeners(): Requesting full value..."); |
| } |
| Part result = (Part) GetEventValueOp |
| .executeOnPrimary(cqEvent.getQueueManager().getPool(), cqEvent.getEventID(), null); |
| Object newVal = result.getObject(); |
| if (result == null || newVal == null) { |
| if (!cache.getCancelCriterion().isCancelInProgress()) { |
| Exception ex = |
| new Exception("Failed to retrieve full value from server for eventID " |
| + cqEvent.getEventID()); |
| logger.warn("Exception in the CqListener of the CQ, CqName: {} Error : {}", |
| new Object[] {cqName, ex.getMessage()}); |
| if (isDebugEnabled) { |
| logger.debug(ex.getMessage(), ex); |
| } |
| } |
| } else { |
| this.cache.getCachePerfStats().incDeltaFullValuesRequested(); |
| cqEvent = new CqEventImpl(cQuery, cqEvent.getBaseOperation(), |
| cqEvent.getQueryOperation(), cqEvent.getKey(), newVal, cqEvent.getDeltaValue(), |
| cqEvent.getQueueManager(), cqEvent.getEventID()); |
| if (cqEvent.getThrowable() != null) { |
| cqListeners[lCnt].onError(cqEvent); |
| } else { |
| cqListeners[lCnt].onEvent(cqEvent); |
| } |
| if (fullValue != null) { |
| fullValue[0] = newVal; |
| } |
| } |
| } |
| } |
| // Handle client side exceptions. |
| } catch (Exception ex) { |
| if (!cache.getCancelCriterion().isCancelInProgress()) { |
| logger.warn("Exception in the CqListener of the CQ, CqName: {} Error : {}", |
| new Object[] {cqName, ex.getMessage()}); |
| if (isDebugEnabled) { |
| logger.debug(ex.getMessage(), ex); |
| } |
| } |
| } catch (VirtualMachineError err) { |
| SystemFailure.initiateFailure(err); |
| // If this ever returns, rethrow the error. We're poisoned |
| // now, so don't let this thread continue. |
| throw err; |
| } catch (Throwable t) { |
| // Whenever you catch Error or Throwable, you must also |
| // catch VirtualMachineError (see above). However, there is |
| // _still_ a possibility that you are dealing with a cascading |
| // error condition, so you also need to check to see if the JVM |
| // is still usable: |
| SystemFailure.checkFailure(); |
| logger.warn("Runtime Exception in the CqListener of the CQ, CqName: {} Error : {}", |
| new Object[] {cqName, t.getLocalizedMessage()}); |
| if (isDebugEnabled) { |
| logger.debug(t.getMessage(), t); |
| } |
| } |
| } |
| } |
| |
| private void invokeCqConnectedListeners(String cqName, ClientCQImpl cQuery, boolean connected) { |
| if (!cQuery.isRunning() || cQuery.getCqAttributes() == null) { |
| return; |
| } |
| cQuery.setConnected(connected); |
| // invoke CQ Listeners. |
| CqListener[] cqListeners = cQuery.getCqAttributes().getCqListeners(); |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("Invoking CQ status listeners for {}, number of listeners : {}", cqName, |
| cqListeners.length); |
| } |
| |
| for (int lCnt = 0; lCnt < cqListeners.length; lCnt++) { |
| try { |
| if (cqListeners[lCnt] != null) { |
| if (cqListeners[lCnt] instanceof CqStatusListener) { |
| CqStatusListener listener = (CqStatusListener) cqListeners[lCnt]; |
| if (connected) { |
| listener.onCqConnected(); |
| } else { |
| listener.onCqDisconnected(); |
| } |
| } |
| } |
| // Handle client side exceptions. |
| } catch (Exception ex) { |
| if (!cache.getCancelCriterion().isCancelInProgress()) { |
| logger.warn("Exception in the CqListener of the CQ, CqName: {} Error : {}", |
| new Object[] {cqName, ex.getMessage()}); |
| if (logger.isDebugEnabled()) { |
| logger.debug(ex.getMessage(), ex); |
| } |
| } |
| } catch (VirtualMachineError err) { |
| SystemFailure.initiateFailure(err); |
| // If this ever returns, rethrow the error. We're poisoned |
| // now, so don't let this thread continue. |
| throw err; |
| } catch (Throwable t) { |
| // Whenever you catch Error or Throwable, you must also |
| // catch VirtualMachineError (see above). However, there is |
| // _still_ a possibility that you are dealing with a cascading |
| // error condition, so you also need to check to see if the JVM |
| // is still usable: |
| SystemFailure.checkFailure(); |
| logger.warn("Runtime Exception in the CqListener of the CQ, CqName: {} Error : {}", |
| new Object[] {cqName, t.getLocalizedMessage()}); |
| if (logger.isDebugEnabled()) { |
| logger.debug(t.getMessage(), t); |
| } |
| } |
| } |
| } |
| |
| /** |
| * Returns the Operation for the given EnumListenerEvent type. |
| */ |
| private Operation getOperation(int eventType) { |
| Operation op = null; |
| switch (eventType) { |
| case MessageType.LOCAL_CREATE: |
| op = Operation.CREATE; |
| break; |
| |
| case MessageType.LOCAL_UPDATE: |
| op = Operation.UPDATE; |
| break; |
| |
| case MessageType.LOCAL_DESTROY: |
| op = Operation.DESTROY; |
| break; |
| |
| case MessageType.LOCAL_INVALIDATE: |
| op = Operation.INVALIDATE; |
| break; |
| |
| case MessageType.CLEAR_REGION: |
| op = Operation.REGION_CLEAR; |
| break; |
| |
| case MessageType.INVALIDATE_REGION: |
| op = Operation.REGION_INVALIDATE; |
| break; |
| } |
| return op; |
| } |
| |
| @Override |
| public void processEvents(CacheEvent event, Profile localProfile, Profile[] profiles, |
| FilterRoutingInfo frInfo) throws CqException { |
| // Is this a region event or an entry event |
| if (event instanceof RegionEvent) { |
| processRegionEvent(event, localProfile, profiles, frInfo); |
| } else { |
| // Use the PDX types in serialized form. |
| Boolean initialPdxReadSerialized = this.cache.getPdxReadSerializedOverride(); |
| this.cache.setPdxReadSerializedOverride(true); |
| try { |
| processEntryEvent(event, localProfile, profiles, frInfo); |
| } finally { |
| this.cache.setPdxReadSerializedOverride(initialPdxReadSerialized); |
| } |
| } |
| } |
| |
| private void processRegionEvent(CacheEvent event, Profile localProfile, Profile[] profiles, |
| FilterRoutingInfo frInfo) throws CqException { |
| |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| if (isDebugEnabled) { |
| logger.debug("CQ service processing region event {}", event); |
| } |
| Integer cqRegionEvent = generateCqRegionEvent(event); |
| |
| for (int i = -1; i < profiles.length; i++) { |
| CacheProfile cf; |
| if (i < 0) { |
| cf = (CacheProfile) localProfile; |
| if (cf == null) |
| continue; |
| } else { |
| cf = (CacheProfile) profiles[i]; |
| } |
| FilterProfile pf = cf.filterProfile; |
| if (pf == null || pf.getCqMap().isEmpty()) { |
| continue; |
| } |
| Map cqs = pf.getCqMap(); |
| HashMap<Long, Integer> cqInfo = new HashMap<>(); |
| Iterator cqIter = cqs.entrySet().iterator(); |
| while (cqIter.hasNext()) { |
| Map.Entry cqEntry = (Map.Entry) cqIter.next(); |
| ServerCQImpl cQuery = (ServerCQImpl) cqEntry.getValue(); |
| if (!event.isOriginRemote() && event.getOperation().isRegionDestroy() |
| && !((LocalRegion) event.getRegion()).isUsedForPartitionedRegionBucket()) { |
| try { |
| if (isDebugEnabled) { |
| logger.debug("Closing CQ on region destroy event. CqName : {}", cQuery.getName()); |
| } |
| cQuery.close(false); |
| } catch (Exception ex) { |
| if (isDebugEnabled) { |
| logger.debug("Failed to Close CQ on region destroy. CqName : {}", cQuery.getName(), |
| ex); |
| } |
| } |
| } |
| cqInfo.put(cQuery.getFilterID(), cqRegionEvent); |
| cQuery.getVsdStats().updateStats(cqRegionEvent); |
| } |
| if (pf.isLocalProfile()) { |
| frInfo.setLocalCqInfo(cqInfo); |
| } else { |
| frInfo.setCqRoutingInfo(cf.getDistributedMember(), cqInfo); |
| } |
| } |
| } |
| |
| private void processEntryEvent(CacheEvent event, Profile localProfile, Profile[] profiles, |
| FilterRoutingInfo frInfo) throws CqException { |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| HashSet<Object> cqUnfilteredEventsSet_newValue = new HashSet<>(); |
| HashSet<Object> cqUnfilteredEventsSet_oldValue = new HashSet<>(); |
| boolean b_cqResults_newValue; |
| boolean b_cqResults_oldValue; |
| boolean queryOldValue; |
| EntryEvent entryEvent = (EntryEvent) event; |
| Object eventKey = entryEvent.getKey(); |
| |
| boolean isDupEvent = ((EntryEventImpl) event).isPossibleDuplicate(); |
| // The CQ query needs to be applied when the op is update, destroy |
| // invalidate and in case when op is create and its an duplicate |
| // event, the reason for this is when peer sends a duplicate event |
| // it marks it as create and sends it, so that the receiving node |
| // applies it (see DR.virtualPut()). |
| boolean opRequiringQueryOnOldValue = (event.getOperation().isUpdate() |
| || event.getOperation().isDestroy() || event.getOperation().isInvalidate() |
| || (event.getOperation().isCreate() && isDupEvent)); |
| |
| HashMap<String, Integer> matchedCqs = new HashMap<>(); |
| long executionStartTime; |
| for (int i = -1; i < profiles.length; i++) { |
| CacheProfile cf; |
| if (i < 0) { |
| cf = (CacheProfile) localProfile; |
| if (cf == null) |
| continue; |
| } else { |
| cf = (CacheProfile) profiles[i]; |
| } |
| FilterProfile pf = cf.filterProfile; |
| if (pf == null || pf.getCqMap().isEmpty()) { |
| continue; |
| } |
| |
| Map cqs = pf.getCqMap(); |
| |
| if (isDebugEnabled) { |
| logger.debug("Profile for {} processing {} CQs", cf.peerMemberId, cqs.size()); |
| } |
| |
| if (cqs.isEmpty()) { |
| continue; |
| } |
| |
| // Get new value. If its not retrieved. |
| if (cqUnfilteredEventsSet_newValue.isEmpty() |
| && (event.getOperation().isCreate() || event.getOperation().isUpdate())) { |
| Object newValue = entryEvent.getNewValue(); |
| if (newValue != null) { |
| // We have a new value to run the query on |
| cqUnfilteredEventsSet_newValue.add(newValue); |
| } |
| } |
| |
| HashMap<Long, Integer> cqInfo = new HashMap<>(); |
| Iterator cqIter = cqs.entrySet().iterator(); |
| |
| while (cqIter.hasNext()) { |
| Map.Entry cqEntry = (Map.Entry) cqIter.next(); |
| ServerCQImpl cQuery = (ServerCQImpl) cqEntry.getValue(); |
| b_cqResults_newValue = false; |
| b_cqResults_oldValue = false; |
| queryOldValue = false; |
| if (cQuery == null) { |
| continue; |
| } |
| String cqName = cQuery.getServerCqName(); |
| Long filterID = cQuery.getFilterID(); |
| |
| if (isDebugEnabled) { |
| logger.debug("Processing CQ : {} Key: {}", cqName, eventKey); |
| } |
| |
| Integer cqEvent = null; |
| if (matchedCqs.containsKey(cqName)) { |
| cqEvent = matchedCqs.get(cqName); |
| if (isDebugEnabled) { |
| logger.debug("query {} has already been processed and returned {}", cqName, cqEvent); |
| } |
| if (cqEvent == null) { |
| continue; |
| } |
| // Update the Cache Results for this CQ. |
| if (cqEvent.intValue() == MessageType.LOCAL_CREATE |
| || cqEvent.intValue() == MessageType.LOCAL_UPDATE) { |
| cQuery.addToCqResultKeys(eventKey); |
| } else if (cqEvent.intValue() == MessageType.LOCAL_DESTROY) { |
| cQuery.markAsDestroyedInCqResultKeys(eventKey); |
| } |
| } else { |
| boolean error = false; |
| { |
| try { |
| synchronized (cQuery) { |
| // Apply query on new value. |
| if (!cqUnfilteredEventsSet_newValue.isEmpty()) { |
| executionStartTime = this.stats.startCqQueryExecution(); |
| |
| b_cqResults_newValue = |
| evaluateQuery(cQuery, new Object[] {cqUnfilteredEventsSet_newValue}); |
| this.stats.endCqQueryExecution(executionStartTime); |
| } |
| } |
| |
| // In case of Update, destroy and invalidate. |
| // Apply query on oldValue. |
| if (opRequiringQueryOnOldValue) { |
| // Check if CQ Result is cached, if not apply query on old |
| // value. Currently the CQ Results are not cached for the |
| // Partitioned Regions. Once this is added remove the check |
| // with PR region. |
| if (cQuery.cqResultKeysInitialized) { |
| b_cqResults_oldValue = cQuery.isPartOfCqResult(eventKey); |
| // For PR if not found in cache, apply the query on old value. |
| // Also apply if the query was not executed during cq execute |
| if ((cQuery.isPR || !CqServiceImpl.EXECUTE_QUERY_DURING_INIT) |
| && b_cqResults_oldValue == false) { |
| queryOldValue = true; |
| } |
| if (isDebugEnabled && !cQuery.isPR && !b_cqResults_oldValue) { |
| logger.debug( |
| "Event Key not found in the CQ Result Queue. EventKey : {} CQ Name : {}", |
| eventKey, cqName); |
| } |
| } else { |
| queryOldValue = true; |
| } |
| |
| if (queryOldValue) { |
| if (cqUnfilteredEventsSet_oldValue.isEmpty()) { |
| Object oldValue = entryEvent.getOldValue(); |
| if (oldValue != null) { |
| cqUnfilteredEventsSet_oldValue.add(oldValue); |
| } |
| } |
| |
| synchronized (cQuery) { |
| // Apply query on old value. |
| if (!cqUnfilteredEventsSet_oldValue.isEmpty()) { |
| executionStartTime = this.stats.startCqQueryExecution(); |
| b_cqResults_oldValue = |
| evaluateQuery(cQuery, new Object[] {cqUnfilteredEventsSet_oldValue}); |
| this.stats.endCqQueryExecution(executionStartTime); |
| } else { |
| if (isDebugEnabled) { |
| logger.debug( |
| "old value for event with key {} is null - query execution not performed", |
| eventKey); |
| } |
| } |
| } |
| } // Query oldValue |
| |
| } |
| } catch (Exception ex) { |
| // Any exception in running the query should be caught here and |
| // buried because this code is running in-line with the message |
| // processing code and we don't want to kill that thread |
| error = true; |
| // CHANGE LOG MESSAGE: |
| logger.info("Error while processing CQ on the event, key : {} CqName: {}, Error: {}", |
| new Object[] {((EntryEvent) event).getKey(), cQuery.getName(), |
| ex.getLocalizedMessage()}); |
| } |
| |
| if (error) { |
| cqEvent = MESSAGE_TYPE_EXCEPTION; |
| } else { |
| if (b_cqResults_newValue) { |
| if (b_cqResults_oldValue) { |
| cqEvent = MESSAGE_TYPE_LOCAL_UPDATE; |
| } else { |
| cqEvent = MESSAGE_TYPE_LOCAL_CREATE; |
| } |
| // If its create and caching is enabled, cache the key |
| // for this CQ. |
| cQuery.addToCqResultKeys(eventKey); |
| } else if (b_cqResults_oldValue) { |
| // Base invalidate operation is treated as destroy. |
| // When the invalidate comes through, the entry will no longer |
| // satisfy the query and will need to be deleted. |
| cqEvent = MESSAGE_TYPE_LOCAL_DESTROY; |
| // If caching is enabled, mark this event's key as removed |
| // from the CQ cache. |
| cQuery.markAsDestroyedInCqResultKeys(eventKey); |
| } |
| } |
| } |
| |
| // Get the matching CQs if any. |
| // synchronized (this.matchingCqMap){ |
| String query = cQuery.getQueryString(); |
| Set matchingCqs = (Set) matchingCqMap.get(query); |
| if (matchingCqs != null) { |
| Iterator iter = matchingCqs.iterator(); |
| while (iter.hasNext()) { |
| String matchingCqName = (String) iter.next(); |
| if (!matchingCqName.equals(cqName)) { |
| matchedCqs.put(matchingCqName, cqEvent); |
| if (isDebugEnabled) { |
| logger.debug("Adding CQ into Matching CQ Map: {} Event is: {}", matchingCqName, |
| cqEvent); |
| } |
| } |
| } |
| } |
| } |
| |
| if (cqEvent != null && cQuery.isRunning()) { |
| if (isDebugEnabled) { |
| logger.debug("Added event to CQ with client-side name: {} key: {} operation : {}", |
| cQuery.cqName, eventKey, cqEvent); |
| } |
| cqInfo.put(filterID, cqEvent); |
| CqQueryVsdStats stats = cQuery.getVsdStats(); |
| if (stats != null) { |
| stats.updateStats(cqEvent); |
| } |
| } |
| } |
| if (cqInfo.size() > 0) { |
| if (pf.isLocalProfile()) { |
| if (isDebugEnabled) { |
| logger.debug("Setting local CQ matches to {}", cqInfo); |
| } |
| frInfo.setLocalCqInfo(cqInfo); |
| } else { |
| if (isDebugEnabled) { |
| logger.debug("Setting CQ matches for {} to {}", cf.getDistributedMember(), cqInfo); |
| } |
| frInfo.setCqRoutingInfo(cf.getDistributedMember(), cqInfo); |
| } |
| } |
| } // iteration over Profiles. |
| } |
| |
| private Integer generateCqRegionEvent(CacheEvent event) { |
| Integer cqEvent = null; |
| if (event.getOperation().isRegionDestroy()) { |
| cqEvent = MessageType.DESTROY_REGION; |
| } else if (event.getOperation().isRegionInvalidate()) { |
| cqEvent = MessageType.INVALIDATE_REGION; |
| } else if (event.getOperation().isClear()) { |
| cqEvent = MessageType.CLEAR_REGION; |
| } |
| return cqEvent; |
| } |
| |
| /** |
| * Manages the CQs created for the base region. This is managed here, instead of on the base |
| * region; since the cq could be created on the base region, before base region is created (using |
| * newCq()). |
| */ |
| private void addToBaseRegionToCqNameMap(String regionName, String cqName) { |
| synchronized (this.baseRegionToCqNameMap) { |
| ArrayList<String> cqs = this.baseRegionToCqNameMap.get(regionName); |
| if (cqs == null) { |
| cqs = new ArrayList<>(); |
| } |
| cqs.add(cqName); |
| this.baseRegionToCqNameMap.put(regionName, cqs); |
| } |
| } |
| |
| void removeFromBaseRegionToCqNameMap(String regionName, String cqName) { |
| synchronized (this.baseRegionToCqNameMap) { |
| ArrayList<String> cqs = this.baseRegionToCqNameMap.get(regionName); |
| if (cqs != null) { |
| cqs.remove(cqName); |
| if (cqs.isEmpty()) { |
| this.baseRegionToCqNameMap.remove(regionName); |
| } else { |
| this.baseRegionToCqNameMap.put(regionName, cqs); |
| } |
| } |
| } |
| } |
| |
| /** |
| * Get the VSD ststs for CQ Service. There is one CQ Service per cache |
| * |
| * @return reference to VSD stats object for the CQ service |
| */ |
| public CqServiceVsdStats getCqServiceVsdStats() { |
| return stats; |
| } |
| |
| /** |
| * Adds the query from the given CQ to the matched CQ map. |
| */ |
| void addToMatchingCqMap(CqQueryImpl cq) { |
| synchronized (this.matchingCqMap) { |
| String cqQuery = cq.getQueryString(); |
| Set<String> matchingCQs; |
| if (!matchingCqMap.containsKey(cqQuery)) { |
| matchingCQs = Collections.newSetFromMap(new ConcurrentHashMap()); |
| matchingCqMap.put(cqQuery, matchingCQs); |
| this.stats.incUniqueCqQuery(); |
| } else { |
| matchingCQs = (Set) matchingCqMap.get(cqQuery); |
| } |
| matchingCQs.add(cq.getServerCqName()); |
| if (logger.isDebugEnabled()) { |
| logger.debug("Adding CQ into MatchingCQ map, CQName: {} Number of matched querys are: {}", |
| cq.getServerCqName(), matchingCQs.size()); |
| } |
| } |
| } |
| |
| /** |
| * Removes the query from the given CQ from the matched CQ map. |
| */ |
| private void removeFromMatchingCqMap(CqQueryImpl cq) { |
| synchronized (this.matchingCqMap) { |
| String cqQuery = cq.getQueryString(); |
| if (matchingCqMap.containsKey(cqQuery)) { |
| Set matchingCQs = (Set) matchingCqMap.get(cqQuery); |
| matchingCQs.remove(cq.getServerCqName()); |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "Removing CQ from MatchingCQ map, CQName: {} Number of matched querys are: {}", |
| cq.getServerCqName(), matchingCQs.size()); |
| } |
| if (matchingCQs.isEmpty()) { |
| matchingCqMap.remove(cqQuery); |
| this.stats.decUniqueCqQuery(); |
| } |
| } |
| } |
| } |
| |
| /** |
| * Returns the matching CQ map. |
| * |
| * @return HashMap matchingCqMap |
| */ |
| public Map<String, HashSet<String>> getMatchingCqMap() { |
| return matchingCqMap; |
| } |
| |
| /** |
| * Applies the query on the event. This method takes care of the performance related changed done |
| * to improve the CQ-query performance. When CQ-query is executed first time, it saves the query |
| * related information in the execution context and uses that info in later executions. |
| */ |
| private boolean evaluateQuery(CqQueryImpl cQuery, Object[] event) throws Exception { |
| ExecutionContext execContext = cQuery.getQueryExecutionContext(); |
| execContext.reset(); |
| execContext.setBindArguments(event); |
| boolean status = false; |
| |
| // Check if the CQ query is executed once. |
| // If not execute the query in normal way. |
| // During this phase the query execution related info are stored in the |
| // ExecutionContext. |
| if (execContext.getScopeNum() <= 0) { |
| SelectResults results = |
| (SelectResults) ((DefaultQuery) cQuery.getQuery()).executeUsingContext(execContext); |
| if (results != null && results.size() > 0) { |
| status = true; |
| } |
| } else { |
| // Execute using the saved query info (in ExecutionContext). |
| // This avoids building resultSet, index look-up, generating build-plans |
| // that are not required for; query execution on single object. |
| CompiledSelect cs = ((DefaultQuery) (cQuery.getQuery())).getSelect(); |
| status = cs.evaluateCq(execContext); |
| } |
| return status; |
| } |
| |
| @Override |
| public UserAttributes getUserAttributes(String cqName) { |
| return this.cqNameToUserAttributesMap.get(cqName); |
| } |
| |
| @Override |
| public void cqsDisconnected(Pool pool) { |
| invokeCqsConnected(pool, false); |
| } |
| |
| @Override |
| public void cqsConnected(Pool pool) { |
| invokeCqsConnected(pool, true); |
| } |
| |
| /** |
| * Let cq listeners know that they are connected or disconnected |
| */ |
| private void invokeCqsConnected(Pool pool, boolean connected) { |
| String poolName = pool.getName(); |
| // Check to see if we are already connected/disconnected. |
| // If state has not changed, do not invoke another connected/disconnected |
| synchronized (cqPoolsConnected) { |
| // don't repeatedly send same connect/disconnect message to cq's on repeated fails of |
| // RedundancySatisfier |
| if (cqPoolsConnected.containsKey(poolName) && connected == cqPoolsConnected.get(poolName)) { |
| return; |
| } |
| cqPoolsConnected.put(poolName, connected); |
| |
| Collection<? extends InternalCqQuery> cqs = this.getAllCqs(); |
| String cqName = null; |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| for (InternalCqQuery query : cqs) { |
| try { |
| if (query == null) { |
| continue; |
| } |
| |
| cqName = query.getName(); |
| ClientCQImpl cQuery = (ClientCQImpl) this.getCq(cqName); |
| |
| // Check cq pool to determine if the pool matches, if not continue. |
| // Also if the connected state is already the same, we do not have to send status again. |
| if (cQuery == null || cQuery.getCQProxy() == null) { |
| continue; |
| } |
| Pool cqPool = cQuery.getCQProxy().getPool(); |
| if (cQuery.isConnected() == connected || !cqPool.getName().equals(poolName)) { |
| continue; |
| } |
| |
| if ((!cQuery.isRunning() && cQuery.getQueuedEvents() == null)) { |
| if (isDebugEnabled) { |
| logger.debug("Unable to invoke CqListener, CQ is Not running, CqName : {}", cqName); |
| } |
| continue; |
| } |
| |
| this.invokeCqConnectedListeners(cqName, cQuery, connected); |
| } catch (VirtualMachineError e) { |
| SystemFailure.initiateFailure(e); |
| throw e; |
| } catch (Throwable t) { |
| SystemFailure.checkFailure(); |
| logger.warn("Error while sending connection status to cq listeners", t); |
| } |
| } |
| } |
| } |
| |
| @Override |
| public List<String> getAllDurableCqsFromServer(InternalPool pool) { |
| return new ServerCQProxyImpl(pool).getAllDurableCqsFromServer(); |
| } |
| } |
