ClusterExecutorImpl implements ClusterExecutor { @Override public FutureClusterResponses execute(ClusterRequest clusterRequest) { Set<String> clusterNodeIds = new HashSet<>(); if (clusterRequest.isMulticast()) { clusterNodeIds = new HashSet<>(_clusterNodeStatuses.keySet()); if (clusterRequest.isSkipLocal()) { clusterNodeIds.remove( _localClusterNodeStatus.getClusterNodeId()); } } else { clusterNodeIds.addAll(clusterRequest.getTargetClusterNodeIds()); } FutureClusterResponses futureClusterResponses = new FutureClusterResponses(clusterNodeIds); if (!clusterRequest.isFireAndForget()) { _futureClusterResponses.put( clusterRequest.getUuid(), futureClusterResponses); } if (clusterNodeIds.remove(_localClusterNodeStatus.getClusterNodeId())) { ClusterNodeResponse clusterNodeResponse = executeClusterRequest( clusterRequest); if (!clusterRequest.isFireAndForget()) { futureClusterResponses.addClusterNodeResponse( clusterNodeResponse); } } if (clusterRequest.isMulticast()) { _clusterChannel.sendMulticastMessage(clusterRequest); } else { for (String clusterNodeId : clusterNodeIds) { ClusterNodeStatus clusterNodeStatus = _clusterNodeStatuses.get( clusterNodeId); if (clusterNodeStatus == null) { if (_log.isWarnEnabled()) { _log.warn( StringBundler.concat( "Unable to get cluster node ", clusterNodeId, " while executing ", clusterRequest)); } continue; } _clusterChannel.sendUnicastMessage( clusterRequest, clusterNodeStatus.getAddress()); } } return futureClusterResponses; }  @Override @Reference( cardinality = ReferenceCardinality.MULTIPLE, policy = ReferencePolicy.DYNAMIC, policyOption = ReferencePolicyOption.GREEDY ) void addClusterEventListener(
		ClusterEventListener clusterEventListener); @Override FutureClusterResponses execute(ClusterRequest clusterRequest); @Override InetAddress getBindInetAddress(); @Override NetworkInterface getBindNetworkInterface(); @Override List<ClusterEventListener> getClusterEventListeners(); @Override List<ClusterNode> getClusterNodes(); @Override ClusterNode getLocalClusterNode(); @Override boolean isClusterNodeAlive(String clusterNodeId); @Override boolean isEnabled(); @Override void removeClusterEventListener(
		ClusterEventListener clusterEventListener);  }
@Test public void testExecute() throws Exception { ClusterExecutorImpl clusterExecutorImpl = getClusterExecutorImpl(); TestClusterChannel.clearAllMessages(); List<Serializable> multicastMessages = TestClusterChannel.getMulticastMessages(); List<ObjectValuePair<Serializable, Address>> unicastMessages = TestClusterChannel.getUnicastMessages(); Assert.assertTrue( multicastMessages.toString(), multicastMessages.isEmpty()); Assert.assertTrue( unicastMessages.toString(), unicastMessages.isEmpty()); ClusterRequest clusterRequest = ClusterRequest.createMulticastRequest( StringPool.BLANK); FutureClusterResponses futureClusterResponses = clusterExecutorImpl.execute(clusterRequest); Assert.assertEquals( multicastMessages.toString(), 1, multicastMessages.size()); Assert.assertTrue( multicastMessages.toString(), multicastMessages.contains(clusterRequest)); Assert.assertTrue( unicastMessages.toString(), unicastMessages.isEmpty()); ClusterNodeResponses clusterNodeResponses = futureClusterResponses.get(); Assert.assertEquals(1, clusterNodeResponses.size()); TestClusterChannel.clearAllMessages(); Assert.assertTrue( multicastMessages.toString(), multicastMessages.isEmpty()); Assert.assertTrue( unicastMessages.toString(), unicastMessages.isEmpty()); clusterRequest = ClusterRequest.createMulticastRequest( StringPool.BLANK, true); futureClusterResponses = clusterExecutorImpl.execute(clusterRequest); Assert.assertEquals( multicastMessages.toString(), 1, multicastMessages.size()); Assert.assertTrue( multicastMessages.toString(), multicastMessages.contains(clusterRequest)); Assert.assertTrue( unicastMessages.toString(), unicastMessages.isEmpty()); clusterNodeResponses = futureClusterResponses.get(); Assert.assertEquals(0, clusterNodeResponses.size()); TestClusterChannel.clearAllMessages(); Assert.assertTrue( multicastMessages.toString(), multicastMessages.isEmpty()); Assert.assertTrue( unicastMessages.toString(), unicastMessages.isEmpty()); ClusterNode localClusterNode = clusterExecutorImpl.getLocalClusterNode(); clusterRequest = ClusterRequest.createUnicastRequest( clusterRequest, localClusterNode.getClusterNodeId()); futureClusterResponses = clusterExecutorImpl.execute(clusterRequest); Assert.assertTrue( multicastMessages.toString(), multicastMessages.isEmpty()); Assert.assertTrue( unicastMessages.toString(), unicastMessages.isEmpty()); clusterNodeResponses = futureClusterResponses.get(); Assert.assertEquals(1, clusterNodeResponses.size()); TestClusterChannel.clearAllMessages(); Assert.assertTrue( multicastMessages.toString(), multicastMessages.isEmpty()); Assert.assertTrue( unicastMessages.toString(), unicastMessages.isEmpty()); ClusterExecutorImpl newClusterExecutorImpl = getClusterExecutorImpl(); Assert.assertEquals( multicastMessages.toString(), 1, multicastMessages.size()); Assert.assertTrue( unicastMessages.toString(), unicastMessages.isEmpty()); Serializable serializable = multicastMessages.get(0); clusterExecutorImpl.handleReceivedClusterRequest( (ClusterRequest)serializable); TestClusterChannel.clearAllMessages(); ClusterNode newClusterNode = newClusterExecutorImpl.getLocalClusterNode(); clusterRequest = ClusterRequest.createUnicastRequest( StringPool.BLANK, newClusterNode.getClusterNodeId()); clusterExecutorImpl.execute(clusterRequest); Assert.assertTrue( multicastMessages.toString(), multicastMessages.isEmpty()); Assert.assertEquals( unicastMessages.toString(), 1, unicastMessages.size()); ObjectValuePair<Serializable, Address> receivedMessage = unicastMessages.get(0); Assert.assertEquals(clusterRequest, receivedMessage.getKey()); }