import org.infinispan.remoting.transport.Address; //导入依赖的package包/类
private void handleViewChangeInternal(ViewChangedEvent e) {
  synchronized (InfinispanClusterManager.this) {
    if (!active) {
      return;
    }

    multimaps.forEach(InfinispanAsyncMultiMap::clearCache);

    List<Address> added = new ArrayList<>(e.getNewMembers());
    added.removeAll(e.getOldMembers());
    log.debug("Members added = " + added);
    added.forEach(address -> {
      if (nodeListener != null) {
        nodeListener.nodeAdded(address.toString());
      }
    });
    List<Address> removed = new ArrayList<>(e.getOldMembers());
    removed.removeAll(e.getNewMembers());
    log.debug("Members removed = " + removed);
    removed.forEach(address -> {
      if (nodeListener != null) {
        nodeListener.nodeLeft(address.toString());
      }
    });
  }
}
 

import org.infinispan.remoting.transport.Address; //导入依赖的package包/类
private InetAddress addressToInetAddress(Address a) {
    EmbeddedCacheManager manager = this.cm;
    if ((manager == null) || (a == null)) {
        // In case we cannot fetch the information, lets assume we
        // are standby, so to have less responsibility.
        return null;
    }
    Transport t = manager.getTransport();
    if (t instanceof JGroupsTransport) {
        JGroupsTransport jt = (JGroupsTransport) t;
        Channel c = jt.getChannel();
        if (a instanceof JGroupsAddress) {
            JGroupsAddress ja = (JGroupsAddress) a;
            org.jgroups.Address phys = (org.jgroups.Address) c
                    .down(new Event(Event.GET_PHYSICAL_ADDRESS, ja
                            .getJGroupsAddress()));
            if (phys instanceof org.jgroups.stack.IpAddress) {
                InetAddress bindAddress = ((org.jgroups.stack.IpAddress) phys)
                        .getIpAddress();
                return bindAddress;
            }
        }
    }
    return null;
}
 

import org.infinispan.remoting.transport.Address; //导入依赖的package包/类
@Override
public List<InetAddress> getClusteredControllers() {
    EmbeddedCacheManager manager = this.cm;
    if (manager == null) {
        return null;
    }
    List<Address> controllers = manager.getMembers();
    if ((controllers == null) || controllers.size() == 0) {
        return null;
    }

    List<InetAddress> clusteredControllers = new ArrayList<InetAddress>();
    for (Address a : controllers) {
        InetAddress inetAddress = addressToInetAddress(a);
        if (inetAddress != null
                && !inetAddress.getHostAddress().equals(loopbackAddress)) {
            clusteredControllers.add(inetAddress);
        }
    }
    return clusteredControllers;
}
 

import org.infinispan.remoting.transport.Address; //导入依赖的package包/类
public <K> Map<K, List<Address>> getKeysAddresses(Cache<K, Object> cache) {
    DistributionManager distributionManager = cache.getAdvancedCache().getDistributionManager();
    Map<K, List<Address>> response = new HashMap<>();
    for(K k : cache.keySet()) {
        response.put(k, distributionManager.locate(k));
    }
    return response;
}
 

import org.infinispan.remoting.transport.Address; //导入依赖的package包/类
@Override
public boolean execute(UI console, Iterator<String> args) throws IllegalParametersException {
    if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("Start execute command 'sessions'");
    }

    Cache<String, Object> sessionCache = hacep.getSessionCache();
    Map<Address, List<SessionData>> sessions = new HashMap<>();
    hacep.getCacheManager().getMembers().forEach(a -> sessions.put(a, new ArrayList<>()));
    for (Map.Entry<String, List<Address>> entry : jdgUtility.getKeysAddresses(sessionCache).entrySet()) {
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("Key [" + entry.getKey() + "] List{" + entry.getValue() + "}");
        }
        List<Address> addresses = entry.getValue() != null ? entry.getValue() : Collections.emptyList();
        for (int i = 0; i < addresses.size(); i++) {
            boolean isPrimary = (i == 0);
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("Key [" + entry.getKey() + "] Address{" + addresses.get(i) + "] isPrimary [" + isPrimary + "]");
            }
            sessions.compute(addresses.get(i), (a, l) -> {
                SessionData object = new SessionData(entry.getKey().toString(), isPrimary ? NodeType.PRIMARY : NodeType.REPLICA);
                l.add(object);
                return l;
            });
        }
    }

    console.print(sessions.entrySet().stream()
            .map(e -> new HACEPNode(e.getKey().toString(), e.getValue()))
            .collect(Collectors.toList()));
    return true;
}
 

