import com.gemstone.gemfire.distributed.DistributedMember; //导入依赖的package包/类
/**
 * Creates and returns an EntryEventImpl.  Generates and assigns a bucket id to the
 * EntryEventImpl if the region parameter is a PartitionedRegion.
 * 
 * This creator does not specify the oldValue as this will be filled in later as part of an
 * operation on the region, or lets it default to null.
 * 
 * {@link EntryEventImpl#EntryEventImpl(LocalRegion, Operation, Object, Object, Object, boolean, DistributedMember, boolean, boolean)}
 */
@Retained
public static EntryEventImpl create(final LocalRegion region,
    Operation op, Object key, @Retained(ENTRY_EVENT_NEW_VALUE) Object newVal,
    Object callbackArgument,
    boolean originRemote, DistributedMember distributedMember,
    boolean generateCallbacks, boolean initializeId)  {
  EntryEventImpl entryEvent = new EntryEventImpl(region,op,key,newVal,callbackArgument,
      originRemote,distributedMember,generateCallbacks,initializeId);

  if(null != region) {
    region.updateKeyInfo(entryEvent, key, newVal);
  }
  
  return entryEvent;
}
 

import com.gemstone.gemfire.distributed.DistributedMember; //导入依赖的package包/类
/**
 * This method will be invoked whenever a member stops being a managing node.
 * The exception Management exception has to be handled by the caller.
 *
 * @throws ManagementException
 *
 */
private void stopManagingActivity() {

  try {

    pooledMembershipExecutor.shutdown();
    Iterator<DistributedMember> it = repo.getMonitoringRegionMap().keySet()
        .iterator();

    while (it.hasNext()) {
      removeMemberArtifacts(it.next(), false);
    }


  } catch (Exception e) {
    throw new ManagementException(e);
  }finally{
    // For future use
  }

}
 

import com.gemstone.gemfire.distributed.DistributedMember; //导入依赖的package包/类
public Set<DistributedMember> getMembers() {
  if (this.anyOneDataStoreMember) {
    final DistributedMember member = GemFireXDUtils.getGfxdAdvisor()
        .adviseDataStore(this.serverGroups, this.preferablySelf);
    if (member != null) {
      return Collections.singleton(member);
    }
    return Collections.emptySet();
  }
  else {
    if (this.serverGroups == null || this.serverGroups.size() == 0) {
      // for default group including all VMs return only datastores
      return GemFireXDUtils.getGfxdAdvisor().adviseDataStores(null);
    }
    // return both datastores and any accessors to execute the DAP
    return GemFireXDUtils.getGfxdAdvisor().adviseOperationNodes(
        this.serverGroups);
  }
}
 

import com.gemstone.gemfire.distributed.DistributedMember; //导入依赖的package包/类
/***
 * Executes a function with arguments on a set of members , ignores the departed members.
 * @param function Function to be executed.
 * @param args Arguments passed to the function, pass null if you wish to pass no arguments to the function.
 * @param targetMembers Set of members on which the function is to be executed.
 *
 * @return ResultCollector
 */
public static ResultCollector<?, ?> executeFunction(final Function function, Object args , final Set<DistributedMember> targetMembers) {
  Execution execution = null;

  if (args != null) {
    execution = FunctionService.onMembers(targetMembers).withArgs(args);
  } else {
    execution = FunctionService.onMembers(targetMembers);
  }

  ((AbstractExecution) execution).setIgnoreDepartedMembers(true);
  return execution.execute(function);
}
 

import com.gemstone.gemfire.distributed.DistributedMember; //导入依赖的package包/类
/** Creates and initializes the singleton instance of ParRegTest in this VM
 *  for HA testing with a PR data store.
 */
public synchronized static void HydraTask_HA_initializeDataStore() {
  if (testInstance == null) {
      PRObserver.installObserverHook();
      testInstance = new ThresholdsTest();
      testInstance.initializeRegion("dataStoreRegion");
      testInstance.initializeInstance();
      testInstance.isDataStore = true;
      if (testInstance.isBridgeConfiguration) {
         testInstance.isBridgeClient = false;
         BridgeHelper.startBridgeServer("bridge");
      }
      // Indicate that the PR(s) are initialized before allowing another
      // member to proceed with the criticalThenDie task
      ((ThresholdsTest) testInstance).initCriticalThenDie();
      DistributedMember dm = DistributedSystemHelper.getDistributedSystem().getDistributedMember();
      ThresholdsTest.putDistributedMember(dm);
      ((ThresholdsTest)testInstance).initNoEvictRegion();
   }
   testInstance.uniqueKeyIndex.set(new Integer(RemoteTestModule.getCurrentThread().getThreadId()));
}
 

