DistributedRegion extends LocalRegion implements InternalDistributedRegion { void validateAsynchronousEventDispatcher(String asyncDispatcherId) { for (GatewaySender sender : getCache().getAllGatewaySenders()) { if (sender.isParallel() && sender.getId().equals(asyncDispatcherId)) { if (sender.getId().contains(AsyncEventQueueImpl.ASYNC_EVENT_QUEUE_PREFIX)) { throw new AsyncEventQueueConfigurationException(String.format( "Parallel Async Event Queue %s can not be used with replicated region %s", getAsyncEventQueueIdFromSenderId(sender.getId()), getFullPath())); } else { throw new GatewaySenderConfigurationException( String.format("Parallel Gateway Sender %s can not be used with replicated region %s", sender.getId(), getFullPath())); } } } } protected  DistributedRegion(String regionName, RegionAttributes attrs, LocalRegion parentRegion,
      InternalCache cache, InternalRegionArguments internalRegionArgs,
      StatisticsClock statisticsClock); boolean allowsPersistence(); @Override boolean virtualPut(EntryEventImpl event, boolean ifNew, boolean ifOld,
      Object expectedOldValue, boolean requireOldValue, long lastModified,
      boolean overwriteDestroyed, boolean invokeCallbacks, boolean throwConcurrentModificaiton); @Override void basicPutPart3(EntryEventImpl event, RegionEntry entry, boolean isInitialized,
      long lastModified, boolean invokeCallbacks, boolean ifNew, boolean ifOld,
      Object expectedOldValue, boolean requireOldValue); @Override boolean hasSeenEvent(EntryEventImpl event); @Override void checkForLimitedOrNoAccess(); @Override void handleReliableDistribution(Set successfulRecipients); boolean doesNotDistribute(); @Override boolean shouldSyncForCrashedMember(InternalDistributedMember id); @Override boolean requiresReliabilityCheck(); @Override void localDestroy(Object key, Object aCallbackArgument); @Override void invalidate(Object key, Object aCallbackArgument); @Override Lock getRegionDistributedLock(); @Override Lock getDistributedLock(Object key); @Override void preInitialize(); @Override void initialize(InputStream snapshotInputStream, InternalDistributedMember imageTarget,
      InternalRegionArguments internalRegionArgs); void scheduleSynchronizeForLostMember(InternalDistributedMember member,
      VersionSource lostVersionID, long delay); void setRegionSynchronizeScheduled(VersionSource lostMemberVersionID); boolean setRegionSynchronizedWithIfNotScheduled(VersionSource lostMemberVersionID); boolean isInitializedWithWait(); @Override boolean lockWhenRegionIsInitializing(); @Override void unlockWhenRegionIsInitializing(); @Override void basicBridgeRemove(Object key, Object expectedOldValue, Object p_callbackArg,
      ClientProxyMembershipID memberId, boolean fromClient, EntryEventImpl clientEvent); @Override void basicBridgeDestroy(Object key, Object p_callbackArg, ClientProxyMembershipID memberId,
      boolean fromClient, EntryEventImpl clientEvent); @Override void basicBridgeInvalidate(Object key, Object p_callbackArg,
      ClientProxyMembershipID memberId, boolean fromClient, EntryEventImpl clientEvent); @Override void basicDestroy(EntryEventImpl event, boolean cacheWrite, Object expectedOldValue); @Override void basicDestroyPart3(RegionEntry re, EntryEventImpl event, boolean inTokenMode,
      boolean duringRI, boolean invokeCallbacks, Object expectedOldValue); @Override void basicInvalidate(EntryEventImpl event); DistributionConfig getDistributionConfig(); @Override CacheDistributionAdvisor getDistributionAdvisor(); @Override CacheDistributionAdvisor getCacheDistributionAdvisor(); PersistenceAdvisor getPersistenceAdvisor(); PersistentMemberID getPersistentID(); @Override Profile getProfile(); @Override void fillInProfile(Profile profile); DistributedLockService getLockService(); @Override boolean isLockGrantor(); @Override void becomeLockGrantor(); @Override boolean cacheWriteBeforeDestroy(EntryEventImpl event, Object expectedOldValue); @Override void postCreateRegion(); @Override void cleanupFailedInitialization(); @Override void handleCacheClose(Operation operation); @Override void cacheWriteBeforePut(EntryEventImpl event, Set netWriteRecipients,
      CacheWriter localWriter, boolean requireOldValue, Object expectedOldValue); @Override void addGatewaySenderId(String gatewaySenderId); @Override void removeGatewaySenderId(String gatewaySenderId); @Override void addAsyncEventQueueId(String asyncEventQueueId); @Override void removeAsyncEventQueueId(String asyncEventQueueId); @Override Iterator<RegionEntry> getBestIterator(boolean includeValues); @Override long postPutAllSend(DistributedPutAllOperation putAllOp,
      VersionedObjectList successfulPuts); @Override long postRemoveAllSend(DistributedRemoveAllOperation op,
      VersionedObjectList successfulOps); @Override VersionedObjectList basicRemoveAll(final Collection<Object> keys,
      final DistributedRemoveAllOperation removeAllOp, final List<VersionTag> retryVersions); Set waitForRequiredRoles(long timeout); boolean isRoleInRegionMembership(Role role); @Override void remoteRegionInitialized(CacheProfile profile); @Override DistributionAdvisee getParentAdvisee(); @Override DistributedMember getOwnerForKey(KeyInfo key); @Override ResultCollector executeFunction(final DistributedRegionFunctionExecutor execution,
      final Function function, final Object args, final ResultCollector rc, final Set filter,
      final ServerToClientFunctionResultSender sender); @Override void removeCriticalMember(DistributedMember member); @Override MemoryThresholdInfo getAtomicThresholdInfo(); boolean hasNetLoader(); @Override Set adviseNetWrite(); @VisibleForTesting SenderIdMonitor getSenderIdMonitor(); @MutableForTesting
  static boolean TEST_HOOK_ADD_PROFILE; @MutableForTesting
  static boolean ignoreReconnect; }
