import net.sf.ehcache.distribution.CacheManagerPeerProvider; //导入依赖的package包/类
/**
 * Checks the configuration for peer provider that work with replication schemes for which place
 * holders exist. If such a peer provider and place holder exists the provider will be attached
 * to the place holder.
 *
 * @param cacheManager
 *            the cache manager to be modified
 * @param helper
 *            the configuration helper
 */
private static void attachPeerProviderToPlaceHolder(CacheManager cacheManager,
        ConfigurationHelper helper) {
    Map<String, CacheManagerPeerProvider> peerProviderMap = helper.createCachePeerProviders();
    if (peerProviderMap != null) {
        for (String scheme : peerProviderMap.keySet()) {
            CacheManagerPeerProvider provider = cacheManager
                    .getCacheManagerPeerProvider(scheme);
            if (provider instanceof PlaceHolderCacheManagerPeerProvider) {
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Adding cache peer provider for scheme " + scheme
                            + " to place holder");
                }
                ((PlaceHolderCacheManagerPeerProvider) provider)
                .attachCacheManagerPeerProvider(peerProviderMap.get(scheme));
            } else {
                LOG.warn("Ignoring cache peer provider for scheme " + scheme
                        + " because there is no place holder for that scheme");
            }
        }
    }
}
 

import net.sf.ehcache.distribution.CacheManagerPeerProvider; //导入依赖的package包/类
protected static int waitForClusterMembership(int time, TimeUnit unit, final Collection<String> cacheNames, final CacheManager... managers) {
    Integer minimumPeers = null;
    for (CacheManager cmanager : managers) {
        CacheManagerPeerProvider peerProvider = cmanager.getCacheManagerPeerProvider("RMI");
        if (peerProvider != null) {
            for (String cacheName : cacheNames) {
                int peers = peerProvider.listRemoteCachePeers(cmanager.getEhcache(cacheName)).size();
                if (minimumPeers == null || peers < minimumPeers) {
                    minimumPeers = peers;
                }
            }
        }
    }
    if (minimumPeers == null) {
        return 0;
    } else {
        return minimumPeers + 1;
    }
}
 

import net.sf.ehcache.distribution.CacheManagerPeerProvider; //导入依赖的package包/类
@Override
public CacheManagerPeerProvider createCachePeerProvider(CacheManager cacheManager, Properties properties) {
    try {
    	String configuration = JGroupsUtils.getConfigurationString(ApplicationProperty.HibernateClusterConfiguration.value());
    	JGroupsCacheManagerPeerProvider peerProvider = new JGroupsCacheManagerPeerProvider(cacheManager, configuration);
    	peerProvider.setChannelName("UniTime:hibernate");
        
        return peerProvider;
    } catch (Exception e) {
    	Debug.error(e.getMessage(), e);
    	return null;
    }
}
 

import net.sf.ehcache.distribution.CacheManagerPeerProvider; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
public CacheManagerPeerProvider createCachePeerProvider(CacheManager cacheManager,
        Properties properties) {
    String providerScheme = properties.getProperty("scheme");
    if (StringUtils.isBlank(providerScheme)) {
        throw new CacheException(
                "No scheme specified for which a place holder should be registered");
    }
    return new PlaceHolderCacheManagerPeerProvider(providerScheme);
}
 

import net.sf.ehcache.distribution.CacheManagerPeerProvider; //导入依赖的package包/类
/**
 * Attaches and initializes a peer provider to be wrapped by the place holder. The peer provider
 * will than be available to the cache manager.
 * 
 * @param peerProvider
 *            the peer provider to add
 * @throws CacheException
 *             in case there is already a peer provider attached or the scheme of the peer
 *             provider does not match
 */
public synchronized void attachCacheManagerPeerProvider(CacheManagerPeerProvider peerProvider)
        throws CacheException {
    if (this.peerProvider != null) {
        throw new CacheException("This placeholder already wraps a peer provider");
    }
    if (!peerProvider.getScheme().equals(scheme)) {
        throw new CacheException(
                "The scheme of the peer provider does not match the scheme of the place holder");
    }
    this.peerProvider = peerProvider;
    this.peerProvider.init();
}
 