import org.infinispan.remoting.transport.Address; //导入依赖的package包/类
/**
 * Simple messaging test.
 * 
 * @throws IOException
 * @throws InterruptedException
 * @throws ExecutionException
 * @throws TimeoutException
 */
@Test
public void testMessage() throws IOException, InterruptedException, ExecutionException, TimeoutException {
	final int nodeNumber = 3;
	
	// start test instance
	EmbeddedCacheManager cacheManager = new TestCacheManagerBuilder("node main", null).build();
	this.cacheManager = cacheManager;
	
	final Object monitor = new Object();
	final Map<Address, String> messages = Maps.newConcurrentMap();
	
	cacheManager.start();
	cacheManager.getCache(TestConstants.CACHE_DEFAULT).addListener(new NodeMessageListener<String>() {
		@Override
		public void onMessage(Message<String> value) {
			messages.put(value.getAddress(), value.getMessage());
			synchronized (monitor) {
				monitor.notify();
			}
		}
	});
	
	prepareCluster(nodeNumber, SimpleMessagingTask.class);
	
	// wait for n (n=nodeNumber) messages
	Callable<Boolean> test = new Callable<Boolean>() {
		@Override
		public Boolean call() throws Exception {
			return messages.keySet().size() >= nodeNumber;
		}
	};
	waitForEvent(monitor, test, 10, TimeUnit.SECONDS);
}
 

import org.infinispan.remoting.transport.Address; //导入依赖的package包/类
@Override
public void runTask() {
	// wait for a message from coordinator to try to lock keys A and B
	// post a message TimeoutException if lock cannot be acquired
	// post a message LockSuccess if lock is successful
	getCacheManager().getCache(TestConstants.CACHE_MESSAGE).addListener(new NodeMessageListener<String>() {
		@Override
		public void onMessage(Message<String> value) {
			LOGGER.debug("message received");
			if (value.getAddress().equals(getCacheManager().getCoordinator())) {
				LOGGER.debug("message received from coordinator");
				if ("tryLock".equals(value.getMessage())) {
					LOGGER.debug("received tryLock message");
					AdvancedCache<String, Object> cache =
							getCacheManager().<String, Object>getCache(TestConstants.CACHE_DEFAULT).getAdvancedCache();
					LOGGER.debug("batch started");
					cache.startBatch();
					boolean successful = false;
					Address address = getCacheManager().getAddress();
					try {
						LOGGER.debug("try A, B locks");
						cache.lock("A", "B");
						LOGGER.debug("lock successful");
						successful = true;
						LOGGER.debug("send success message");
						getCacheManager().<String, Message<String>>getCache(TestConstants.CACHE_MESSAGE)
							.put("messageBus", Message.from(address, LOCK_SUCCESS));
					} catch (TimeoutException e) {
						LOGGER.debug("send failure message");
						getCacheManager().<String, Message<String>>getCache(TestConstants.CACHE_MESSAGE)
							.put("messageBus", Message.from(address, TIMEOUT_EXCEPTION));
					} finally {
						cache.endBatch(successful);
						LOGGER.debug("batch ended");
					}
				}
			}
		}
	});
}
 

import org.infinispan.remoting.transport.Address; //导入依赖的package包/类
/**
 * Triggered when a view is changed, signaling that a member has joined or dropped from the cluster.
 * @param event change details
 */
@ViewChanged
public void viewChanged(ViewChangedEvent event) {
    List<Address> dropped = new ArrayList<Address>(event.getOldMembers());
    dropped.removeAll(event.getNewMembers());
    for (Address addr : dropped) {
        dropAllServices(addr);
    }
}
 

import org.infinispan.remoting.transport.Address; //导入依赖的package包/类
void dropAllServices(Address address) {
    for (String node : _serviceCache.keySet()) {
        if (node.endsWith("/" + address.toString())) {
            _serviceCache.remove(node);
        }
    }
}
 

import org.infinispan.remoting.transport.Address; //导入依赖的package包/类
@Override
public int compareTo(Address address) {
    if (address == null) {
        return -1;
    }
    return _address.compareTo(address.toString());
}
 