import com.gemstone.gemfire.distributed.DistributedMember; //导入依赖的package包/类
private Map<DistributedMember, Properties[]> getMemberPropsMap()
    throws SQLException {
  if (this.memberPropsMap == null) {
    try {
      final MemberSingleResultCollector<Properties[]> collector =
        new MemberSingleResultCollector<Properties[]>();
      final DistributedMember myId = this.dm.getDistributionManagerId();
      /* [sb] restricting to self for fixing #43219
       final GfxdConfigMessage<TreeMap<DistributedMember, TreeMap<?, ?>[]>> msg
        = new GfxdConfigMessage<TreeMap<DistributedMember, TreeMap<?, ?>[]>>(
            collector, null, GfxdConfigMessage.Operation.GET_ALLPROPS,
            null, true); 
       */
      final GfxdConfigMessage<TreeMap<DistributedMember, Properties[]>> msg =
        new GfxdConfigMessage<TreeMap<DistributedMember, Properties[]>>(
          collector, Collections.singleton(myId),
          GfxdConfigMessage.Operation.GET_ALLPROPS, null, true);
      this.memberPropsMap = msg.executeFunction();
    } catch (Throwable t) {
      throw Util.javaException(t);
    }
  }
  return this.memberPropsMap;
}
 

import com.gemstone.gemfire.distributed.DistributedMember; //导入依赖的package包/类
@Test
public void test012GetMembersContainsOnlySelf() {
  final Cache mockCache = mockContext.mock(Cache.class, "Cache");

  final DistributedSystem mockDistributedSystem = mockContext.mock(DistributedSystem.class, "DistributedSystem");
  final DistributedMember mockMemberSelf = createMockMember("S", "Self");

  mockContext.checking(new Expectations() {{
    oneOf(mockCache).getMembers();
    will(returnValue(Collections.emptySet()));
    oneOf(mockCache).getDistributedSystem();
    will(returnValue(mockDistributedSystem));
    oneOf(mockDistributedSystem).getDistributedMember();
    will(returnValue(mockMemberSelf));
  }});

  final AbstractCommandsSupport commands = createAbstractCommandsSupport(mockCache);

  final Set<DistributedMember> expectedMembers = CollectionUtils.asSet(mockMemberSelf);
  final Set<DistributedMember> actualMembers = commands.getMembers(mockCache);

  assertNotNull(actualMembers);
  assertEquals(expectedMembers.size(), actualMembers.size());
  assertTrue(actualMembers.containsAll(expectedMembers));
}
 

import com.gemstone.gemfire.distributed.DistributedMember; //导入依赖的package包/类
public GfxdFunctionReplyMessageProcessor(DM dm,
    Set<DistributedMember> members, GfxdFunctionMessage<T> msg) {
  super(dm, members, false);
  this.userCollector = msg.userCollector;
  this.recordStats = msg.timeStatsEnabled;
  this.replyReceivedMsgs = msg.replyReceivedMsgs;
  this.allowExecutionOnAdminMembers = msg.allowExecutionOnAdminMembers();
  this.isDirectReplyMessage = (msg instanceof DirectReplyMessage
      && ((DirectReplyMessage)msg).supportsDirectAck());
  if (msg.requiresSync()) {
    this.latch = new StoppableCountDownLatch(dm.getCancelCriterion(), 1);
  }
  else {
    this.latch = null;
  }
  this.helper = new GfxdResultCollectorHelper();
}
 

import com.gemstone.gemfire.distributed.DistributedMember; //导入依赖的package包/类
private static void verifyColocatedRegions(Region regionA) {
  Object[] keys = PartitionRegionHelper.getLocalPrimaryData(regionA).keySet().toArray();
  Map colocatedRegionsMap = PartitionRegionHelper.getColocatedRegions(regionA);
  Object[] regionNames = colocatedRegionsMap.keySet().toArray();
  Log.getLogWriter().info("Verifying PR " + regionA.getFullPath() + " with " + colocatedRegionsMap.size() + " colocated regions");
  for (Object regionName: regionNames) {
    Region regionB = (Region)colocatedRegionsMap.get(regionName);
    for (Object aKey: keys) {
      DistributedMember primaryA = PartitionRegionHelper.getPrimaryMemberForKey(regionA, aKey);
      DistributedMember primaryB = PartitionRegionHelper.getPrimaryMemberForKey(regionB, aKey);
      if ((primaryB != null) && !(primaryA.equals(primaryB))) {
        throw new TestException("verifyColocatedRegions reports that primary for " + aKey + " is " + primaryA + " for " + regionA.getName() + ", but the primary for the same entry in " + regionB.getName() + " is " + primaryB);
      } 
    }
    Log.getLogWriter().info("Done verifying consistency in colocated regions");
  }
}
 

