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

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