@Test public void validateAsynchronousEventDispatcherShouldDoNothingWhenDispatcherIdCanNotBeFound() { InternalCache internalCache = mock(InternalCache.class); when(internalCache.getAllGatewaySenders()) .thenReturn(Collections.singleton(mock(GatewaySender.class))); DistributedRegion distributedRegion = mock(DistributedRegion.class); when(distributedRegion.getCache()).thenReturn(internalCache); doCallRealMethod().when(distributedRegion).validateAsynchronousEventDispatcher(anyString()); distributedRegion.validateAsynchronousEventDispatcher("nonExistingDispatcher"); }
@Test public void validateAsynchronousEventDispatcherShouldDoNothingWhenFoundDispatcherIsSerial() { String senderId = "mySender"; GatewaySender serialSender = mock(GatewaySender.class); when(serialSender.isParallel()).thenReturn(false); when(serialSender.getId()).thenReturn(senderId); InternalCache internalCache = mock(InternalCache.class); when(internalCache.getAllGatewaySenders()).thenReturn(Collections.singleton(serialSender)); DistributedRegion distributedRegion = mock(DistributedRegion.class); when(distributedRegion.getCache()).thenReturn(internalCache); doCallRealMethod().when(distributedRegion).validateAsynchronousEventDispatcher(anyString()); distributedRegion.validateAsynchronousEventDispatcher(senderId); }
@Test public void validateAsynchronousEventDispatcherShouldThrowExceptionWhenDispatcherIdMatchesAnExistingParallelAsyncEventQueue() { String senderId = "senderId"; String regionPath = "thisRegion"; String internalSenderId = getSenderIdFromAsyncEventQueueId(senderId); GatewaySender parallelAsyncEventQueue = mock(GatewaySender.class); when(parallelAsyncEventQueue.isParallel()).thenReturn(true); when(parallelAsyncEventQueue.getId()).thenReturn(internalSenderId); InternalCache internalCache = mock(InternalCache.class); when(internalCache.getAllGatewaySenders()) .thenReturn(Collections.singleton(parallelAsyncEventQueue)); DistributedRegion distributedRegion = mock(DistributedRegion.class); when(distributedRegion.getCache()).thenReturn(internalCache); when(distributedRegion.getFullPath()).thenReturn(regionPath); doCallRealMethod().when(distributedRegion).validateAsynchronousEventDispatcher(anyString()); assertThatThrownBy( () -> distributedRegion.validateAsynchronousEventDispatcher(internalSenderId)) .isInstanceOf(AsyncEventQueueConfigurationException.class) .hasMessage("Parallel Async Event Queue " + senderId + " can not be used with replicated region " + regionPath); }
@Test public void validateAsynchronousEventDispatcherShouldThrowExceptionWhenDispatcherIdMatchesAnExistingParallelGatewaySender() { String senderId = "senderId"; String regionPath = "thisRegion"; GatewaySender parallelGatewaySender = mock(GatewaySender.class); when(parallelGatewaySender.isParallel()).thenReturn(true); when(parallelGatewaySender.getId()).thenReturn(senderId); InternalCache internalCache = mock(InternalCache.class); when(internalCache.getAllGatewaySenders()) .thenReturn(Collections.singleton(parallelGatewaySender)); DistributedRegion distributedRegion = mock(DistributedRegion.class); when(distributedRegion.getCache()).thenReturn(internalCache); when(distributedRegion.getFullPath()).thenReturn(regionPath); doCallRealMethod().when(distributedRegion).validateAsynchronousEventDispatcher(anyString()); assertThatThrownBy(() -> distributedRegion.validateAsynchronousEventDispatcher(senderId)) .isInstanceOf(GatewaySenderConfigurationException.class) .hasMessage("Parallel Gateway Sender " + senderId + " can not be used with replicated region " + regionPath); }