import net.sf.ehcache.distribution.CacheManagerPeerProvider; //导入依赖的package包/类
/**
 * Given an {@link CacheManager} get the corresponding instance of this class.
 */
public static HazelcastCacheManagerPeerProvider getCachePeerProvider(final CacheManager cacheManager) {
    final CacheManagerPeerProvider provider = cacheManager.getCacheManagerPeerProvider(SCHEME_NAME);
    if (provider == null) {
        LOG.warn("No CacheManagerPeerProvider registered for " + SCHEME_NAME + " scheme.");
        return null;
    }
    if (!(provider instanceof HazelcastCacheManagerPeerProvider)) {
        LOG.warn("CacheManagerPeerProvider for scheme " + SCHEME_NAME + " cannot be cast to " + HazelcastCacheManagerPeerProvider.class.getName());
        return null;
    }
    return (HazelcastCacheManagerPeerProvider) provider;
}
 

import net.sf.ehcache.distribution.CacheManagerPeerProvider; //导入依赖的package包/类
private static void displayCacheInfo(PrintWriter writer, CacheManager cm) {
    writer.println(cm.getName() + " EHCACHE INFORMATION:");
    writer.println("-------------------:");
    writer.println("Manager Status: " + cm.getStatus());
    writer.println("DiskStore Path: " + cm.getConfiguration().getDiskStoreConfiguration().getPath());
    writer.println();

    writer.println("Caches: ");
    Map<String, CacheManagerPeerProvider> peerProvs = cm.getCacheManagerPeerProviders();
    String[] cacheNames = cm.getCacheNames();
    for(String n : cacheNames) {
        Ehcache c = cm.getCache(n);
        writer.println("\t" + c.getName() + " @" + c.hashCode());
        writer.println("\tCache Status    : " + c.getStatus());
        writer.println("\tMax Heap       : " + c.getCacheConfiguration().getMaxBytesLocalHeap()/(1024 * 1024) + "MB");
        writer.println("\tMax Entries    : " + c.getCacheConfiguration().getMaxEntriesLocalHeap());
        writer.println("\tStatistics     : " + getStats(c));
        for (CacheManagerPeerProvider peerProv : peerProvs.values()) {
            List peers = peerProv.listRemoteCachePeers(c);
            for(Object o : peers) {
                CachePeer cp = (CachePeer) o;
                try {
                    writer.println("\tReplicating with: " + cp.getUrl());
                } catch (RemoteException e) {}
            }
        }
        writer.println();
    }
}
 

import net.sf.ehcache.distribution.CacheManagerPeerProvider; //导入依赖的package包/类
@Override
public CacheManagerPeerProvider createCachePeerProvider(final CacheManager cacheManager, final Properties properties) {
    return new HazelcastCacheManagerPeerProvider(cacheManager);
}
 

