Coverage Report

Coverage Report - org.kuali.rice.ksb.impl.bus.ServiceBusImpl

Classes in this File

Line Coverage

Branch Coverage

Complexity

ServiceBusImpl

0/260

0/136

3.529

ServiceBusImpl$1

0/6

N/A

3.529

 package org.kuali.rice.ksb.impl.bus;
 
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Random;
 import java.util.Set;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 
 import javax.xml.namespace.QName;
 
 import org.apache.commons.lang.StringUtils;
 import org.apache.log4j.Logger;
 import org.kuali.rice.core.api.config.property.Config;
 import org.kuali.rice.core.api.config.property.ConfigContext;
 import org.kuali.rice.core.api.lifecycle.BaseLifecycle;
 import org.kuali.rice.ksb.api.bus.Endpoint;
 import org.kuali.rice.ksb.api.bus.ServiceBus;
 import org.kuali.rice.ksb.api.bus.ServiceConfiguration;
 import org.kuali.rice.ksb.api.bus.ServiceDefinition;
 import org.kuali.rice.ksb.api.registry.RemoveAndPublishResult;
 import org.kuali.rice.ksb.api.registry.ServiceEndpoint;
 import org.kuali.rice.ksb.api.registry.ServiceInfo;
 import org.kuali.rice.ksb.api.registry.ServiceRegistry;
 import org.kuali.rice.ksb.impl.registry.diff.CompleteServiceDiff;
 import org.kuali.rice.ksb.impl.registry.diff.LocalServicesDiff;
 import org.kuali.rice.ksb.impl.registry.diff.RemoteServicesDiff;
 import org.kuali.rice.ksb.impl.registry.diff.ServiceRegistryDiffCalculator;
 import org.kuali.rice.ksb.messaging.serviceexporters.ServiceExportManager;
 import org.kuali.rice.ksb.messaging.threadpool.KSBScheduledPool;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
 
 public class ServiceBusImpl extends BaseLifecycle implements ServiceBus, InitializingBean, DisposableBean {
         
         private static final Logger LOG = Logger.getLogger(ServiceBusImpl.class);
         
         private final Object serviceLock = new Object();
         private final Object synchronizeLock = new Object();
         private final Random randomNumber = new Random();
         
         // injected values
         private String instanceId;
         private ServiceRegistry serviceRegistry;
         private ServiceRegistryDiffCalculator diffCalculator;
         private ServiceExportManager serviceExportManager;
         private KSBScheduledPool scheduledPool;
         
         private ScheduledFuture<?> registrySyncFuture;
         
         /**
          * Contains endpoints for services which were published by this client application.
          */
         private final Map<QName, LocalService> localServices;
         
         /**
          * Contains endpoints for services which exist remotely.  This list may not be
          * entirely complete as entries get lazily loaded into it as services are requested.
          */
         private final Map<QName, Set<RemoteService>> clientRegistryCache;
                 
         public ServiceBusImpl() {
                 this.localServices = new HashMap<QName, LocalService>();
                 this.clientRegistryCache = new HashMap<QName, Set<RemoteService>>();
         }
         
         @Override
         public void afterPropertiesSet() throws Exception {
                 if (StringUtils.isBlank(instanceId)) {
                         throw new IllegalStateException("a valid instanceId was not injected");
                 }
                 if (serviceRegistry == null) {
                         throw new IllegalStateException("serviceRegistry was not injected");
                 }
                 if (diffCalculator == null) {
                         throw new IllegalStateException("diffCalculator was not injected");
                 }
                 if (scheduledPool == null) {
                         throw new IllegalStateException("scheduledPool was not injected");
                 }
         }
         
         @Override
         public void start() throws Exception {
                 startSynchronizationThread();
                 super.start();
         }
                 
         protected boolean isDevMode() {
                 return ConfigContext.getCurrentContextConfig().getDevMode();
         }
 
         protected void startSynchronizationThread() {
                 synchronized (synchronizeLock) {
                         LOG.info("Starting Service Bus synchronization thread...");
                         if (!isDevMode()) {
                                 int refreshRate = ConfigContext.getCurrentContextConfig().getRefreshRate();
                                 Runnable runnable = new Runnable() {
                                         public void run() {
                                                 try {
                                                         synchronize();
                                                 } catch (Throwable t) {
                                                         LOG.error("Failed to execute background service bus synchronization.", t);
                                                 }
                                         }
                                 };
                                 this.registrySyncFuture = scheduledPool.scheduleWithFixedDelay(runnable, 30, refreshRate, TimeUnit.SECONDS);
                         }
                         LOG.info("...Service Bus synchronization thread successfully started.");
                 }
         }
         
         @Override
         public void destroy() throws Exception {
                 LOG.info("Stopping the Service Bus...");
                 stopSynchronizationThread();
                 serviceRegistry.takeInstanceOffline(getInstanceId());
                 LOG.info("...Service Bus successfully stopped.");
         }
         
         protected void stopSynchronizationThread() {
                 synchronized (synchronizeLock) {
                         // remove services from the bus
                         if (this.registrySyncFuture != null) {
                                 if (!this.registrySyncFuture.cancel(false)) {
                                         LOG.warn("Failed to cancel registry sychronization.");
                                 }
                                 this.registrySyncFuture = null;
                         }
                 }
         }
 
         @Override
         public String getInstanceId() {
                 return this.instanceId;
         }
         
         public void setInstanceId(String instanceId) {
                 this.instanceId = instanceId;
         }
         
         @Override
         public List<Endpoint> getEndpoints(QName serviceName) {
                 if (serviceName == null) {
                         throw new IllegalArgumentException("serviceName cannot be null");
                 }
                 List<Endpoint> endpoints = new ArrayList<Endpoint>();
                 synchronized (serviceLock) {
                         endpoints.addAll(getRemoteEndpoints(serviceName));
                         Endpoint localEndpoint = getLocalEndpoint(serviceName);
                         if (localEndpoint != null) {
                                 for (Iterator<Endpoint> iterator = endpoints.iterator(); iterator.hasNext();) {
                                         Endpoint endpoint = (Endpoint) iterator.next();
                                         if (localEndpoint.getServiceConfiguration().equals(endpoint.getServiceConfiguration())) {
                                                 iterator.remove();
                                                 break;
                                         }
                                 }
                                 // add at first position, just because we like the local endpoint the best, it's our friend ;)
                                 endpoints.add(0, localEndpoint);
                         }
                 }
                 return Collections.unmodifiableList(endpoints);
         }
         
         @Override
         public List<Endpoint> getRemoteEndpoints(QName serviceName) {
                 if (serviceName == null) {
                         throw new IllegalArgumentException("serviceName cannot be null");
                 }
                 List<Endpoint> endpoints = new ArrayList<Endpoint>();
                 synchronized (serviceLock) {
                         Set<RemoteService> remoteServices = clientRegistryCache.get(serviceName);
                         if (remoteServices != null) {
                                 for (RemoteService remoteService : remoteServices) {
                                         endpoints.add(remoteService.getEndpoint());
                                 }
                         }
                 }
                 return Collections.unmodifiableList(endpoints);
         }
 
         @Override
         public Endpoint getLocalEndpoint(QName serviceName) {
                 if (serviceName == null) {
                         throw new IllegalArgumentException("serviceName cannot be null");
                 }
                 synchronized (serviceLock) {
                         LocalService localService = localServices.get(serviceName);
                         if (localService != null) {
                                 return localService.getEndpoint();
                         }
                         return null;
                 }
         }
 
         @Override
         public Map<QName, Endpoint> getLocalEndpoints() {
                 Map<QName, Endpoint> localEndpoints = new HashMap<QName, Endpoint>();
                 synchronized (serviceLock) {
                         for (QName localServiceName : localServices.keySet()) {
                                 LocalService localService = localServices.get(localServiceName);
                                 localEndpoints.put(localServiceName, localService.getEndpoint());
                         }
                 }
                 return Collections.unmodifiableMap(localEndpoints);
         }
 
         @Override
         public List<Endpoint> getAllEndpoints() {
                 List<Endpoint> allEndpoints = new ArrayList<Endpoint>();
                 synchronized (serviceLock) {
                         for (QName serviceName : this.localServices.keySet()) {
                                 allEndpoints.add(this.localServices.get(serviceName).getEndpoint());
                         }
                         for (QName serviceName : this.clientRegistryCache.keySet()) {
                                 Set<RemoteService> remoteServices = clientRegistryCache.get(serviceName);
                                 for (RemoteService remoteService : remoteServices) {
                                         allEndpoints.add(remoteService.getEndpoint());
                                 }
                         }
                 }
                 return Collections.unmodifiableList(allEndpoints);
         }
 
         @Override
         public Endpoint getEndpoint(QName serviceName) {
                 return getEndpoint(serviceName, null);
         }
         
         @Override
     public Endpoint getEndpoint(QName serviceName, String applicationId) {
         if (serviceName == null) {
             throw new IllegalArgumentException("serviceName cannot be null");
         }
         Endpoint availableEndpoint = null;
         synchronized (serviceLock) {
             // look at local services first
             availableEndpoint = getLocalEndpoint(serviceName);
             if (availableEndpoint == null || (!StringUtils.isBlank(applicationId) && !availableEndpoint.getServiceConfiguration().getApplicationId().equals(applicationId))) {
                  // TODO - would be better to return an Endpoint that contained an internal proxy to all the services so fail-over would be easier to implement!
                 Set<RemoteService> remoteServices = clientRegistryCache.get(serviceName);
                 remoteServices = filterByApplicationId(applicationId, remoteServices);
                 if (remoteServices != null && !remoteServices.isEmpty()) {
                     // TODO - this should also probably check the current status of the service?
                     RemoteService[] remoteServiceArray = remoteServices.toArray(new RemoteService[0]);
                     RemoteService availableRemoteService = remoteServiceArray[this.randomNumber.nextInt(remoteServiceArray.length)];
                     availableEndpoint = availableRemoteService.getEndpoint();
                 }
             }
         }
         return availableEndpoint;
     }
         
         protected Set<RemoteService> filterByApplicationId(String applicationId, Set<RemoteService> remoteServices) {
             if (StringUtils.isBlank(applicationId) || remoteServices == null || remoteServices.isEmpty()) {
                 return remoteServices;
             }
             Set<RemoteService> filtered = new HashSet<RemoteService>();
             for (RemoteService remoteService : remoteServices) {
                 if (remoteService.getServiceInfo().getApplicationId().equals(applicationId)) {
                     filtered.add(remoteService);
                 }
             }
             return filtered;
         }
         
         @Override
         public Endpoint getConfiguredEndpoint(ServiceConfiguration serviceConfiguration) {
                 if (serviceConfiguration == null) {
                         throw new IllegalArgumentException("serviceConfiguration cannot be null");
                 }
                 synchronized (serviceLock) {
                         Endpoint localEndpoint = getLocalEndpoint(serviceConfiguration.getServiceName());
                         if (localEndpoint != null && localEndpoint.getServiceConfiguration().equals(serviceConfiguration)) {
                                 return localEndpoint;
                         }
                         List<Endpoint> remoteEndpoints = getRemoteEndpoints(serviceConfiguration.getServiceName());
                         for (Endpoint remoteEndpoint : remoteEndpoints) {
                                 if (remoteEndpoint.getServiceConfiguration().equals(serviceConfiguration)) {
                                         return remoteEndpoint;
                                 }
                         }
                 }
                 return null;
         }
 
         @Override
     public Object getService(QName serviceName) {
         return getService(serviceName, null);
     }
         
         @Override
         public Object getService(QName serviceName, String applicationId) {
                 Endpoint availableEndpoint = getEndpoint(serviceName, applicationId);
                 if (availableEndpoint == null) {
                         return null;
                 }
                 return availableEndpoint.getService();
         }
 
         @Override
         public ServiceConfiguration publishService(ServiceDefinition serviceDefinition, boolean synchronize) {
                 if (serviceDefinition == null) {
                         throw new IllegalArgumentException("serviceDefinition cannot be null");
                 }
                 LocalService localService = new LocalService(getInstanceId(), serviceDefinition);
                 synchronized (serviceLock) {
                         serviceExportManager.exportService(serviceDefinition);
                         localServices.put(serviceDefinition.getServiceName(), localService);
                 }
                 if (synchronize) {
                         synchronize();
                 }
                 return localService.getEndpoint().getServiceConfiguration();
         }
 
         @Override
         public List<ServiceConfiguration> publishServices(List<ServiceDefinition> serviceDefinitions, boolean synchronize) {
                 if (serviceDefinitions == null) {
                         throw new IllegalArgumentException("serviceDefinitions list cannot be null");
                 }
                 List<ServiceConfiguration> serviceConfigurations = new ArrayList<ServiceConfiguration>();
                 synchronized (serviceLock) {
                         for (ServiceDefinition serviceDefinition : serviceDefinitions) {
                                 ServiceConfiguration serviceConfiguration = publishService(serviceDefinition, false);
                                 serviceConfigurations.add(serviceConfiguration);
                         }
                 }
                 if (synchronize) {
                         synchronize();
                 }
                 return Collections.unmodifiableList(serviceConfigurations);
         }
 
         @Override
         public boolean removeService(QName serviceName, boolean synchronize) {
                 if (serviceName == null) {
                         throw new IllegalArgumentException("serviceName cannot be null");
                 }
                 boolean serviceRemoved = false;
                 synchronized (serviceLock) {
                         LocalService localService = localServices.remove(serviceName);
                         serviceRemoved = localService != null;
                         serviceExportManager.removeService(serviceName);
                 }
                 if (serviceRemoved && synchronize) {
                         synchronize();
                 }
                 return serviceRemoved;
         }
 
         @Override
         public List<Boolean> removeServices(List<QName> serviceNames, boolean synchronize) {
                 if (serviceNames == null) {
                         throw new IllegalArgumentException("serviceNames cannot be null");
                 }
                 boolean serviceRemoved = false;
                 List<Boolean> servicesRemoved = new ArrayList<Boolean>();
                 synchronized (serviceLock) {
                         for (QName serviceName : serviceNames) {
                                 serviceExportManager.removeService(serviceName);
                                 LocalService localService = localServices.remove(serviceName);
                                 if (localService != null) {
                                         servicesRemoved.add(Boolean.TRUE);
                                         serviceRemoved = true;
                                 } else {
                                         servicesRemoved.add(Boolean.FALSE);
                                 }
                         }
                 }
                 if (serviceRemoved && synchronize) {
                         synchronize();
                 }
                 return servicesRemoved;
         }
 
         @Override
         public void synchronize() {
                 if (!isDevMode() && isStarted()) {
                         synchronized (synchronizeLock) {
                                 List<LocalService> localServicesList;
                                 List<RemoteService> clientRegistryCacheList;
                                 synchronized (serviceLock) {
                                         // first, flatten the lists
                                         localServicesList = new ArrayList<LocalService>(this.localServices.values());
                                         clientRegistryCacheList = new ArrayList<RemoteService>();
                                         for (Set<RemoteService> remoteServices : this.clientRegistryCache.values()) {
                                                 clientRegistryCacheList.addAll(remoteServices);
                                         }
                                 }
                                 CompleteServiceDiff serviceDiff = diffCalculator.diffServices(getInstanceId(), localServicesList, clientRegistryCacheList);
                         
                                 RemoteServicesDiff remoteServicesDiff = serviceDiff.getRemoteServicesDiff();
                                 processRemoteServiceDiff(remoteServicesDiff);
                         
                                 LocalServicesDiff localServicesDiff = serviceDiff.getLocalServicesDiff();
                                 processLocalServiceDiff(localServicesDiff);
                         }
                 }
         }
                 
         protected void processRemoteServiceDiff(RemoteServicesDiff remoteServicesDiff) {
                 // note that since there is a gap between when the original services are acquired, the diff, and this subsequent critical section
                 // the list of local and client registry services could have changed, so that needs to be considered in the remaining code
                 synchronized (serviceLock) {
                         // first, let's update what we know about the remote services
                         List<RemoteService> removedServices = remoteServicesDiff.getRemovedServices();
                         for (RemoteService removedRemoteService : removedServices) {
                                 Set<RemoteService> remoteServiceSet = this.clientRegistryCache.get(removedRemoteService.getServiceName());
                                 if (remoteServiceSet != null) {
                                         boolean wasRemoved = remoteServiceSet.remove(removedRemoteService);
                                         if (!wasRemoved) {
                                                 LOG.warn("Failed to remove remoteService during synchronization: " + removedRemoteService);
                                         }
                                 }
                         }
                         List<ServiceInfo> newServices = remoteServicesDiff.getNewServices();
                         for (ServiceInfo newService : newServices) {
                                 Set<RemoteService> remoteServiceSet = clientRegistryCache.get(newService.getServiceName());
                                 if (remoteServiceSet == null) {
                                         remoteServiceSet = new HashSet<RemoteService>();
                                         clientRegistryCache.put(newService.getServiceName(), remoteServiceSet);
                                 }
                                 remoteServiceSet.add(new RemoteService(newService, this.serviceRegistry));
                         }
                 }
         }
         
         protected void processLocalServiceDiff(LocalServicesDiff localServicesDiff) {
                 List<String> removeServiceEndpointIds = new ArrayList<String>();
                 List<ServiceEndpoint> publishServiceEndpoints = new ArrayList<ServiceEndpoint>();
                 for (ServiceInfo serviceToRemove : localServicesDiff.getServicesToRemoveFromRegistry()) {
                         removeServiceEndpointIds.add(serviceToRemove.getServiceId());
                 }
                 for (LocalService localService : localServicesDiff.getLocalServicesToPublish()) {
                         publishServiceEndpoints.add(localService.getServiceEndpoint());
                 }
                 for (LocalService localService : localServicesDiff.getLocalServicesToUpdate().keySet()) {
                         ServiceInfo registryServiceInfo = localServicesDiff.getLocalServicesToUpdate().get(localService);
                         publishServiceEndpoints.add(rebuildServiceEndpointForUpdate(localService.getServiceEndpoint(), registryServiceInfo));
                 }
                 boolean batchMode = ConfigContext.getCurrentContextConfig().getBooleanProperty(Config.BATCH_MODE, false);
                 if (!batchMode && (!removeServiceEndpointIds.isEmpty() || !publishServiceEndpoints.isEmpty())) {
                         RemoveAndPublishResult result = this.serviceRegistry.removeAndPublish(removeServiceEndpointIds, publishServiceEndpoints);
                         // now update the ServiceEndpoints for our local services so we can get the proper id for them
                         if (!result.getServicesPublished().isEmpty()) {
                                 synchronized (serviceLock) {
                                         for (ServiceEndpoint publishedService : result.getServicesPublished()) {
                                                 rebuildLocalServiceEndpointAfterPublishing(publishedService);
                                         }
                                 }
                         }
                 }
         }
         
         protected ServiceEndpoint rebuildServiceEndpointForUpdate(ServiceEndpoint originalEndpoint, ServiceInfo registryServiceInfo) {
                 ServiceEndpoint.Builder builder = ServiceEndpoint.Builder.create(originalEndpoint);
                 builder.getInfo().setServiceId(registryServiceInfo.getServiceId());
                 builder.getInfo().setServiceDescriptorId(registryServiceInfo.getServiceDescriptorId());
                 builder.getInfo().setVersionNumber(registryServiceInfo.getVersionNumber());
                 builder.getDescriptor().setId(registryServiceInfo.getServiceDescriptorId());
                 return builder.build();
         }
         
         protected void rebuildLocalServiceEndpointAfterPublishing(ServiceEndpoint publishedService) {
                 // verify the service is still published
                 QName serviceName = publishedService.getInfo().getServiceName();
                 if (localServices.containsKey(serviceName)) {
                         LocalService newLocalService = new LocalService(localServices.get(serviceName), publishedService);
                         localServices.put(serviceName, newLocalService);
                 }
         }
 
         public void setServiceRegistry(ServiceRegistry serviceRegistry) {
                 this.serviceRegistry = serviceRegistry;
         }
         
         public void setDiffCalculator(ServiceRegistryDiffCalculator diffCalculator) {
                 this.diffCalculator = diffCalculator;
         }
         
         public void setServiceExportManager(ServiceExportManager serviceExportManager) {
                 this.serviceExportManager = serviceExportManager;
         }
         
         public void setScheduledPool(KSBScheduledPool scheduledPool) {
                 this.scheduledPool = scheduledPool;
         }
         
 }