import org.infinispan.remoting.transport.Address; //导入依赖的package包/类
@Override
public void accept(Address address, @Nullable Optional<V> value,
        @Nullable Throwable throwable) {
    if (throwable != null) {
        logger.warn("received error from {}: {}", address, throwable.getMessage(),
                throwable);
        logStackTrace = true;
        return;
    }
    // value is only null when throwable is not null
    if (checkNotNull(value).isPresent()) {
        values.add(value.get());
    }
}
 

import org.infinispan.remoting.transport.Address; //导入依赖的package包/类
@ViewChanged
public void viewChanged(ViewChangedEvent event) {
    log.debugf("view changed: %s", event);
    List<Address> old = new ArrayList<>(event.getOldMembers());
    old.removeAll(event.getNewMembers());
    for (Address address : old) {
        locksCache.remove(key, address.toString());
    }
}
 

import org.infinispan.remoting.transport.Address; //导入依赖的package包/类
public Set<String> primaryValuesFromKeys(Cache<Key, Object> cache) {
    DistributionManager distributionManager = cache.getAdvancedCache().getDistributionManager();
    Address address = cache.getCacheManager().getAddress();
    return valuesFromKeys(cache, k -> distributionManager.getPrimaryLocation(k).equals(address));
}
 

import org.infinispan.remoting.transport.Address; //导入依赖的package包/类
public Set<String> replicaValuesFromKeys(Cache<Key, Object> cache) {
    DistributionManager distributionManager = cache.getAdvancedCache().getDistributionManager();
    Address address = cache.getCacheManager().getAddress();
    return valuesFromKeys(cache, k -> distributionManager.getLocality(k).isLocal() && !distributionManager.getPrimaryLocation(k).equals(address));
}
 

import org.infinispan.remoting.transport.Address; //导入依赖的package包/类
@Override
public List<String> getNodes() {
  return cacheManager.getTransport().getMembers().stream().map(Address::toString).collect(toList());
}
 

import org.infinispan.remoting.transport.Address; //导入依赖的package包/类
public Address getAddress() {
	return address;
}
 

import org.infinispan.remoting.transport.Address; //导入依赖的package包/类
public void setAddress(Address address) {
	this.address = address;
}
 

import org.infinispan.remoting.transport.Address; //导入依赖的package包/类
public static <M extends Serializable> Message<M> from(Address address, M message) {
	Message<M> m = new Message<M>();
	m.address = address;
	m.message = message;
	return m;
}
 

import org.infinispan.remoting.transport.Address; //导入依赖的package包/类
/**
 * Simple messaging test with other nodes' restart.
 * 
 * @throws IOException
 * @throws InterruptedException
 * @throws ExecutionException
 * @throws TimeoutException
 */
@Test
public void testMessage() throws IOException, InterruptedException, ExecutionException, TimeoutException {
	final int nodeNumber = 3;
	
	// start test instance
	final EmbeddedCacheManager cacheManager = new TestCacheManagerBuilder("node main", null).build();
	this.cacheManager = cacheManager;
	
	final Object monitor = new Object();
	final Map<Address, String> messages = Maps.newConcurrentMap();
	
	cacheManager.start();
	cacheManager.getCache(TestConstants.CACHE_DEFAULT).addListener(new NodeMessageListener<String>() {
		@Override
		public void onMessage(Message<String> value) {
			messages.put(value.getAddress(), value.getMessage());
			synchronized (monitor) {
				monitor.notify();
			}
		}
	});
	
	// SimpleMessagingTask: on connect, each node send a message
	prepareCluster(nodeNumber, SimpleMessagingTask.class);
	
	// wait for n (n=nodeNumber) messages
	Callable<Boolean> testOne = new Callable<Boolean>() {
		@Override
		public Boolean call() throws Exception {
			return messages.keySet().size() == nodeNumber;
		}
	};
	// wait for messages from other nodes
	waitForEvent(monitor, testOne, 10, TimeUnit.SECONDS);
	
	// shutdown all nodes
	shutdownProcesses(false);
	
	// wait alone state (1 node)
	Callable<Boolean> aloneTest = new Callable<Boolean>() {
		@Override
		public Boolean call() throws Exception {
			return cacheManager.getMembers().size() == 1;
		}
	};
	waitForEvent(monitor, aloneTest, 20, TimeUnit.SECONDS);
	
	// start new nodes
	prepareCluster(nodeNumber, SimpleMessagingTask.class);
	
	// wait joining nodes
	Callable<Boolean> allTest = new Callable<Boolean>() {
		@Override
		public Boolean call() throws Exception {
			return cacheManager.getMembers().size() == 4;
		}
	};
	waitForEvent(monitor, allTest, 20, TimeUnit.SECONDS);
	
	// wait 6 messages (as new nodes use new addresses, new messages are added)
	Callable<Boolean> testTwo = new Callable<Boolean>() {
		@Override
		public Boolean call() throws Exception {
			return messages.keySet().size() == 6;
		}
	};
	waitForEvent(monitor, testTwo, 10, TimeUnit.SECONDS);
}
 

