From e5b7380c874745c989d1816b8f552504f038e1bc Mon Sep 17 00:00:00 2001 From: lresende Date: Thu, 26 Sep 2013 20:33:20 +0000 Subject: 2.0 branch for possible maintenance release git-svn-id: http://svn.us.apache.org/repos/asf/tuscany@1526672 13f79535-47bb-0310-9956-ffa450edef68 --- .../discovery/impl/AbstractDiscoveryService.java | 265 +++++++++++++++++++++ 1 file changed, 265 insertions(+) create mode 100644 sca-java-2.x/branches/2.0/modules/node-impl-osgi/src/main/java/org/apache/tuscany/sca/osgi/service/discovery/impl/AbstractDiscoveryService.java (limited to 'sca-java-2.x/branches/2.0/modules/node-impl-osgi/src/main/java/org/apache/tuscany/sca/osgi/service/discovery/impl/AbstractDiscoveryService.java') diff --git a/sca-java-2.x/branches/2.0/modules/node-impl-osgi/src/main/java/org/apache/tuscany/sca/osgi/service/discovery/impl/AbstractDiscoveryService.java b/sca-java-2.x/branches/2.0/modules/node-impl-osgi/src/main/java/org/apache/tuscany/sca/osgi/service/discovery/impl/AbstractDiscoveryService.java new file mode 100644 index 0000000000..fd57dbded8 --- /dev/null +++ b/sca-java-2.x/branches/2.0/modules/node-impl-osgi/src/main/java/org/apache/tuscany/sca/osgi/service/discovery/impl/AbstractDiscoveryService.java @@ -0,0 +1,265 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache 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.apache.org/licenses/LICENSE-2.0 + * + * 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.apache.tuscany.sca.osgi.service.discovery.impl; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.Dictionary; +import java.util.Hashtable; +import java.util.Iterator; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.logging.Level; +import java.util.logging.Logger; + +import org.apache.tuscany.sca.core.ExtensionPointRegistry; +import org.apache.tuscany.sca.core.LifeCycleListener; +import org.apache.tuscany.sca.node.NodeFactory; +import org.apache.tuscany.sca.node.impl.NodeFactoryImpl; +import org.apache.tuscany.sca.osgi.remoteserviceadmin.impl.OSGiHelper; +import org.osgi.framework.Bundle; +import org.osgi.framework.BundleContext; +import org.osgi.framework.Constants; +import org.osgi.framework.Filter; +import org.osgi.framework.ServiceReference; +import org.osgi.service.remoteserviceadmin.EndpointDescription; +import org.osgi.service.remoteserviceadmin.EndpointListener; +import org.osgi.service.remoteserviceadmin.RemoteConstants; +import org.osgi.util.tracker.ServiceTracker; + +/** + * + */ +public abstract class AbstractDiscoveryService implements Discovery, LifeCycleListener { + protected final static int ADDED = 0x1; + protected final static int REMOVED = 0x2; + protected final static int MODIFIED = 0x4; + + protected final static Logger logger = Logger.getLogger(AbstractDiscoveryService.class.getName()); + + protected BundleContext context; + protected ExtensionPointRegistry registry; + // private WorkScheduler workScheduler; + + private Map> listenersToFilters = + new ConcurrentHashMap>(); + + protected Map endpointDescriptions = + new ConcurrentHashMap(); + private ServiceTracker trackerTracker; + + public AbstractDiscoveryService(BundleContext context) { + super(); + this.context = context; + } + + public void start() { + getExtensionPointRegistry(); + // UtilityExtensionPoint utilityExtensionPoint = registry.getExtensionPoint(UtilityExtensionPoint.class); + // this.workScheduler = utilityExtensionPoint.getUtility(WorkScheduler.class); + + // track the registration of EndpointListener + trackerTracker = new ServiceTracker(this.context, EndpointListener.class.getName(), null) { + public Object addingService(ServiceReference reference) { + Object result = super.addingService(reference); + cacheTracker(reference, result); + return result; + } + + public void modifiedService(ServiceReference reference, Object service) { + super.modifiedService(reference, service); + updateTracker(reference, service); + } + + public void removedService(ServiceReference reference, Object service) { + super.removedService(reference, service); + clearTracker(service); + } + }; + + trackerTracker.open(); + } + + public void stop() { + trackerTracker.close(); + } + + protected ExtensionPointRegistry getExtensionPointRegistry() { + NodeFactoryImpl factory = (NodeFactoryImpl)NodeFactory.getInstance(); + factory.init(); + ServiceTracker tracker = new ServiceTracker(context, ExtensionPointRegistry.class.getName(), null); + tracker.open(); + // tracker.waitForService(1000); + registry = (ExtensionPointRegistry)tracker.getService(); + tracker.close(); + return registry; + } + + protected Dictionary getProperties() { + Dictionary headers = context.getBundle().getHeaders(); + Hashtable props = new Hashtable(); + props.put(PRODUCT_NAME, "Apache Tuscany SCA"); + props.put(PRODUCT_VERSION, headers.get(Constants.BUNDLE_VERSION)); + props.put(VENDOR_NAME, headers.get(Constants.BUNDLE_VENDOR)); + // props.put(SUPPORTED_PROTOCOLS, new String[] {"local", "org.osgi.sca"}); + return props; + } + + private void cacheTracker(ServiceReference reference, Object service) { + if (service instanceof EndpointListener) { + EndpointListener listener = (EndpointListener)service; + Collection filters = null; + synchronized (this) { + filters = addTracker(reference, listener, EndpointListener.ENDPOINT_LISTENER_SCOPE); + // Take a snapshot of the endpoints + triggerCallbacks(null, filters, listener); + } + } + } + + private void clearTracker(Object service) { + if (service instanceof EndpointListener) { + synchronized (this) { + removeTracker((EndpointListener)service); + } + } + } + + private void updateTracker(ServiceReference reference, Object service) { + if (service instanceof EndpointListener) { + EndpointListener listener = (EndpointListener)service; + Collection oldFilters = null; + Collection newFilters = null; + synchronized (this) { + if (logger.isLoggable(Level.FINE)) { + logger.fine("updating listener: " + listener); + } + oldFilters = removeTracker(listener); + newFilters = addTracker(reference, listener, EndpointListener.ENDPOINT_LISTENER_SCOPE); + triggerCallbacks(oldFilters, newFilters, listener); + } + } + } + + private void triggerCallbacks(Collection oldInterest, + Collection newInterest, + EndpointListener listener) { + // compute delta between old & new interfaces/filters and + // trigger callbacks for any entries in servicesInfo that + // match any *additional* interface/filters + Collection deltaInterest = getDelta(oldInterest, newInterest); + + Iterator i = deltaInterest.iterator(); + while (i.hasNext()) { + String next = i.next(); + for (EndpointDescription sd : endpointDescriptions.keySet()) { + triggerCallbacks(listener, next, sd, ADDED); + } + } + // Find removed filters + deltaInterest = getDelta(newInterest, oldInterest); + + i = deltaInterest.iterator(); + while (i.hasNext()) { + String next = i.next(); + for (EndpointDescription sd : endpointDescriptions.keySet()) { + triggerCallbacks(listener, next, sd, REMOVED); + } + } + } + + private Collection getDelta(Collection oldInterest, Collection newInterest) { + if (newInterest == null) { + newInterest = Collections.emptySet(); + } + + Collection deltaInterest = new ArrayList(newInterest); + if (oldInterest == null) { + oldInterest = Collections.emptySet(); + } + deltaInterest.removeAll(oldInterest); + return deltaInterest; + } + + /** + * Notify the endpoint listener + * @param listener + * @param matchedFilter + * @param endpoint + * @param type + */ + private static void notify(EndpointListener listener, String matchedFilter, EndpointDescription endpoint, int type) { + switch (type) { + case ADDED: + listener.endpointAdded(endpoint, matchedFilter); + break; + case REMOVED: + listener.endpointRemoved(endpoint, matchedFilter); + break; + case MODIFIED: + listener.endpointRemoved(endpoint, matchedFilter); + listener.endpointAdded(endpoint, matchedFilter); + break; + } + } + + private void triggerCallbacks(EndpointListener listener, + String matchedFilter, + EndpointDescription endpoint, + int type) { + // workScheduler.scheduleWork(new Notifier(listener, matchedFilter, endpoint, type)); + notify(listener, matchedFilter, endpoint, type); + } + + private boolean filterMatches(String filterValue, EndpointDescription sd) { + Filter filter = OSGiHelper.createFilter(context, filterValue); + Hashtable props = new Hashtable(sd.getProperties()); + // Add two faked properties to make the filter match + props.put(Constants.OBJECTCLASS, sd.getInterfaces()); + props.put(RemoteConstants.SERVICE_IMPORTED, "true"); + return filter != null ? filter.match(props) : false; + } + + private Collection removeTracker(EndpointListener listener) { + return listenersToFilters.remove(listener); + } + + private Collection addTracker(ServiceReference reference, EndpointListener listener, String property) { + Collection collection = OSGiHelper.getStringCollection(reference, property); + if (collection != null && !collection.isEmpty()) { + listenersToFilters.put(listener, new ArrayList(collection)); + } + return collection; + } + + protected void endpointChanged(EndpointDescription sd, int type) { + synchronized (this) { + for (Map.Entry> entry : listenersToFilters.entrySet()) { + for (String filter : entry.getValue()) { + if (filterMatches(filter, sd)) { + triggerCallbacks(entry.getKey(), filter, sd, type); + } + } + } + } + } + +} -- cgit v1.2.3