import com.gemstone.gemfire.distributed.DistributedMember; //导入依赖的package包/类
@Override
protected void addExceptionFromMember(Throwable t,
    final DistributedMember member) {
  super.addExceptionFromMember(t, member);
  if (this.memberProcessorIds != null) {
    //Fix for 43877 - make sure we don't double count
    //crashed members that also send a reply.
    //indicate that the number failed. -1 is not a valid reply processor
    //id - see ProcessorKeeeper21.put.
    this.memberProcessorIds.put(member, -1);
    signalDeletesAtEndOfReferenceKeyCheck();
    if (GemFireXDUtils.TraceRSIter) {
      SanityManager.DEBUG_PRINT(GfxdConstants.TRACE_RSITER, toString()
          + "#addExceptionFromMember: added -1 expected processorId "
          + "due to exception " + t + " from member " + member);
    }
  }
  else {
    if (GemFireXDUtils.TraceRSIter) {
      SanityManager.DEBUG_PRINT(GfxdConstants.TRACE_RSITER, toString()
          + "#addExceptionFromMember: received exception " + t
          + " from member " + member + " waitForReferencedKeyResults="
          + this.waitForReferencedKeyResults);
    }
  }
}
 

import com.gemstone.gemfire.distributed.DistributedMember; //导入依赖的package包/类
@Override
public void memberRemoved(PersistentMemberID persistentID, boolean revoked) {
  if(!revoked) {
    return;
  }
  
  DistributedMember dmem = prRegion.getSystem().getDistributedMember();
  if(getLogger().fineEnabled()) {
    getLogger().fine(
        "Persistent Membership Listener invoked on DistributedMember = " + dmem
        + " for removed memberId = " + persistentID);
  }
  
  if (! prRegion.isCacheClosing() && !prRegion.isDestroyed() && !prRegion.isFixedPartitionedRegion()) {
    scheduleRedundancyRecovery(persistentID);
  }
}
 

import com.gemstone.gemfire.distributed.DistributedMember; //导入依赖的package包/类
@Override
public void fromData(DataInput in)
    throws IOException, ClassNotFoundException {
  super.fromData(in);
  try {
    // Read EventID
    EventID eventID = (EventID)DataSerializer.readObject(in);
    DistributedMember member = DSFIDFactory.readInternalDistributedMember(in);
    String regionName = DataSerializer.readString(in);
    final LocalRegion rgn = Misc.getGemFireCache()
        .getRegionByPathForProcessing(regionName);

    if (rgn != null) {
      this.initializeEvent(rgn, eventID, member);
    }
  } catch (IOException ioe) {
    throw ioe;
  } catch (Exception e) {
    throw new IOException(e);
  }
}
 

import com.gemstone.gemfire.distributed.DistributedMember; //导入依赖的package包/类
public final void applyOperation() throws StandardException {
  // Distribute the message only in case of PR
  try {
    @SuppressWarnings("unchecked")
    Set<DistributedMember> members = ((CacheDistributionAdvisee)rgn)
        .getCacheDistributionAdvisor()
        .adviseSerialAsyncEventQueueOrGatewaySender();
    if (GemFireXDUtils.TraceDBSynchronizer) {
      SanityManager.DEBUG_PRINT(GfxdConstants.TRACE_DB_SYNCHRONIZER,
          "BaseActivation: distributing bulk DML to remote nodes. "
              + "Message={" + this + "} to members=" + members);
    }
    this.send(this.rgn.getSystem(), members);

  } catch (SQLException sqle) {
    throw Misc.wrapSQLException(sqle, sqle);
  }
  addToLocalDBSynchronizerConditionally(true);
}
 

import com.gemstone.gemfire.distributed.DistributedMember; //导入依赖的package包/类
public SystemMember lookupSystemMember(DistributedMember distributedMember) 
throws AdminException {
  checkForRmiConnection();
  try {
    ObjectName name = (ObjectName) 
        this.mbs.invoke(this.objectName, "manageSystemMember",
                        new Object[] { distributedMember }, 
                        new String[] { DistributedMember.class.getName() });
    if (name == null) {
      return null;
    }
    return new JMXSystemMember(this.mbs, name);
    
  } catch (Exception ex) {
    String s = "While calling manageSystemMemberApplications";
    throw new InternalGemFireException(s, ex);
  }
}
 

import com.gemstone.gemfire.distributed.DistributedMember; //导入依赖的package包/类
private String getParticipatingServerGroups(
    final InternalDistributedMember[] members) {

  String s = null;

  for (final DistributedMember m : members) {
    final SortedSet<String> g = ServerGroupUtils.getServerGroups(m);
    if (s == null) {
      s = SharedUtils.toCSV(g);
      continue;
    }
    s = SortedCSVProcedures.groupsIntersection(s, SharedUtils.toCSV(g));
  }

  return s;
}
 