import org.infinispan.remoting.transport.Address; //导入依赖的package包/类
/**
 * Test {@link AdvancedCache#startBatch()} (to handle concurrent read / update).
 * We lock to keys (A and B) and asks for second nodes (using message) to confirm that these keys cannot be locked
 * concurrently.
 * After keys' release, confirm that second node can now lock these keys.
 * 
 * @throws IOException
 * @throws InterruptedException
 * @throws TimeoutException
 * @throws ExecutionException
 */
@Test
public void testBatch() throws IOException, InterruptedException, TimeoutException, ExecutionException {
	// start infinispan
	final int nodeNumber = 1;
	
	// start test instance
	final EmbeddedCacheManager cacheManager = new TestCacheManagerBuilder("node main", null).build();
	this.cacheManager = cacheManager;
	
	final Object monitor = new Object();
	final Map<Address, String> messages = Maps.newConcurrentMap();
	
	cacheManager.start();
	cacheManager.getCache(TestConstants.CACHE_MESSAGE).addListener(new NodeMessageListener<String>() {
		@Override
		public void onMessage(Message<String> value) {
			messages.put(value.getAddress(), value.getMessage());
			synchronized (monitor) {
				monitor.notify();
			}
		}
	});
	// initializes two keys A and B
	cacheManager.getCache(TestConstants.CACHE_DEFAULT).put("A", "A");
	cacheManager.getCache(TestConstants.CACHE_DEFAULT).put("B", "B");
	
	// start another node that register a task that waits a message to lock A and B
	prepareCluster(nodeNumber, LockTask.class);
	
	// startBatch
	AdvancedCache<Object, Object> cache = cacheManager.<Object, Object>getCache(TestConstants.CACHE_DEFAULT).getAdvancedCache();
	cache.startBatch();
	boolean successful = false;
	final Address address = cacheManager.getAddress();
	
	Callable<Boolean> test = new Callable<Boolean>() {
		@Override
		public Boolean call() throws Exception {
			return ! address.equals(cacheManager.<String, Message<String>>getCache(TestConstants.CACHE_MESSAGE).get(TestConstants.CACHE_KEY_MESSAGE_BUS).getAddress());
		}
	};
	
	try {
		// lock A and B
		cache.lock("A", "B");
		// send message -> second node cannot lock -> message TimeoutException
		cacheManager.getCache(TestConstants.CACHE_MESSAGE).put(TestConstants.CACHE_KEY_MESSAGE_BUS, Message.from(address, LockTask.TRY_LOCK));
		// must be set accordingly with lockAcquisitionTimeout
		waitForEvent(monitor, test, 25, TimeUnit.SECONDS);
		Assert.assertEquals(LockTask.TIMEOUT_EXCEPTION, cacheManager.<String, Message<String>>getCache(TestConstants.CACHE_MESSAGE).get(TestConstants.CACHE_KEY_MESSAGE_BUS).getMessage());
		successful = true;
	} finally {
		// stopBatch
		cache.endBatch(successful);
	}
	// second node can lock
	cacheManager.getCache(TestConstants.CACHE_MESSAGE).put(TestConstants.CACHE_KEY_MESSAGE_BUS, Message.from(address, LockTask.TRY_LOCK));
	// must be set accordingly with lockAcquisitionTimeout
	waitForEvent(monitor, test, 25, TimeUnit.SECONDS);
	Assert.assertEquals(LockTask.LOCK_SUCCESS, cacheManager.<String, Message<String>>getCache(TestConstants.CACHE_MESSAGE).get(TestConstants.CACHE_KEY_MESSAGE_BUS).getMessage());
}