Source code

001/**
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *     http://www.apache.org/licenses/LICENSE-2.0
010 *
011 *  Unless required by applicable law or agreed to in writing, software
012 *  distributed under the License is distributed on an "AS IS" BASIS,
013 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 *  See the License for the specific language governing permissions and
015 *  limitations under the License.
016 */
017package org.apache.activemq.util.osgi;
018
019import static org.osgi.framework.wiring.BundleRevision.PACKAGE_NAMESPACE;
020
021import java.io.BufferedReader;
022import java.io.IOException;
023import java.io.InputStream;
024import java.io.InputStreamReader;
025import java.net.URL;
026import java.util.ArrayList;
027import java.util.HashSet;
028import java.util.List;
029import java.util.Properties;
030import java.util.Set;
031import java.util.concurrent.ConcurrentHashMap;
032import java.util.concurrent.ConcurrentMap;
033import java.net.URL;
034
035import org.apache.activemq.Service;
036import org.apache.activemq.store.PersistenceAdapter;
037import org.apache.activemq.transport.Transport;
038import org.apache.activemq.transport.discovery.DiscoveryAgent;
039import org.apache.activemq.util.FactoryFinder;
040import org.apache.activemq.util.FactoryFinder.ObjectFactory;
041import org.slf4j.LoggerFactory;
042import org.slf4j.Logger;
043
044import org.osgi.framework.Bundle;
045import org.osgi.framework.BundleActivator;
046import org.osgi.framework.BundleContext;
047import org.osgi.framework.BundleEvent;
048import org.osgi.framework.SynchronousBundleListener;
049import org.osgi.framework.wiring.BundleCapability;
050import org.osgi.framework.wiring.BundleWire;
051import org.osgi.framework.wiring.BundleWiring;
052
053/**
054 * An OSGi bundle activator for ActiveMQ which adapts the {@link org.apache.activemq.util.FactoryFinder}
055 * to the OSGi environment.
056 *
057 */
058public class Activator implements BundleActivator, SynchronousBundleListener, ObjectFactory {
059
060    private static final Logger LOG = LoggerFactory.getLogger(Activator.class);
061
062    private final ConcurrentMap<String, Class<?>> serviceCache = new ConcurrentHashMap<String, Class<?>>();
063    private final ConcurrentMap<Long, BundleWrapper> bundleWrappers = new ConcurrentHashMap<Long, BundleWrapper>();
064    private BundleContext bundleContext;
065    private Set<BundleCapability> packageCapabilities = new HashSet<BundleCapability>();
066
067    // ================================================================
068    // BundleActivator interface impl
069    // ================================================================
070
071    public synchronized void start(BundleContext bundleContext) throws Exception {
072
073        // This is how we replace the default FactoryFinder strategy
074        // with one that is more compatible in an OSGi env.
075        FactoryFinder.setObjectFactory(this);
076
077        debug("activating");
078        this.bundleContext = bundleContext;
079        
080        cachePackageCapabilities(Service.class, Transport.class, DiscoveryAgent.class, PersistenceAdapter.class);
081        
082        debug("checking existing bundles");
083        bundleContext.addBundleListener(this);
084        for (Bundle bundle : bundleContext.getBundles()) {
085            if (bundle.getState() == Bundle.RESOLVED || bundle.getState() == Bundle.STARTING ||
086                bundle.getState() == Bundle.ACTIVE || bundle.getState() == Bundle.STOPPING) {
087                register(bundle);
088            }
089        }
090        debug("activated");
091    }
092
093    /**
094     * Caches the package capabilities that are needed for a set of interface classes
095     *  
096     * @param classes interfaces we want to track
097     */
098    private void cachePackageCapabilities(Class<?> ... classes) {
099        BundleWiring ourWiring = bundleContext.getBundle().adapt(BundleWiring.class);
100        Set<String> packageNames = new HashSet<String>();
101        for (Class<?> clazz: classes) {
102            packageNames.add(clazz.getPackage().getName());
103        }
104        
105        List<BundleCapability> ourExports = ourWiring.getCapabilities(PACKAGE_NAMESPACE);
106        for (BundleCapability ourExport : ourExports) {
107            String ourPkgName = (String) ourExport.getAttributes().get(PACKAGE_NAMESPACE);
108            if (packageNames.contains(ourPkgName)) {
109                packageCapabilities.add(ourExport);
110            }
111        }
112    }
113
114
115    public synchronized void stop(BundleContext bundleContext) throws Exception {
116        debug("deactivating");
117        bundleContext.removeBundleListener(this);
118        while (!bundleWrappers.isEmpty()) {
119            unregister(bundleWrappers.keySet().iterator().next());
120        }
121        debug("deactivated");
122        this.bundleContext = null;
123    }
124
125    // ================================================================
126    // SynchronousBundleListener interface impl
127    // ================================================================
128
129    public void bundleChanged(BundleEvent event) {
130        if (event.getType() == BundleEvent.RESOLVED) {
131            register(event.getBundle());
132        } else if (event.getType() == BundleEvent.UNRESOLVED || event.getType() == BundleEvent.UNINSTALLED) {
133            unregister(event.getBundle().getBundleId());
134        }
135    }
136
137    protected void register(final Bundle bundle) {
138        debug("checking bundle " + bundle.getBundleId());
139        if (isOurBundle(bundle) || isImportingUs(bundle) ) {
140            debug("Registering bundle for extension resolution: "+ bundle.getBundleId());
141            bundleWrappers.put(bundle.getBundleId(), new BundleWrapper(bundle));
142        }
143    }
144
145    private boolean isOurBundle(final Bundle bundle) {
146        return bundle.getBundleId() == bundleContext.getBundle().getBundleId();
147    }
148
149    /**
150     * When bundles unload.. we remove them thier cached Class entries from the
151     * serviceCache.  Future service lookups for the service will fail.
152     *
153     * TODO: consider a way to get the Broker release any references to
154     * instances of the service.
155     *
156     * @param bundleId
157     */
158    protected void unregister(long bundleId) {
159        BundleWrapper bundle = bundleWrappers.remove(bundleId);
160        if (bundle != null) {
161            for (String path : bundle.cachedServices) {
162                debug("unregistering service for key: " +path );
163                serviceCache.remove(path);
164            }
165        }
166    }
167
168    // ================================================================
169    // ObjectFactory interface impl
170    // ================================================================
171
172    public Object create(String path) throws IllegalAccessException, InstantiationException, IOException, ClassNotFoundException {
173        Class<?> clazz = serviceCache.get(path);
174        if (clazz == null) {
175            StringBuffer warnings = new StringBuffer();
176            // We need to look for a bundle that has that class.
177            int wrrningCounter=1;
178            for (BundleWrapper wrapper : bundleWrappers.values()) {
179                URL resource = wrapper.bundle.getResource(path);
180                if( resource == null ) {
181                    continue;
182                }
183
184                Properties properties = loadProperties(resource);
185
186                String className = properties.getProperty("class");
187                if (className == null) {
188                    warnings.append("("+(wrrningCounter++)+") Invalid service file in bundle "+wrapper+": 'class' property not defined.");
189                    continue;
190                }
191
192                try {
193                    clazz = wrapper.bundle.loadClass(className);
194                } catch (ClassNotFoundException e) {
195                    warnings.append("("+(wrrningCounter++)+") Bundle "+wrapper+" could not load "+className+": "+e);
196                    continue;
197                }
198
199                // Yay.. the class was found.  Now cache it.
200                serviceCache.put(path, clazz);
201                wrapper.cachedServices.add(path);
202                break;
203            }
204
205            if( clazz == null ) {
206                // Since OSGi is such a tricky environment to work in.. lets give folks the
207                // most information we can in the error message.
208                String msg = "Service not found: '" + path + "'";
209                if (warnings.length()!= 0) {
210                    msg += ", "+warnings;
211                }
212                throw new IOException(msg);
213            }
214        }
215        return clazz.newInstance();
216    }
217
218    // ================================================================
219    // Internal Helper Methods
220    // ================================================================
221
222    private void debug(Object msg) {
223        LOG.debug(msg.toString());
224    }
225
226    private Properties loadProperties(URL resource) throws IOException {
227        InputStream in = resource.openStream();
228        try {
229            BufferedReader br = new BufferedReader(new InputStreamReader(in, "UTF-8"));
230            Properties properties = new Properties();
231            properties.load(in);
232            return properties;
233        } finally {
234            try {
235                in.close();
236            } catch (Exception e) {
237            }
238        }
239    }
240
241    /**
242     * We consider a bundle to be a candidate for objects if it imports at least
243     * one of the packages of our interfaces
244     * 
245     * @param bundle
246     * @return
247     */
248    private boolean isImportingUs(Bundle bundle) {
249        BundleWiring wiring = bundle.adapt(BundleWiring.class);
250        List<BundleWire> imports = wiring.getRequiredWires(PACKAGE_NAMESPACE);
251        for (BundleWire importWire : imports) {
252            if (packageCapabilities.contains(importWire.getCapability())) {
253                return true;
254            }
255        }
256        return false;
257    }
258
259    private static class BundleWrapper {
260        private final Bundle bundle;
261        private final List<String> cachedServices = new ArrayList<String>();
262
263        public BundleWrapper(Bundle bundle) {
264            this.bundle = bundle;
265        }
266    }
267}