1		package org.kuali.rice.ksb.impl.bus;
2
3		import java.util.ArrayList;
4		import java.util.Collections;
5		import java.util.HashMap;
6		import java.util.HashSet;
7		import java.util.Iterator;
8		import java.util.List;
9		import java.util.Map;
10		import java.util.Random;
11		import java.util.Set;
12		import java.util.concurrent.ScheduledFuture;
13		import java.util.concurrent.TimeUnit;
14
15		import javax.xml.namespace.QName;
16
17		import org.apache.commons.lang.StringUtils;
18		import org.apache.log4j.Logger;
19		import org.kuali.rice.core.api.config.property.Config;
20		import org.kuali.rice.core.api.config.property.ConfigContext;
21		import org.kuali.rice.core.api.lifecycle.BaseLifecycle;
22		import org.kuali.rice.ksb.api.bus.Endpoint;
23		import org.kuali.rice.ksb.api.bus.ServiceBus;
24		import org.kuali.rice.ksb.api.bus.ServiceConfiguration;
25		import org.kuali.rice.ksb.api.bus.ServiceDefinition;
26		import org.kuali.rice.ksb.api.registry.RemoveAndPublishResult;
27		import org.kuali.rice.ksb.api.registry.ServiceEndpoint;
28		import org.kuali.rice.ksb.api.registry.ServiceInfo;
29		import org.kuali.rice.ksb.api.registry.ServiceRegistry;
30		import org.kuali.rice.ksb.impl.registry.diff.CompleteServiceDiff;
31		import org.kuali.rice.ksb.impl.registry.diff.LocalServicesDiff;
32		import org.kuali.rice.ksb.impl.registry.diff.RemoteServicesDiff;
33		import org.kuali.rice.ksb.impl.registry.diff.ServiceRegistryDiffCalculator;
34		import org.kuali.rice.ksb.messaging.serviceexporters.ServiceExportManager;
35		import org.kuali.rice.ksb.messaging.threadpool.KSBScheduledPool;
36		import org.springframework.beans.factory.DisposableBean;
37		import org.springframework.beans.factory.InitializingBean;
38
39	0	public class ServiceBusImpl extends BaseLifecycle implements ServiceBus, InitializingBean, DisposableBean {
40
41	0	private static final Logger LOG = Logger.getLogger(ServiceBusImpl.class);
42
43	0	private final Object serviceLock = new Object();
44	0	private final Object synchronizeLock = new Object();
45	0	private final Random randomNumber = new Random();
46
47		// injected values
48		private String instanceId;
49		private ServiceRegistry serviceRegistry;
50		private ServiceRegistryDiffCalculator diffCalculator;
51		private ServiceExportManager serviceExportManager;
52		private KSBScheduledPool scheduledPool;
53
54		private ScheduledFuture<?> registrySyncFuture;
55
56		/**
57		* Contains endpoints for services which were published by this client application.
58		*/
59		private final Map<QName, LocalService> localServices;
60
61		/**
62		* Contains endpoints for services which exist remotely. This list may not be
63		* entirely complete as entries get lazily loaded into it as services are requested.
64		*/
65		private final Map<QName, Set<RemoteService>> clientRegistryCache;
66
67	0	public ServiceBusImpl() {
68	0	this.localServices = new HashMap<QName, LocalService>();
69	0	this.clientRegistryCache = new HashMap<QName, Set<RemoteService>>();
70	0	}
71
72		@Override
73		public void afterPropertiesSet() throws Exception {
74	0	if (StringUtils.isBlank(instanceId)) {
75	0	throw new IllegalStateException("a valid instanceId was not injected");
76		}
77	0	if (serviceRegistry == null) {
78	0	throw new IllegalStateException("serviceRegistry was not injected");
79		}
80	0	if (diffCalculator == null) {
81	0	throw new IllegalStateException("diffCalculator was not injected");
82		}
83	0	if (scheduledPool == null) {
84	0	throw new IllegalStateException("scheduledPool was not injected");
85		}
86	0	}
87
88		@Override
89		public void start() throws Exception {
90	0	startSynchronizationThread();
91	0	super.start();
92	0	}
93
94		protected boolean isDevMode() {
95	0	return ConfigContext.getCurrentContextConfig().getDevMode();
96		}
97
98		protected void startSynchronizationThread() {
99	0	synchronized (synchronizeLock) {
100	0	LOG.info("Starting Service Bus synchronization thread...");
101	0	if (!isDevMode()) {
102	0	int refreshRate = ConfigContext.getCurrentContextConfig().getRefreshRate();
103	0	Runnable runnable = new Runnable() {
104		public void run() {
105		try {
106	0	synchronize();
107	0	} catch (Throwable t) {
108	0	LOG.error("Failed to execute background service bus synchronization.", t);
109	0	}
110	0	}
111		};
112	0	this.registrySyncFuture = scheduledPool.scheduleWithFixedDelay(runnable, 30, refreshRate, TimeUnit.SECONDS);
113		}
114	0	LOG.info("...Service Bus synchronization thread successfully started.");
115	0	}
116	0	}
117
118		@Override
119		public void destroy() throws Exception {
120	0	LOG.info("Stopping the Service Bus...");
121	0	stopSynchronizationThread();
122	0	serviceRegistry.takeInstanceOffline(getInstanceId());
123	0	LOG.info("...Service Bus successfully stopped.");
124	0	}
125
126		protected void stopSynchronizationThread() {
127	0	synchronized (synchronizeLock) {
128		// remove services from the bus
129	0	if (this.registrySyncFuture != null) {
130	0	if (!this.registrySyncFuture.cancel(false)) {
131	0	LOG.warn("Failed to cancel registry sychronization.");
132		}
133	0	this.registrySyncFuture = null;
134		}
135	0	}
136	0	}
137
138		@Override
139		public String getInstanceId() {
140	0	return this.instanceId;
141		}
142
143		public void setInstanceId(String instanceId) {
144	0	this.instanceId = instanceId;
145	0	}
146
147		@Override
148		public List<Endpoint> getEndpoints(QName serviceName) {
149	0	if (serviceName == null) {
150	0	throw new IllegalArgumentException("serviceName cannot be null");
151		}
152	0	List<Endpoint> endpoints = new ArrayList<Endpoint>();
153	0	synchronized (serviceLock) {
154	0	endpoints.addAll(getRemoteEndpoints(serviceName));
155	0	Endpoint localEndpoint = getLocalEndpoint(serviceName);
156	0	if (localEndpoint != null) {
157	0	for (Iterator<Endpoint> iterator = endpoints.iterator(); iterator.hasNext();) {
158	0	Endpoint endpoint = (Endpoint) iterator.next();
159	0	if (localEndpoint.getServiceConfiguration().equals(endpoint.getServiceConfiguration())) {
160	0	iterator.remove();
161	0	break;
162		}
163	0	}
164		// add at first position, just because we like the local endpoint the best, it's our friend ;)
165	0	endpoints.add(0, localEndpoint);
166		}
167	0	}
168	0	return Collections.unmodifiableList(endpoints);
169		}
170
171		@Override
172		public List<Endpoint> getRemoteEndpoints(QName serviceName) {
173	0	if (serviceName == null) {
174	0	throw new IllegalArgumentException("serviceName cannot be null");
175		}
176	0	List<Endpoint> endpoints = new ArrayList<Endpoint>();
177	0	synchronized (serviceLock) {
178	0	Set<RemoteService> remoteServices = clientRegistryCache.get(serviceName);
179	0	if (remoteServices != null) {
180	0	for (RemoteService remoteService : remoteServices) {
181	0	endpoints.add(remoteService.getEndpoint());
182		}
183		}
184	0	}
185	0	return Collections.unmodifiableList(endpoints);
186		}
187
188		@Override
189		public Endpoint getLocalEndpoint(QName serviceName) {
190	0	if (serviceName == null) {
191	0	throw new IllegalArgumentException("serviceName cannot be null");
192		}
193	0	synchronized (serviceLock) {
194	0	LocalService localService = localServices.get(serviceName);
195	0	if (localService != null) {
196	0	return localService.getEndpoint();
197		}
198	0	return null;
199	0	}
200		}
201
202		@Override
203		public Map<QName, Endpoint> getLocalEndpoints() {
204	0	Map<QName, Endpoint> localEndpoints = new HashMap<QName, Endpoint>();
205	0	synchronized (serviceLock) {
206	0	for (QName localServiceName : localServices.keySet()) {
207	0	LocalService localService = localServices.get(localServiceName);
208	0	localEndpoints.put(localServiceName, localService.getEndpoint());
209	0	}
210	0	}
211	0	return Collections.unmodifiableMap(localEndpoints);
212		}
213
214		@Override
215		public List<Endpoint> getAllEndpoints() {
216	0	List<Endpoint> allEndpoints = new ArrayList<Endpoint>();
217	0	synchronized (serviceLock) {
218	0	for (QName serviceName : this.localServices.keySet()) {
219	0	allEndpoints.add(this.localServices.get(serviceName).getEndpoint());
220		}
221	0	for (QName serviceName : this.clientRegistryCache.keySet()) {
222	0	Set<RemoteService> remoteServices = clientRegistryCache.get(serviceName);
223	0	for (RemoteService remoteService : remoteServices) {
224	0	allEndpoints.add(remoteService.getEndpoint());
225		}
226	0	}
227	0	}
228	0	return Collections.unmodifiableList(allEndpoints);
229		}
230
231		@Override
232		public Endpoint getEndpoint(QName serviceName) {
233	0	return getEndpoint(serviceName, null);
234		}
235
236		@Override
237		public Endpoint getEndpoint(QName serviceName, String applicationId) {
238	0	if (serviceName == null) {
239	0	throw new IllegalArgumentException("serviceName cannot be null");
240		}
241	0	Endpoint availableEndpoint = null;
242	0	synchronized (serviceLock) {
243		// look at local services first
244	0	availableEndpoint = getLocalEndpoint(serviceName);
245	0	if (availableEndpoint == null \|\| (!StringUtils.isBlank(applicationId) && !availableEndpoint.getServiceConfiguration().getApplicationId().equals(applicationId))) {
246		// TODO - would be better to return an Endpoint that contained an internal proxy to all the services so fail-over would be easier to implement!
247	0	Set<RemoteService> remoteServices = clientRegistryCache.get(serviceName);
248	0	remoteServices = filterByApplicationId(applicationId, remoteServices);
249	0	if (remoteServices != null && !remoteServices.isEmpty()) {
250		// TODO - this should also probably check the current status of the service?
251	0	RemoteService[] remoteServiceArray = remoteServices.toArray(new RemoteService[0]);
252	0	RemoteService availableRemoteService = remoteServiceArray[this.randomNumber.nextInt(remoteServiceArray.length)];
253	0	availableEndpoint = availableRemoteService.getEndpoint();
254		}
255		}
256	0	}
257	0	return availableEndpoint;
258		}
259
260		protected Set<RemoteService> filterByApplicationId(String applicationId, Set<RemoteService> remoteServices) {
261	0	if (StringUtils.isBlank(applicationId) \|\| remoteServices == null \|\| remoteServices.isEmpty()) {
262	0	return remoteServices;
263		}
264	0	Set<RemoteService> filtered = new HashSet<RemoteService>();
265	0	for (RemoteService remoteService : remoteServices) {
266	0	if (remoteService.getServiceInfo().getApplicationId().equals(applicationId)) {
267	0	filtered.add(remoteService);
268		}
269		}
270	0	return filtered;
271		}
272
273		@Override
274		public Endpoint getConfiguredEndpoint(ServiceConfiguration serviceConfiguration) {
275	0	if (serviceConfiguration == null) {
276	0	throw new IllegalArgumentException("serviceConfiguration cannot be null");
277		}
278	0	synchronized (serviceLock) {
279	0	Endpoint localEndpoint = getLocalEndpoint(serviceConfiguration.getServiceName());
280	0	if (localEndpoint != null && localEndpoint.getServiceConfiguration().equals(serviceConfiguration)) {
281	0	return localEndpoint;
282		}
283	0	List<Endpoint> remoteEndpoints = getRemoteEndpoints(serviceConfiguration.getServiceName());
284	0	for (Endpoint remoteEndpoint : remoteEndpoints) {
285	0	if (remoteEndpoint.getServiceConfiguration().equals(serviceConfiguration)) {
286	0	return remoteEndpoint;
287		}
288		}
289	0	}
290	0	return null;
291		}
292
293		@Override
294		public Object getService(QName serviceName) {
295	0	return getService(serviceName, null);
296		}
297
298		@Override
299		public Object getService(QName serviceName, String applicationId) {
300	0	Endpoint availableEndpoint = getEndpoint(serviceName, applicationId);
301	0	if (availableEndpoint == null) {
302	0	return null;
303		}
304	0	return availableEndpoint.getService();
305		}
306
307		@Override
308		public ServiceConfiguration publishService(ServiceDefinition serviceDefinition, boolean synchronize) {
309	0	if (serviceDefinition == null) {
310	0	throw new IllegalArgumentException("serviceDefinition cannot be null");
311		}
312	0	LocalService localService = new LocalService(getInstanceId(), serviceDefinition);
313	0	synchronized (serviceLock) {
314	0	serviceExportManager.exportService(serviceDefinition);
315	0	localServices.put(serviceDefinition.getServiceName(), localService);
316	0	}
317	0	if (synchronize) {
318	0	synchronize();
319		}
320	0	return localService.getEndpoint().getServiceConfiguration();
321		}
322
323		@Override
324		public List<ServiceConfiguration> publishServices(List<ServiceDefinition> serviceDefinitions, boolean synchronize) {
325	0	if (serviceDefinitions == null) {
326	0	throw new IllegalArgumentException("serviceDefinitions list cannot be null");
327		}
328	0	List<ServiceConfiguration> serviceConfigurations = new ArrayList<ServiceConfiguration>();
329	0	synchronized (serviceLock) {
330	0	for (ServiceDefinition serviceDefinition : serviceDefinitions) {
331	0	ServiceConfiguration serviceConfiguration = publishService(serviceDefinition, false);
332	0	serviceConfigurations.add(serviceConfiguration);
333	0	}
334	0	}
335	0	if (synchronize) {
336	0	synchronize();
337		}
338	0	return Collections.unmodifiableList(serviceConfigurations);
339		}
340
341		@Override
342		public boolean removeService(QName serviceName, boolean synchronize) {
343	0	if (serviceName == null) {
344	0	throw new IllegalArgumentException("serviceName cannot be null");
345		}
346	0	boolean serviceRemoved = false;
347	0	synchronized (serviceLock) {
348	0	LocalService localService = localServices.remove(serviceName);
349	0	serviceRemoved = localService != null;
350	0	serviceExportManager.removeService(serviceName);
351	0	}
352	0	if (serviceRemoved && synchronize) {
353	0	synchronize();
354		}
355	0	return serviceRemoved;
356		}
357
358		@Override
359		public List<Boolean> removeServices(List<QName> serviceNames, boolean synchronize) {
360	0	if (serviceNames == null) {
361	0	throw new IllegalArgumentException("serviceNames cannot be null");
362		}
363	0	boolean serviceRemoved = false;
364	0	List<Boolean> servicesRemoved = new ArrayList<Boolean>();
365	0	synchronized (serviceLock) {
366	0	for (QName serviceName : serviceNames) {
367	0	serviceExportManager.removeService(serviceName);
368	0	LocalService localService = localServices.remove(serviceName);
369	0	if (localService != null) {
370	0	servicesRemoved.add(Boolean.TRUE);
371	0	serviceRemoved = true;
372		} else {
373	0	servicesRemoved.add(Boolean.FALSE);
374		}
375	0	}
376	0	}
377	0	if (serviceRemoved && synchronize) {
378	0	synchronize();
379		}
380	0	return servicesRemoved;
381		}
382
383		@Override
384		public void synchronize() {
385	0	if (!isDevMode() && isStarted()) {
386	0	synchronized (synchronizeLock) {
387		List<LocalService> localServicesList;
388		List<RemoteService> clientRegistryCacheList;
389	0	synchronized (serviceLock) {
390		// first, flatten the lists
391	0	localServicesList = new ArrayList<LocalService>(this.localServices.values());
392	0	clientRegistryCacheList = new ArrayList<RemoteService>();
393	0	for (Set<RemoteService> remoteServices : this.clientRegistryCache.values()) {
394	0	clientRegistryCacheList.addAll(remoteServices);
395		}
396	0	}
397	0	CompleteServiceDiff serviceDiff = diffCalculator.diffServices(getInstanceId(), localServicesList, clientRegistryCacheList);
398
399	0	RemoteServicesDiff remoteServicesDiff = serviceDiff.getRemoteServicesDiff();
400	0	processRemoteServiceDiff(remoteServicesDiff);
401
402	0	LocalServicesDiff localServicesDiff = serviceDiff.getLocalServicesDiff();
403	0	processLocalServiceDiff(localServicesDiff);
404	0	}
405		}
406	0	}
407
408		protected void processRemoteServiceDiff(RemoteServicesDiff remoteServicesDiff) {
409		// note that since there is a gap between when the original services are acquired, the diff, and this subsequent critical section
410		// the list of local and client registry services could have changed, so that needs to be considered in the remaining code
411	0	synchronized (serviceLock) {
412		// first, let's update what we know about the remote services
413	0	List<RemoteService> removedServices = remoteServicesDiff.getRemovedServices();
414	0	for (RemoteService removedRemoteService : removedServices) {
415	0	Set<RemoteService> remoteServiceSet = this.clientRegistryCache.get(removedRemoteService.getServiceName());
416	0	if (remoteServiceSet != null) {
417	0	boolean wasRemoved = remoteServiceSet.remove(removedRemoteService);
418	0	if (!wasRemoved) {
419	0	LOG.warn("Failed to remove remoteService during synchronization: " + removedRemoteService);
420		}
421		}
422	0	}
423	0	List<ServiceInfo> newServices = remoteServicesDiff.getNewServices();
424	0	for (ServiceInfo newService : newServices) {
425	0	Set<RemoteService> remoteServiceSet = clientRegistryCache.get(newService.getServiceName());
426	0	if (remoteServiceSet == null) {
427	0	remoteServiceSet = new HashSet<RemoteService>();
428	0	clientRegistryCache.put(newService.getServiceName(), remoteServiceSet);
429		}
430	0	remoteServiceSet.add(new RemoteService(newService, this.serviceRegistry));
431	0	}
432	0	}
433	0	}
434
435		protected void processLocalServiceDiff(LocalServicesDiff localServicesDiff) {
436	0	List<String> removeServiceEndpointIds = new ArrayList<String>();
437	0	List<ServiceEndpoint> publishServiceEndpoints = new ArrayList<ServiceEndpoint>();
438	0	for (ServiceInfo serviceToRemove : localServicesDiff.getServicesToRemoveFromRegistry()) {
439	0	removeServiceEndpointIds.add(serviceToRemove.getServiceId());
440		}
441	0	for (LocalService localService : localServicesDiff.getLocalServicesToPublish()) {
442	0	publishServiceEndpoints.add(localService.getServiceEndpoint());
443		}
444	0	for (LocalService localService : localServicesDiff.getLocalServicesToUpdate().keySet()) {
445	0	ServiceInfo registryServiceInfo = localServicesDiff.getLocalServicesToUpdate().get(localService);
446	0	publishServiceEndpoints.add(rebuildServiceEndpointForUpdate(localService.getServiceEndpoint(), registryServiceInfo));
447	0	}
448	0	boolean batchMode = ConfigContext.getCurrentContextConfig().getBooleanProperty(Config.BATCH_MODE, false);
449	0	if (!batchMode && (!removeServiceEndpointIds.isEmpty() \|\| !publishServiceEndpoints.isEmpty())) {
450	0	RemoveAndPublishResult result = this.serviceRegistry.removeAndPublish(removeServiceEndpointIds, publishServiceEndpoints);
451		// now update the ServiceEndpoints for our local services so we can get the proper id for them
452	0	if (!result.getServicesPublished().isEmpty()) {
453	0	synchronized (serviceLock) {
454	0	for (ServiceEndpoint publishedService : result.getServicesPublished()) {
455	0	rebuildLocalServiceEndpointAfterPublishing(publishedService);
456		}
457	0	}
458		}
459		}
460	0	}
461
462		protected ServiceEndpoint rebuildServiceEndpointForUpdate(ServiceEndpoint originalEndpoint, ServiceInfo registryServiceInfo) {
463	0	ServiceEndpoint.Builder builder = ServiceEndpoint.Builder.create(originalEndpoint);
464	0	builder.getInfo().setServiceId(registryServiceInfo.getServiceId());
465	0	builder.getInfo().setServiceDescriptorId(registryServiceInfo.getServiceDescriptorId());
466	0	builder.getInfo().setVersionNumber(registryServiceInfo.getVersionNumber());
467	0	builder.getDescriptor().setId(registryServiceInfo.getServiceDescriptorId());
468	0	return builder.build();
469		}
470
471		protected void rebuildLocalServiceEndpointAfterPublishing(ServiceEndpoint publishedService) {
472		// verify the service is still published
473	0	QName serviceName = publishedService.getInfo().getServiceName();
474	0	if (localServices.containsKey(serviceName)) {
475	0	LocalService newLocalService = new LocalService(localServices.get(serviceName), publishedService);
476	0	localServices.put(serviceName, newLocalService);
477		}
478	0	}
479
480		public void setServiceRegistry(ServiceRegistry serviceRegistry) {
481	0	this.serviceRegistry = serviceRegistry;
482	0	}
483
484		public void setDiffCalculator(ServiceRegistryDiffCalculator diffCalculator) {
485	0	this.diffCalculator = diffCalculator;
486	0	}
487
488		public void setServiceExportManager(ServiceExportManager serviceExportManager) {
489	0	this.serviceExportManager = serviceExportManager;
490	0	}
491
492		public void setScheduledPool(KSBScheduledPool scheduledPool) {
493	0	this.scheduledPool = scheduledPool;
494	0	}
495
496		}