import com.gemstone.gemfire.distributed.DistributedMember; //导入依赖的package包/类
/**
 * Constructs new RoleEventImpl.
 *
 * @param requiredRoles set of required roles that are affected by this event 
 */
RoleEventImpl(Region region, Operation op, Object callbackArgument,
                     boolean originRemote, 
                     DistributedMember distributedMember, Set requiredRoles) {
  super(region, op, callbackArgument, originRemote, distributedMember);
  this.requiredRoles = Collections.unmodifiableSet(requiredRoles);
}
 

import com.gemstone.gemfire.distributed.DistributedMember; //导入依赖的package包/类
public void sendManagerInfo(DistributedMember receiver) {

    String levelName = jmxAdapter.getAlertLevel();
    int alertCode = LogWriterImpl.levelNameToCode(levelName);
    ManagerStartupMessage msg = ManagerStartupMessage.create(alertCode);
    msg.setRecipient((InternalDistributedMember) receiver);
    sendAsync(msg);

  }
 

import com.gemstone.gemfire.distributed.DistributedMember; //导入依赖的package包/类
public String listOfAllMembers(ArrayList<DistributedMember> dsMemberList) {
  StringBuilder listMembersId = new StringBuilder();    
  for (int j = 0; j < dsMemberList.size() - 1; j++) {
    listMembersId.append(dsMemberList.get(j).getId());      
    listMembersId.append(" ; ");
  }
  return listMembersId.toString();
}
 

import com.gemstone.gemfire.distributed.DistributedMember; //导入依赖的package包/类
@Test
public void testGetDiskStoreDescription() {
  final String diskStoreName = "mockDiskStore";
  final String memberId = "mockMember";

  final Cache mockCache = mockContext.mock(Cache.class, "Cache");

  final DistributedMember mockMember = mockContext.mock(DistributedMember.class, "DistributedMember");

  final Execution mockFunctionExecutor = mockContext.mock(Execution.class, "Function Executor");

  final ResultCollector mockResultCollector = mockContext.mock(ResultCollector.class, "ResultCollector");

  final DiskStoreDetails expectedDiskStoredDetails = createDiskStoreDetails(memberId, diskStoreName);

  mockContext.checking(new Expectations() {{
    oneOf(mockMember).getName();
    will(returnValue(null));
    oneOf(mockMember).getId();
    will(returnValue(memberId));
    oneOf(mockFunctionExecutor).withArgs(with(equal(diskStoreName)));
    will(returnValue(mockFunctionExecutor));
    oneOf(mockFunctionExecutor).execute(with(aNonNull(DescribeDiskStoreFunction.class)));
    will(returnValue(mockResultCollector));
    oneOf(mockResultCollector).getResult();
    will(returnValue(Arrays.asList(expectedDiskStoredDetails)));
  }});

  final DiskStoreCommands commands = createDiskStoreCommands(mockCache, mockMember, mockFunctionExecutor);

  final DiskStoreDetails actualDiskStoreDetails = commands.getDiskStoreDescription(memberId, diskStoreName);

  assertNotNull(actualDiskStoreDetails);
  assertEquals(expectedDiskStoredDetails, actualDiskStoreDetails);
}
 

import com.gemstone.gemfire.distributed.DistributedMember; //导入依赖的package包/类
/**
 * 
 * @param query
 * @param sqi
 * @param partioningType
 * @param routingInfo . It is two dimensional Object[][]. Each row represents the pruning  which 
 * can be accomplished by a condition. Two consecutive rows (conditions) are tied by an AND or OR junction  
 * @return
 * @throws StandardException
 */

public static Set<DistributedMember> getExpectedNodes(final String query,
    QueryInfo sqi, Object[][] routingInfo, LogWriter logger) throws StandardException {
  PartitionedRegion pr = (PartitionedRegion)sqi.getRegion();
  pr
      .getPartitionResolver();
  Set<InternalDistributedMember> allNodes = pr
      .getRegionAdvisor().adviseDataStore();
  Set<DistributedMember> expected = new HashSet<DistributedMember>();
  expected.addAll(allNodes);
  for (int i = 0; i < routingInfo.length; ++i) {
    Object[] currentConditionRow = routingInfo[i];
    int partitioningType = ((Integer)currentConditionRow[0]).intValue();
    switch (partitioningType) {
      case byrange:
        handleRangePartitioning(sqi, currentConditionRow, allNodes, expected,i, logger);
        break;      
      case bylist : 
        handleListPartitioning(sqi, currentConditionRow, allNodes, expected,i,logger);
      break; 
      case bycolumn :
        handleColumnPartitioning(sqi, currentConditionRow, allNodes, expected,i,logger); 
        break;
      case allnodes:
        expected.addAll(allNodes);   
        break;    
    default:
      throw new IllegalArgumentException("The partition type is undefined");
    }
  }
  return expected;

}