import net.sf.ehcache.distribution.CacheManagerPeerProvider; //导入依赖的package包/类
@Override
public CacheManagerPeerProvider createCachePeerProvider(
		CacheManager cacheManager, Properties properties)
		throws CacheException {
   	CacheManagerPeerListener cachePeerListener = cacheManager.getCachePeerListener("RMI");
   	if (cachePeerListener == null) {
   		throw new CacheException("Could not retrieve RMI cache peer listener. " +
   				"Please check your configuration, the class of the cacheManagerPeerListenerFactory " +
   				"element should be set to 'net.sf.ehcache.distribution.RMICacheManagerPeerListenerFactory'");
   	}
   	
     	// This returns "RMI listener port: 61616" from which we can parse the port number out
   	Matcher portMatcher =
   		UNIQUE_RESOURCE_PORT_REGEX.matcher(cachePeerListener.getUniqueResourceIdentifier());
   	if (!portMatcher.matches()) {
   		throw new CacheException("Could not parse RMI listener port from unique " +
   				"resource identifier string: '" + 
   				cachePeerListener.getUniqueResourceIdentifier() + 
   				"'");
   	}

   	AwsDiscoveryServiceConfig config = new AwsDiscoveryServiceConfig();
   	config.setRmiListenerPort(Integer.parseInt(portMatcher.group(1)));
   	config.setAccessKey(AwsSecurityUtils.getInstance().getAwsAccessKey());
   	config.setSecretKey(AwsSecurityUtils.getInstance().getAwsSecretKey());
  	 	config.setSecurityGroup(PropertyUtil.extractAndLogProperty(AWS_SECURITY_GROUP_PROP, properties));
  	 	String securityGroupRefreshIntervalString =
  	 			PropertyUtil.extractAndLogProperty(AWS_SECURITY_GROUP_REFRESH_INTERVAL_PROP, properties);
  	 	
  	 	try {
  	 		config.setSecurityGroupRefreshInterval(Integer.parseInt(securityGroupRefreshIntervalString));
  	 	} catch (NumberFormatException e) {
  	 		throw new CacheException("Expected an integer interval in ms for " +
  	 				AWS_SECURITY_GROUP_REFRESH_INTERVAL_PROP +
  	 				", but got '" + securityGroupRefreshIntervalString + "'");
  	 	}
  	 	
  	 	// Now attempt to get the class for the service factory
  	 	DiscoveryServiceFactory discoveryServiceFactory =
  	 			getDefaultOrConfiguredDiscoveryServiceFactory(
  	 					PropertyUtil.extractAndLogProperty(DISCOVERY_SERVICE_CLASS_PROP, properties));

  	 	cachePeerProvidersLock.lock();
	try {
		// Does this provider already exist?
		AwsSecurityGroupAwareCacheManagerPeerProvider rmiPeerProvider = cachePeerProviders.get(config.hashCode());
		if (rmiPeerProvider != null && rmiPeerProvider.isActive()) {
			LOG.debug("Returning cached RMI peer provider instance " + rmiPeerProvider.toString());
			return rmiPeerProvider;
		}

		// Create a new provider
		LOG.debug("Creating a new AWS security group aware RMI peer provider instance...");
		rmiPeerProvider = new AwsSecurityGroupAwareCacheManagerPeerProvider(config, discoveryServiceFactory);
		LOG.debug("Created a new AWS security group aware RMI peer provider instance {}", rmiPeerProvider.toString());
        cachePeerProviders.put(config.hashCode(), rmiPeerProvider);
		return rmiPeerProvider;
	} finally {
		cachePeerProvidersLock.unlock();
	}
}
 

import net.sf.ehcache.distribution.CacheManagerPeerProvider; //导入依赖的package包/类
private void waitForCachesJoiningCluster() throws InterruptedException {
    CacheManager cacheManager = CacheManager.newInstance(getClass().getResource("/ehcache_distributed_test.xml"));

    CacheManagerPeerProvider peerProvider = cacheManager.getCacheManagerPeerProvider("RMI");

    int peers = 0;
    long timeoutMs = 10000;
    long startMs = System.currentTimeMillis();
    do {
        Thread.sleep(1000);

        peers = peerProvider.listRemoteCachePeers(cacheManager.getEhcache("bpCache1")).size();

    } while (peers < 2 && (System.currentTimeMillis() - startMs < timeoutMs));

    Assert.assertEquals(2, peers);

    cacheManager.shutdown();
}
 

import net.sf.ehcache.distribution.CacheManagerPeerProvider; //导入依赖的package包/类
private void tracePeers() {
    CacheManagerPeerProvider peerProvider = manager.getCacheManagerPeerProvider("RMI");
    int peers = peerProvider.listRemoteCachePeers(manager.getEhcache(cacheName)).size();
    log.trace("Found {} remote cache peer(s)", peers);
}