CXF-DOSGI学习笔记(二)-OSGI Declare Service注入的实现
颛孙和颂
一、服务注入(有4种场景)
- 场景1：Service changed,针对RemoteServiceAdmin class的
1.RemoteServiceAdminList.
stRemoteServiceAdmin = new ServiceTracker(bctx, RemoteServiceAdmin.class.getName(), null) { @Override public Object addingService(ServiceReference reference) { LOG.info("Adding RemoteServiceAdmin to list of admins "); RemoteServiceAdmin rsa = (RemoteServiceAdmin)bctx.getService(reference); synchronized (rsal) { rsal.add(rsa); } LOG.info("enlisted RemoteEventAdmins: " + this.size()); triggerExportImportForRemoteSericeAdmin(rsa); return super.addingService(reference); } @Override public void removedService(ServiceReference reference, Object service) { LOG.info("TopologyManager: Removing RemoteServiceAdmin from list of admins "); synchronized (rsal) { rsal.remove(service); } // TODO: remove service exports from management structure and notify discovery stuff... removeRemoteServiceAdmin((RemoteServiceAdmin)service); LOG.info("TopologyManager: enlisted RemoteEventAdmins: " + rsal.size()); super.removedService(reference, service); } }; protected void triggerExportImportForRemoteSericeAdmin(RemoteServiceAdmin rsa) { topManager.triggerExportImportForRemoteSericeAdmin(rsa); -- 接着调用 triggerExport 见blog 1 topManagerImport.triggerExportImportForRemoteSericeAdmin(rsa); }
2.TopologyManagerImport
public void triggerExportImportForRemoteSericeAdmin(RemoteServiceAdmin rsa) { LOG.fine("New RSA detected trying to import services with it"); synchronized (importPossibilities) { Set<Map.Entry<String, List<EndpointDescription>>> entries = importPossibilities.entrySet(); for (Entry<String, List<EndpointDescription>> entry : entries) { triggerImport(entry.getKey()); } } } private void triggerImport(final String filter) { LOG.log(Level.FINE, "import of a service for filter {0} was queued", filter); execService.execute(new Runnable() { public void run() { synchronized (importedServices) { // deadlock possibility ? synchronized (importPossibilities) { if (importAllAvailable) { importAllServicesStrategy(filter); } else { importSingleServiceStrategy(filter); } } } // Notify EndpointListeners ? NO! } }); } 调用importAllServicesStrategy( importSingleServiceStrategy类似),同时也做服务清理，用于删除importservice private void importAllServicesStrategy(String filter) { List<ImportRegistration> irs = importedServices.get(filter); if (irs == null) { irs = new ArrayList<ImportRegistration>(); importedServices.put(filter, irs); } if (irs.size() > 0) { // remove old services that are not available anymore List<EndpointDescription> ips = importPossibilities.get(filter); Iterator<ImportRegistration> it = irs.iterator(); while (it.hasNext()) { ImportRegistration ir = it.next(); EndpointDescription ep = ir.getImportReference().getImportedEndpoint(); // if service is already imported, check if endpoint is still in the list of // possible imports if ((ips != null && !ips.contains(ep)) || ips == null) { // unexport service ir.close(); it.remove(); } } } for (EndpointDescription epd : importPossibilities.get(filter)) { if (!irs.contains(epd)) { // service not imported yet -> import it now ImportRegistration ir = importService(epd); if (ir != null) { // import was successful irs.add(ir); } } } } private ImportRegistration importService(EndpointDescription ep) { synchronized (remoteServiceAdminList) { if(remoteServiceAdminList == null || remoteServiceAdminList.size() == 0) { LOG.log(Level.WARNING, "Unable to import service ({0}): no RemoteServiceAdmin service available!", ep); } for (RemoteServiceAdmin rsa : remoteServiceAdminList) { ImportRegistration ir = rsa.importService(ep); if (ir != null && ir.getException() == null) { // successful LOG.fine("service impoort was successful: " + ir); return ir; } else { // failed -> next RSA } } } return null; }
3.RemoteServiceAdminInstance
public ImportRegistration importService(EndpointDescription endpoint) { final EndpointDescription epd = endpoint; SecurityManager sm = System.getSecurityManager(); EndpointPermission epp = new EndpointPermission(epd, OsgiUtils.getUUID(bctx), EndpointPermission.IMPORT); if (sm != null) { sm.checkPermission(epp); } return AccessController.doPrivileged(new PrivilegedAction<ImportRegistration>() { public ImportRegistration run() { if (closed) return null; synchronized (importedServices) { ImportRegistration ir = rsaCore.importService(epd); if (ir != null) importedServices.add(ir); return ir; } } }); }
4.RemoteServiceAdminCore
public ImportRegistration importService(EndpointDescription endpoint) { LOG.info("importService() Endpoint: " + endpoint.getProperties()); synchronized (importedServices) { if (importedServices.containsKey(endpoint) && importedServices.get(endpoint).size() > 0) { LOG.fine("creating copy of existing import registrations"); Collection<ImportRegistrationImpl> imRegs = importedServices.get(endpoint); ImportRegistrationImpl irParent = imRegs.iterator().next(); ImportRegistrationImpl ir = new ImportRegistrationImpl(irParent); imRegs.add(ir); eventProducer.publishNotifcation(ir); return ir; } List remoteConfigurationTypes = endpoint.getConfigurationTypes(); if (remoteConfigurationTypes == null) { LOG.severe("the supplied endpoint has no configuration type"); return null; } List<String> usableConfigurationTypes = new ArrayList<String>(); for (String ct : supportedConfigurationTypes) { if (remoteConfigurationTypes.contains(ct)) { usableConfigurationTypes.add(ct); } } if (usableConfigurationTypes.size() == 0) { LOG .severe("the supplied endpoint has no compatible configuration type. Supported types are: " + supportedConfigurationTypes + " Types needed by the endpoint: " + remoteConfigurationTypes); return null; } Map<String, Object> emptyProps = Collections.EMPTY_MAP; ConfigurationTypeHandler handler = getHandler(usableConfigurationTypes, endpoint.getProperties(), emptyProps); if (handler == null) { LOG.severe("no handler found"); return null; } LOG.fine("Handler: " + handler); // // TODO: somehow select the interfaces that should be imported ----> job of the TopologyManager // ? List<String> matchingInterfaces = endpoint.getInterfaces(); LOG.info("Interfaces: " + matchingInterfaces); if (matchingInterfaces.size() == 1) { LOG.info("Proxifying interface : " + matchingInterfaces.get(0)); ImportRegistrationImpl imReg = new ImportRegistrationImpl(endpoint, this); proxifyMatchingInterface(matchingInterfaces.get(0), imReg, handler, bctx); Collection<ImportRegistrationImpl> imRegs = importedServices.get(endpoint); if (imRegs == null) { imRegs = new ArrayList<ImportRegistrationImpl>(); importedServices.put(endpoint, imRegs); } imRegs.add(imReg); eventProducer.publishNotifcation(imReg); return imReg; } else { return null; } } } protected void proxifyMatchingInterface(String interfaceName, ImportRegistrationImpl imReg, ConfigurationTypeHandler handler, BundleContext requestingContext) { try { // MARC: relies on dynamic imports ? Class<?> iClass = bctx.getBundle().loadClass(interfaceName); if (iClass != null) { BundleContext actualContext = bctx; Class<?> actualClass = requestingContext.getBundle().loadClass(interfaceName); if (actualClass != iClass) { LOG.info("Class " + interfaceName + " loaded by DSW's bundle context is not " + "equal to the one loaded by the requesting bundle context, " + "DSW will use the requesting bundle context to register " + "a proxy service"); iClass = actualClass; actualContext = requestingContext; } /* TODO: add additional local params ... */ Dictionary serviceProps = new Hashtable(imReg.getImportedEndpointDescription() .getProperties()); serviceProps.put(RemoteConstants.SERVICE_IMPORTED, true); serviceProps.remove(RemoteConstants.SERVICE_EXPORTED_INTERFACES); // synchronized (discoveredServices) { ClientServiceFactory csf = new ClientServiceFactory(actualContext, iClass, imReg .getImportedEndpointDescription(), handler, imReg); imReg.setClientServiceFactory(csf); ServiceRegistration proxyRegistration = actualContext.registerService(interfaceName, csf, serviceProps); imReg.setImportedServiceRegistration(proxyRegistration); // cacheEndpointId(sd, proxyRegistration); // } } else { LOG.info("not proxifying service, cannot load interface class: " + interfaceName); imReg.setException(new ClassNotFoundException( "not proxifying service, cannot load interface class: " + interfaceName)); } } catch (ClassNotFoundException ex) { LOG.warning("No class can be found for " + interfaceName); imReg.setException(ex); } }
stRemoteServiceAdmin = new ServiceTracker(bctx, RemoteServiceAdmin.class.getName(), null) { @Override public Object addingService(ServiceReference reference) { LOG.info("Adding RemoteServiceAdmin to list of admins "); RemoteServiceAdmin rsa = (RemoteServiceAdmin)bctx.getService(reference); synchronized (rsal) { rsal.add(rsa); } LOG.info("enlisted RemoteEventAdmins: " + this.size()); triggerExportImportForRemoteSericeAdmin(rsa); return super.addingService(reference); } @Override public void removedService(ServiceReference reference, Object service) { LOG.info("TopologyManager: Removing RemoteServiceAdmin from list of admins "); synchronized (rsal) { rsal.remove(service); } // TODO: remove service exports from management structure and notify discovery stuff... removeRemoteServiceAdmin((RemoteServiceAdmin)service); LOG.info("TopologyManager: enlisted RemoteEventAdmins: " + rsal.size()); super.removedService(reference, service); } }; protected void triggerExportImportForRemoteSericeAdmin(RemoteServiceAdmin rsa) { topManager.triggerExportImportForRemoteSericeAdmin(rsa); -- 接着调用 triggerExport 见blog 1 topManagerImport.triggerExportImportForRemoteSericeAdmin(rsa); }
2.TopologyManagerImport
public void triggerExportImportForRemoteSericeAdmin(RemoteServiceAdmin rsa) { LOG.fine("New RSA detected trying to import services with it"); synchronized (importPossibilities) { Set<Map.Entry<String, List<EndpointDescription>>> entries = importPossibilities.entrySet(); for (Entry<String, List<EndpointDescription>> entry : entries) { triggerImport(entry.getKey()); } } } private void triggerImport(final String filter) { LOG.log(Level.FINE, "import of a service for filter {0} was queued", filter); execService.execute(new Runnable() { public void run() { synchronized (importedServices) { // deadlock possibility ? synchronized (importPossibilities) { if (importAllAvailable) { importAllServicesStrategy(filter); } else { importSingleServiceStrategy(filter); } } } // Notify EndpointListeners ? NO! } }); } 调用importAllServicesStrategy( importSingleServiceStrategy类似),同时也做服务清理，用于删除importservice private void importAllServicesStrategy(String filter) { List<ImportRegistration> irs = importedServices.get(filter); if (irs == null) { irs = new ArrayList<ImportRegistration>(); importedServices.put(filter, irs); } if (irs.size() > 0) { // remove old services that are not available anymore List<EndpointDescription> ips = importPossibilities.get(filter); Iterator<ImportRegistration> it = irs.iterator(); while (it.hasNext()) { ImportRegistration ir = it.next(); EndpointDescription ep = ir.getImportReference().getImportedEndpoint(); // if service is already imported, check if endpoint is still in the list of // possible imports if ((ips != null && !ips.contains(ep)) || ips == null) { // unexport service ir.close(); it.remove(); } } } for (EndpointDescription epd : importPossibilities.get(filter)) { if (!irs.contains(epd)) { // service not imported yet -> import it now ImportRegistration ir = importService(epd); if (ir != null) { // import was successful irs.add(ir); } } } } private ImportRegistration importService(EndpointDescription ep) { synchronized (remoteServiceAdminList) { if(remoteServiceAdminList == null || remoteServiceAdminList.size() == 0) { LOG.log(Level.WARNING, "Unable to import service ({0}): no RemoteServiceAdmin service available!", ep); } for (RemoteServiceAdmin rsa : remoteServiceAdminList) { ImportRegistration ir = rsa.importService(ep); if (ir != null && ir.getException() == null) { // successful LOG.fine("service impoort was successful: " + ir); return ir; } else { // failed -> next RSA } } } return null; }
3.RemoteServiceAdminInstance
public ImportRegistration importService(EndpointDescription endpoint) { final EndpointDescription epd = endpoint; SecurityManager sm = System.getSecurityManager(); EndpointPermission epp = new EndpointPermission(epd, OsgiUtils.getUUID(bctx), EndpointPermission.IMPORT); if (sm != null) { sm.checkPermission(epp); } return AccessController.doPrivileged(new PrivilegedAction<ImportRegistration>() { public ImportRegistration run() { if (closed) return null; synchronized (importedServices) { ImportRegistration ir = rsaCore.importService(epd); if (ir != null) importedServices.add(ir); return ir; } } }); }
4.RemoteServiceAdminCore
public ImportRegistration importService(EndpointDescription endpoint) { LOG.info("importService() Endpoint: " + endpoint.getProperties()); synchronized (importedServices) { if (importedServices.containsKey(endpoint) && importedServices.get(endpoint).size() > 0) { LOG.fine("creating copy of existing import registrations"); Collection<ImportRegistrationImpl> imRegs = importedServices.get(endpoint); ImportRegistrationImpl irParent = imRegs.iterator().next(); ImportRegistrationImpl ir = new ImportRegistrationImpl(irParent); imRegs.add(ir); eventProducer.publishNotifcation(ir); return ir; } List remoteConfigurationTypes = endpoint.getConfigurationTypes(); if (remoteConfigurationTypes == null) { LOG.severe("the supplied endpoint has no configuration type"); return null; } List<String> usableConfigurationTypes = new ArrayList<String>(); for (String ct : supportedConfigurationTypes) { if (remoteConfigurationTypes.contains(ct)) { usableConfigurationTypes.add(ct); } } if (usableConfigurationTypes.size() == 0) { LOG .severe("the supplied endpoint has no compatible configuration type. Supported types are: " + supportedConfigurationTypes + " Types needed by the endpoint: " + remoteConfigurationTypes); return null; } Map<String, Object> emptyProps = Collections.EMPTY_MAP; ConfigurationTypeHandler handler = getHandler(usableConfigurationTypes, endpoint.getProperties(), emptyProps); if (handler == null) { LOG.severe("no handler found"); return null; } LOG.fine("Handler: " + handler); // // TODO: somehow select the interfaces that should be imported ----> job of the TopologyManager // ? List<String> matchingInterfaces = endpoint.getInterfaces(); LOG.info("Interfaces: " + matchingInterfaces); if (matchingInterfaces.size() == 1) { LOG.info("Proxifying interface : " + matchingInterfaces.get(0)); ImportRegistrationImpl imReg = new ImportRegistrationImpl(endpoint, this); proxifyMatchingInterface(matchingInterfaces.get(0), imReg, handler, bctx); Collection<ImportRegistrationImpl> imRegs = importedServices.get(endpoint); if (imRegs == null) { imRegs = new ArrayList<ImportRegistrationImpl>(); importedServices.put(endpoint, imRegs); } imRegs.add(imReg); eventProducer.publishNotifcation(imReg); return imReg; } else { return null; } } } protected void proxifyMatchingInterface(String interfaceName, ImportRegistrationImpl imReg, ConfigurationTypeHandler handler, BundleContext requestingContext) { try { // MARC: relies on dynamic imports ? Class<?> iClass = bctx.getBundle().loadClass(interfaceName); if (iClass != null) { BundleContext actualContext = bctx; Class<?> actualClass = requestingContext.getBundle().loadClass(interfaceName); if (actualClass != iClass) { LOG.info("Class " + interfaceName + " loaded by DSW's bundle context is not " + "equal to the one loaded by the requesting bundle context, " + "DSW will use the requesting bundle context to register " + "a proxy service"); iClass = actualClass; actualContext = requestingContext; } /* TODO: add additional local params ... */ Dictionary serviceProps = new Hashtable(imReg.getImportedEndpointDescription() .getProperties()); serviceProps.put(RemoteConstants.SERVICE_IMPORTED, true); serviceProps.remove(RemoteConstants.SERVICE_EXPORTED_INTERFACES); // synchronized (discoveredServices) { ClientServiceFactory csf = new ClientServiceFactory(actualContext, iClass, imReg .getImportedEndpointDescription(), handler, imReg); imReg.setClientServiceFactory(csf); ServiceRegistration proxyRegistration = actualContext.registerService(interfaceName, csf, serviceProps); imReg.setImportedServiceRegistration(proxyRegistration); // cacheEndpointId(sd, proxyRegistration); // } } else { LOG.info("not proxifying service, cannot load interface class: " + interfaceName); imReg.setException(new ClassNotFoundException( "not proxifying service, cannot load interface class: " + interfaceName)); } } catch (ClassNotFoundException ex) { LOG.warning("No class can be found for " + interfaceName); imReg.setException(ex); } }
- 场景2： LocalDiscovery初始化
1.LocalDiscovery初始化
public LocalDiscovery(BundleContext bc) { ...... bundleContext.addBundleListener(this); processExistingBundles(); } private void processExistingBundles() { Bundle [] bundles = bundleContext.getBundles(); if (bundles == null) { return; } for (Bundle b : bundles) { if (b.getState() == Bundle.ACTIVE) { findDeclaredRemoteServices(b); } } } private void findDeclaredRemoteServices(Bundle bundle) { List<EndpointDescription> eds = LocalDiscoveryUtils.getAllEndpointDescriptions(bundle); for (EndpointDescription ed : eds) { endpointDescriptions.put(ed, bundle); addedEndpointDescription(ed); } } private void addedEndpointDescription(EndpointDescription ed) { triggerCallbacks(ed, true); } private void triggerCallbacks(EndpointDescription ed, boolean added) { for (Map.Entry<EndpointListener, Collection<String>> entry : listenerToFilters.entrySet()) { for (String match : entry.getValue()) { triggerCallbacks(entry.getKey(), match, ed, added); } } } private void triggerCallbacks(EndpointListener listener, String toMatch, EndpointDescription ed, boolean added) { if (!filterMatches(toMatch, ed)) { return; } if (added) { listener.endpointAdded(ed, toMatch); } else { listener.endpointRemoved(ed, toMatch); } }
2.org.apache.cxf.dosgi.topologymanager.EndpointListenerImpl
public void endpointAdded(EndpointDescription epd, String filter) { LOG.info("EndpointListenerImpl: EndpointAdded() filter:"+filter+" EndpointDesc:"+epd); if(filter==null){ LOG.severe("Endpoint is not handled because no matching filter was provided! Filter: "+filter); return; } // Decide if it is worth it ? topManager.addImportableService(filter,epd); }
3.TopologyManagerImport
public void addImportableService(String filter, EndpointDescription epd) { LOG.log(Level.FINE, "importable service added for filter {0} -> {1}", new Object[]{filter, epd}); synchronized (importPossibilities) { List<EndpointDescription> ips = importPossibilities.get(filter); if (ips == null) { ips = new ArrayList<EndpointDescription>(); importPossibilities.put(filter, ips); } ips.add(epd); } triggerImport(filter); } private void triggerImport(final String filter) { LOG.log(Level.FINE, "import of a service for filter {0} was queued", filter); execService.execute(new Runnable() { public void run() { synchronized (importedServices) { // deadlock possibility ? synchronized (importPossibilities) { if (importAllAvailable) { importAllServicesStrategy(filter); } else { importSingleServiceStrategy(filter); } } } // Notify EndpointListeners ? NO! } }); }
4.TopologyManagerImport后续步骤同上 Service changed
注意：删除服务时也会重新判断一下是否有需要重新importservice
public void removeImportableService(String filter, EndpointDescription epd) { synchronized (importPossibilities) { List<EndpointDescription> ips = importPossibilities.get(filter); if (ips != null) { ips.remove(epd); } else { // should not happen } } triggerImport(filter); }
public LocalDiscovery(BundleContext bc) { ...... bundleContext.addBundleListener(this); processExistingBundles(); } private void processExistingBundles() { Bundle [] bundles = bundleContext.getBundles(); if (bundles == null) { return; } for (Bundle b : bundles) { if (b.getState() == Bundle.ACTIVE) { findDeclaredRemoteServices(b); } } } private void findDeclaredRemoteServices(Bundle bundle) { List<EndpointDescription> eds = LocalDiscoveryUtils.getAllEndpointDescriptions(bundle); for (EndpointDescription ed : eds) { endpointDescriptions.put(ed, bundle); addedEndpointDescription(ed); } } private void addedEndpointDescription(EndpointDescription ed) { triggerCallbacks(ed, true); } private void triggerCallbacks(EndpointDescription ed, boolean added) { for (Map.Entry<EndpointListener, Collection<String>> entry : listenerToFilters.entrySet()) { for (String match : entry.getValue()) { triggerCallbacks(entry.getKey(), match, ed, added); } } } private void triggerCallbacks(EndpointListener listener, String toMatch, EndpointDescription ed, boolean added) { if (!filterMatches(toMatch, ed)) { return; } if (added) { listener.endpointAdded(ed, toMatch); } else { listener.endpointRemoved(ed, toMatch); } }
2.org.apache.cxf.dosgi.topologymanager.EndpointListenerImpl
public void endpointAdded(EndpointDescription epd, String filter) { LOG.info("EndpointListenerImpl: EndpointAdded() filter:"+filter+" EndpointDesc:"+epd); if(filter==null){ LOG.severe("Endpoint is not handled because no matching filter was provided! Filter: "+filter); return; } // Decide if it is worth it ? topManager.addImportableService(filter,epd); }
3.TopologyManagerImport
public void addImportableService(String filter, EndpointDescription epd) { LOG.log(Level.FINE, "importable service added for filter {0} -> {1}", new Object[]{filter, epd}); synchronized (importPossibilities) { List<EndpointDescription> ips = importPossibilities.get(filter); if (ips == null) { ips = new ArrayList<EndpointDescription>(); importPossibilities.put(filter, ips); } ips.add(epd); } triggerImport(filter); } private void triggerImport(final String filter) { LOG.log(Level.FINE, "import of a service for filter {0} was queued", filter); execService.execute(new Runnable() { public void run() { synchronized (importedServices) { // deadlock possibility ? synchronized (importPossibilities) { if (importAllAvailable) { importAllServicesStrategy(filter); } else { importSingleServiceStrategy(filter); } } } // Notify EndpointListeners ? NO! } }); }
4.TopologyManagerImport后续步骤同上 Service changed
注意：删除服务时也会重新判断一下是否有需要重新importservice
public void removeImportableService(String filter, EndpointDescription epd) { synchronized (importPossibilities) { List<EndpointDescription> ips = importPossibilities.get(filter); if (ips != null) { ips.remove(epd); } else { // should not happen } } triggerImport(filter); }
- 场景3：service changed,针对EndpointListener class
1.LocalDiscovery中的ServiceTracker
public LocalDiscovery(BundleContext bc) { bundleContext = bc; listenerTracker = new ServiceTracker(bundleContext, EndpointListener.class.getName(), null) { @Override public Object addingService(ServiceReference reference) { Object svc = super.addingService(reference); registerTracker(reference, svc); return svc; } @Override public void modifiedService(ServiceReference reference, Object service) { super.modifiedService(reference, service); clearTracker(service); // This may cause duplicate registrations of remote services, // but that's fine and should be filtered out on another level. // See Remove Service Admin spec section 122.6.3 registerTracker(reference, service); } @Override public void removedService(ServiceReference reference, Object service) { super.removedService(reference, service); clearTracker(service); } }; ...... } void registerTracker(ServiceReference reference, Object svc) { if (svc instanceof EndpointListener) { EndpointListener listener = (EndpointListener) svc; Collection<String> filters = addListener(reference, listener); triggerCallbacks(filters, listener); } } private void triggerCallbacks(Collection<String> filters, EndpointListener listener) { for (String filter : filters) { for (EndpointDescription ed : endpointDescriptions.keySet()) { triggerCallbacks(listener, filter, ed, true); } } }
2.以上步骤同上 LocalDiscovery初始化
2.以上步骤同上 LocalDiscovery初始化
- 场景4：bundle changed
1.LocalDiscovery
// BundleListener method public void bundleChanged(BundleEvent be) { switch (be.getType()) { case BundleEvent.STARTED: findDeclaredRemoteServices(be.getBundle()); break; case BundleEvent.STOPPED: removeServicesDeclaredInBundle(be.getBundle()); break; } } private void findDeclaredRemoteServices(Bundle bundle) { List<EndpointDescription> eds = LocalDiscoveryUtils.getAllEndpointDescriptions(bundle); for (EndpointDescription ed : eds) { endpointDescriptions.put(ed, bundle); addedEndpointDescription(ed); } }
2.以下步骤同LocalDiscovery初始化
// BundleListener method public void bundleChanged(BundleEvent be) { switch (be.getType()) { case BundleEvent.STARTED: findDeclaredRemoteServices(be.getBundle()); break; case BundleEvent.STOPPED: removeServicesDeclaredInBundle(be.getBundle()); break; } } private void findDeclaredRemoteServices(Bundle bundle) { List<EndpointDescription> eds = LocalDiscoveryUtils.getAllEndpointDescriptions(bundle); for (EndpointDescription ed : eds) { endpointDescriptions.put(ed, bundle); addedEndpointDescription(ed); } }
2.以下步骤同LocalDiscovery初始化
二、服务注入删除 （3种场景）
- 场景1：Bundle changed
public void bundleChanged(BundleEvent be) { switch (be.getType()) { case BundleEvent.STARTED: findDeclaredRemoteServices(be.getBundle()); break; case BundleEvent.STOPPED: removeServicesDeclaredInBundle(be.getBundle()); break; } } private void removeServicesDeclaredInBundle(Bundle bundle) { for (Iterator<Entry<EndpointDescription, Bundle>> i = endpointDescriptions.entrySet().iterator(); i.hasNext(); ) { Entry<EndpointDescription, Bundle> entry = i.next(); if (bundle.equals(entry.getValue())) { removedEndpointDescription(entry.getKey()); i.remove(); } } } private void removedEndpointDescription(EndpointDescription ed) { triggerCallbacks(ed, false); } private void triggerCallbacks(EndpointListener listener, String toMatch, EndpointDescription ed, boolean added) { if (!filterMatches(toMatch, ed)) { return; } if (added) { listener.endpointAdded(ed, toMatch); } else { listener.endpointRemoved(ed, toMatch); } }
2.org.apache.cxf.dosgi.topologymanager.EndpointListenerImpl
public void endpointRemoved(EndpointDescription epd, String filter) { LOG.info("EndpointListenerImpl: EndpointRemoved() -> "+epd); topManager.removeImportableService(filter, epd); }
3.TopologyManagerImport
public void removeImportableService(String filter, EndpointDescription epd) { synchronized (importPossibilities) { List<EndpointDescription> ips = importPossibilities.get(filter); if (ips != null) { ips.remove(epd); } else { // should not happen } } triggerImport(filter); }
public RemoteServiceAdminList(BundleContext bc) { bctx = bc; final RemoteServiceAdminList rsal = this; stRemoteServiceAdmin = new ServiceTracker(bctx, RemoteServiceAdmin.class.getName(), null) { @Override public Object addingService(ServiceReference reference) { LOG.info("Adding RemoteServiceAdmin to list of admins "); RemoteServiceAdmin rsa = (RemoteServiceAdmin)bctx.getService(reference); synchronized (rsal) { rsal.add(rsa); } LOG.info("enlisted RemoteEventAdmins: " + this.size()); triggerExportImportForRemoteSericeAdmin(rsa); return super.addingService(reference); } @Override public void removedService(ServiceReference reference, Object service) { LOG.info("TopologyManager: Removing RemoteServiceAdmin from list of admins "); synchronized (rsal) { rsal.remove(service); } // TODO: remove service exports from management structure and notify discovery stuff... removeRemoteServiceAdmin((RemoteServiceAdmin)service); LOG.info("TopologyManager: enlisted RemoteEventAdmins: " + rsal.size()); super.removedService(reference, service); } }; } protected void removeRemoteServiceAdmin(RemoteServiceAdmin service) { topManager.removeRemoteServiceAdmin(service); }
2.TopologyManager
protected void removeRemoteServiceAdmin(RemoteServiceAdmin rsa) { synchronized (exportedServices) { for (Map.Entry<ServiceReference, HashMap<RemoteServiceAdmin, Collection<ExportRegistration>>> exports : exportedServices .entrySet()) { if (exports.getValue().containsKey(rsa)) { // service was handled by this RemoteServiceAdmin Collection<ExportRegistration> endpoints = exports.getValue().get(rsa); // TODO for each notify discovery...... try { ServiceReference[] refs = Utils.getEndpointListeners(bctx); for (ServiceReference sref : refs) { notifyListenersOfRemovalIfAppropriate(sref, endpoints); } } catch (InvalidSyntaxException e) { e.printStackTrace(); } // remove all management information for the RemoteServiceAdmin exports.getValue().remove(rsa); } } } } protected void notifyListenersOfRemovalIfAppropriate(ServiceReference sref, Collection<ExportRegistration> exportRegistrations) { EndpointListener epl = (EndpointListener)bctx.getService(sref); LOG.info("TopologyManager: notifyListenerOfREMOVALIfAppropriate() "); List<Filter> filters; try { filters = Utils.normalizeScope(sref, bctx); for (ExportRegistration exReg : exportRegistrations) { // FIXME!!!!!!!!!!!!! There needs to be a better way ?!?!?! Map props = exReg.getExportReference().getExportedEndpoint().getProperties(); Dictionary d = new Hashtable(props); for (Filter filter : filters) { LOG.info("Matching: " + filter + " against " + d); } for (Filter filter : filters) { if (filter.match(d)) { LOG.info("Listener matched one of the Endpoints !!!! --> calling removed() ..."); epl.endpointRemoved(exReg.getExportReference().getExportedEndpoint(), filter .toString()); } } } } catch (InvalidSyntaxException e) { // TODO Auto-generated catch block e.printStackTrace(); } }
3.调用EndpointListenerImpl.endpointRemoved以下步骤同Bundle changed的步骤
处理通知事件 public void process(WatchedEvent event) { LOG.finer("ZooKeeper watcher callback " + event); processDelta(); } private void processDelta() { if(closed) return; if(zookeeper.getState() != ZooKeeper.States.CONNECTED){ LOG.info("zookeeper connection was already closed! Not processing changed event."); return; } try { if (zookeeper.exists(znode, false) != null) { zookeeper.getChildren(znode, this); listener.change(); }else{ LOG.fine(znode+" doesn't exist -> not processing any changes"); } } catch (Exception ke) { LOG.log(Level.SEVERE, "Error getting ZooKeeper data.", ke); } }
2.InterfaceDataMonitorListenerImpl
public synchronized void change() { Map<String, Map<String, Object>> newNodes = new HashMap<String, Map<String, Object>>(); Map<String, Map<String, Object>> prevNodes = nodes; LOG.info("Zookeeper callback on node: " + znode); processChildren(znode, newNodes, prevNodes); LOG.fine("processChildren done nodes that are missing now and need to be removed: " + prevNodes.values()); for (Map<String, Object> props : prevNodes.values()) { // whatever's left in prevNodes now has been removed from Discovery EndpointDescription epd = new EndpointDescription(props); //notifyListeners(epd, true); for (ServiceReference sref : discoveredServiceTracker.relatedServiceListeners) { if (bctx.getService(sref) instanceof EndpointListener) { EndpointListener epl = (EndpointListener)bctx.getService(sref); // return the >first< matching scope of the listener // TODO: this code also exists for the endpoint adding in the processChild() method -> // refactor ! String[] scopes = Util.getScopes(sref); for (final String currentScope : scopes) { LOG.fine("matching " + epd + " against " + currentScope); Filter f = null; try { f = FrameworkUtil.createFilter(currentScope); Dictionary d = new Properties(); Set<Map.Entry<String, Object>> entries = props.entrySet(); for (Map.Entry<String, Object> entry : entries) { d.put(entry.getKey(), entry.getValue()); } if (f.match(d)) { LOG.fine("MATCHED " + epd + "against " + currentScope); LOG.info("calling EndpointListener endpointRemoved: " + epl + "from bundle " + sref.getBundle().getSymbolicName() + " for endpoint: " + epd); epl.endpointRemoved(epd, currentScope); break; } } catch (InvalidSyntaxException e) { e.printStackTrace(); } } } } } nodes = newNodes; } private boolean processChildren(String znode, Map<String, Map<String, Object>> newNodes, Map<String, Map<String, Object>> prevNodes) { List<String> children; try { LOG.info("Processing the children of " + znode); children = zookeeper.getChildren(znode, false); boolean foundANode = false; for (String child : children) { Map<String, Object> p = processChild(znode, child, prevNodes.get(child)); if (p != null) { LOG.fine("found new node " + znode + "/[" + child + "] ( []->child ) props: " + p.values()); newNodes.put(child, p); prevNodes.remove(child); foundANode = true; } if (recursive) { String newNode = znode + '/' + child; if (processChildren(newNode, newNodes, prevNodes)) zookeeper.getChildren(newNode, parent); } } return foundANode; } catch (KeeperException e) { LOG.log(Level.SEVERE, "Problem processing Zookeeper node: " + e.getMessage(), e); } catch (InterruptedException e) { LOG.log(Level.SEVERE, "Problem processing Zookeeper node: " + e.getMessage(), e); } return false; } private Map<String, Object> processChild(String znode, String child, Map<String, Object> prevVal) { String node = znode + '/' + child; try { Stat s = zookeeper.exists(node, false); if (s.getDataLength() <= 0) { return null; } byte[] data = zookeeper.getData(node, false, null); LOG.info("Child: " + node); List<Element> elements = LocalDiscoveryUtils.getElements(new ByteArrayInputStream(data)); EndpointDescription epd = null; if (elements.size() > 0) epd = LocalDiscoveryUtils.getEndpointDescription(elements.get(0)); else { LOG.warning("No Discovery information found for node: " + node); return null; } LOG.finest("Properties: " + epd.getProperties()); if (prevVal == null) { // This guy is new notifyListeners(epd, false); } else if (!prevVal.equals(epd.getProperties())) { // TODO } return epd.getProperties(); } catch (Exception e) { LOG.log(Level.SEVERE, "Problem processing Zookeeper callback: " + e.getMessage(), e); } return null; } private Map<String, Object> processChild(String znode, String child, Map<String, Object> prevVal) { String node = znode + '/' + child; try { Stat s = zookeeper.exists(node, false); if (s.getDataLength() <= 0) { return null; } byte[] data = zookeeper.getData(node, false, null); LOG.info("Child: " + node); List<Element> elements = LocalDiscoveryUtils.getElements(new ByteArrayInputStream(data)); EndpointDescription epd = null; if (elements.size() > 0) epd = LocalDiscoveryUtils.getEndpointDescription(elements.get(0)); else { LOG.warning("No Discovery information found for node: " + node); return null; } LOG.finest("Properties: " + epd.getProperties()); if (prevVal == null) { // This guy is new notifyListeners(epd, false); } else if (!prevVal.equals(epd.getProperties())) { // TODO } return epd.getProperties(); } catch (Exception e) { LOG.log(Level.SEVERE, "Problem processing Zookeeper callback: " + e.getMessage(), e); } return null; } private void notifyListeners(EndpointDescription epd, boolean isRemoval) { System.out.println("**************** notifyListeners("+epd+" , "+isRemoval+")"); for (ServiceReference sref : discoveredServiceTracker.relatedServiceListeners) { if (bctx.getService(sref) instanceof EndpointListener) { final EndpointListener epl = (EndpointListener)bctx.getService(sref); String[] scopes = Util.getScopes(sref); for (final String currentScope : scopes) { Filter f; try { f = FrameworkUtil.createFilter(currentScope); Dictionary d = new Properties(); Map<String, Object> props = epd.getProperties(); Set<Map.Entry<String, Object>> entries = props.entrySet(); for (Map.Entry<String, Object> entry : entries) { d.put(entry.getKey(), entry.getValue()); } LOG.fine("matching " + epd + " against " + currentScope); if (f.match(d)) { LOG.fine("MATCHED " + epd + "against " + currentScope); LOG.info("scheduling EndpointListener call for listener ; " + epl + " from bundle " + sref.getBundle().getSymbolicName() + " based on scope [" + currentScope + "]"); if (isRemoval) epl.endpointRemoved(epd, currentScope); else epl.endpointAdded(epd, currentScope); break; } } catch (InvalidSyntaxException e) { LOG.warning("skipping scope [" + currentScope + "] of endpoint listener from bundle "+sref.getBundle().getSymbolicName()+" becaue it is invalid: " + e.getMessage()); } } } }
3.org.apache.cxf.dosgi.topologymanager.EndpointListenerImpl
public void endpointRemoved(EndpointDescription epd, String filter) { LOG.info("EndpointListenerImpl: EndpointRemoved() -> "+epd); topManager.removeImportableService(filter, epd); }
4.以下同 LocalDiscovery初始化。
2.org.apache.cxf.dosgi.topologymanager.EndpointListenerImpl
public void endpointRemoved(EndpointDescription epd, String filter) { LOG.info("EndpointListenerImpl: EndpointRemoved() -> "+epd); topManager.removeImportableService(filter, epd); }
3.TopologyManagerImport
public void removeImportableService(String filter, EndpointDescription epd) { synchronized (importPossibilities) { List<EndpointDescription> ips = importPossibilities.get(filter); if (ips != null) { ips.remove(epd); } else { // should not happen } } triggerImport(filter); }
- 场景2： RemoteServiceAdminList中注册的 ServiceTracker, 针对RemoteServiceAdmin.class
public RemoteServiceAdminList(BundleContext bc) { bctx = bc; final RemoteServiceAdminList rsal = this; stRemoteServiceAdmin = new ServiceTracker(bctx, RemoteServiceAdmin.class.getName(), null) { @Override public Object addingService(ServiceReference reference) { LOG.info("Adding RemoteServiceAdmin to list of admins "); RemoteServiceAdmin rsa = (RemoteServiceAdmin)bctx.getService(reference); synchronized (rsal) { rsal.add(rsa); } LOG.info("enlisted RemoteEventAdmins: " + this.size()); triggerExportImportForRemoteSericeAdmin(rsa); return super.addingService(reference); } @Override public void removedService(ServiceReference reference, Object service) { LOG.info("TopologyManager: Removing RemoteServiceAdmin from list of admins "); synchronized (rsal) { rsal.remove(service); } // TODO: remove service exports from management structure and notify discovery stuff... removeRemoteServiceAdmin((RemoteServiceAdmin)service); LOG.info("TopologyManager: enlisted RemoteEventAdmins: " + rsal.size()); super.removedService(reference, service); } }; } protected void removeRemoteServiceAdmin(RemoteServiceAdmin service) { topManager.removeRemoteServiceAdmin(service); }
2.TopologyManager
protected void removeRemoteServiceAdmin(RemoteServiceAdmin rsa) { synchronized (exportedServices) { for (Map.Entry<ServiceReference, HashMap<RemoteServiceAdmin, Collection<ExportRegistration>>> exports : exportedServices .entrySet()) { if (exports.getValue().containsKey(rsa)) { // service was handled by this RemoteServiceAdmin Collection<ExportRegistration> endpoints = exports.getValue().get(rsa); // TODO for each notify discovery...... try { ServiceReference[] refs = Utils.getEndpointListeners(bctx); for (ServiceReference sref : refs) { notifyListenersOfRemovalIfAppropriate(sref, endpoints); } } catch (InvalidSyntaxException e) { e.printStackTrace(); } // remove all management information for the RemoteServiceAdmin exports.getValue().remove(rsa); } } } } protected void notifyListenersOfRemovalIfAppropriate(ServiceReference sref, Collection<ExportRegistration> exportRegistrations) { EndpointListener epl = (EndpointListener)bctx.getService(sref); LOG.info("TopologyManager: notifyListenerOfREMOVALIfAppropriate() "); List<Filter> filters; try { filters = Utils.normalizeScope(sref, bctx); for (ExportRegistration exReg : exportRegistrations) { // FIXME!!!!!!!!!!!!! There needs to be a better way ?!?!?! Map props = exReg.getExportReference().getExportedEndpoint().getProperties(); Dictionary d = new Hashtable(props); for (Filter filter : filters) { LOG.info("Matching: " + filter + " against " + d); } for (Filter filter : filters) { if (filter.match(d)) { LOG.info("Listener matched one of the Endpoints !!!! --> calling removed() ..."); epl.endpointRemoved(exReg.getExportReference().getExportedEndpoint(), filter .toString()); } } } } catch (InvalidSyntaxException e) { // TODO Auto-generated catch block e.printStackTrace(); } }
3.调用EndpointListenerImpl.endpointRemoved以下步骤同Bundle changed的步骤
- 场景3：zookeeper server的反向通知机制，意思是暴露服务的远程bundle从zookeeper上删除服务节点。可以看做服务变化service changed
处理通知事件 public void process(WatchedEvent event) { LOG.finer("ZooKeeper watcher callback " + event); processDelta(); } private void processDelta() { if(closed) return; if(zookeeper.getState() != ZooKeeper.States.CONNECTED){ LOG.info("zookeeper connection was already closed! Not processing changed event."); return; } try { if (zookeeper.exists(znode, false) != null) { zookeeper.getChildren(znode, this); listener.change(); }else{ LOG.fine(znode+" doesn't exist -> not processing any changes"); } } catch (Exception ke) { LOG.log(Level.SEVERE, "Error getting ZooKeeper data.", ke); } }
2.InterfaceDataMonitorListenerImpl
public synchronized void change() { Map<String, Map<String, Object>> newNodes = new HashMap<String, Map<String, Object>>(); Map<String, Map<String, Object>> prevNodes = nodes; LOG.info("Zookeeper callback on node: " + znode); processChildren(znode, newNodes, prevNodes); LOG.fine("processChildren done nodes that are missing now and need to be removed: " + prevNodes.values()); for (Map<String, Object> props : prevNodes.values()) { // whatever's left in prevNodes now has been removed from Discovery EndpointDescription epd = new EndpointDescription(props); //notifyListeners(epd, true); for (ServiceReference sref : discoveredServiceTracker.relatedServiceListeners) { if (bctx.getService(sref) instanceof EndpointListener) { EndpointListener epl = (EndpointListener)bctx.getService(sref); // return the >first< matching scope of the listener // TODO: this code also exists for the endpoint adding in the processChild() method -> // refactor ! String[] scopes = Util.getScopes(sref); for (final String currentScope : scopes) { LOG.fine("matching " + epd + " against " + currentScope); Filter f = null; try { f = FrameworkUtil.createFilter(currentScope); Dictionary d = new Properties(); Set<Map.Entry<String, Object>> entries = props.entrySet(); for (Map.Entry<String, Object> entry : entries) { d.put(entry.getKey(), entry.getValue()); } if (f.match(d)) { LOG.fine("MATCHED " + epd + "against " + currentScope); LOG.info("calling EndpointListener endpointRemoved: " + epl + "from bundle " + sref.getBundle().getSymbolicName() + " for endpoint: " + epd); epl.endpointRemoved(epd, currentScope); break; } } catch (InvalidSyntaxException e) { e.printStackTrace(); } } } } } nodes = newNodes; } private boolean processChildren(String znode, Map<String, Map<String, Object>> newNodes, Map<String, Map<String, Object>> prevNodes) { List<String> children; try { LOG.info("Processing the children of " + znode); children = zookeeper.getChildren(znode, false); boolean foundANode = false; for (String child : children) { Map<String, Object> p = processChild(znode, child, prevNodes.get(child)); if (p != null) { LOG.fine("found new node " + znode + "/[" + child + "] ( []->child ) props: " + p.values()); newNodes.put(child, p); prevNodes.remove(child); foundANode = true; } if (recursive) { String newNode = znode + '/' + child; if (processChildren(newNode, newNodes, prevNodes)) zookeeper.getChildren(newNode, parent); } } return foundANode; } catch (KeeperException e) { LOG.log(Level.SEVERE, "Problem processing Zookeeper node: " + e.getMessage(), e); } catch (InterruptedException e) { LOG.log(Level.SEVERE, "Problem processing Zookeeper node: " + e.getMessage(), e); } return false; } private Map<String, Object> processChild(String znode, String child, Map<String, Object> prevVal) { String node = znode + '/' + child; try { Stat s = zookeeper.exists(node, false); if (s.getDataLength() <= 0) { return null; } byte[] data = zookeeper.getData(node, false, null); LOG.info("Child: " + node); List<Element> elements = LocalDiscoveryUtils.getElements(new ByteArrayInputStream(data)); EndpointDescription epd = null; if (elements.size() > 0) epd = LocalDiscoveryUtils.getEndpointDescription(elements.get(0)); else { LOG.warning("No Discovery information found for node: " + node); return null; } LOG.finest("Properties: " + epd.getProperties()); if (prevVal == null) { // This guy is new notifyListeners(epd, false); } else if (!prevVal.equals(epd.getProperties())) { // TODO } return epd.getProperties(); } catch (Exception e) { LOG.log(Level.SEVERE, "Problem processing Zookeeper callback: " + e.getMessage(), e); } return null; } private Map<String, Object> processChild(String znode, String child, Map<String, Object> prevVal) { String node = znode + '/' + child; try { Stat s = zookeeper.exists(node, false); if (s.getDataLength() <= 0) { return null; } byte[] data = zookeeper.getData(node, false, null); LOG.info("Child: " + node); List<Element> elements = LocalDiscoveryUtils.getElements(new ByteArrayInputStream(data)); EndpointDescription epd = null; if (elements.size() > 0) epd = LocalDiscoveryUtils.getEndpointDescription(elements.get(0)); else { LOG.warning("No Discovery information found for node: " + node); return null; } LOG.finest("Properties: " + epd.getProperties()); if (prevVal == null) { // This guy is new notifyListeners(epd, false); } else if (!prevVal.equals(epd.getProperties())) { // TODO } return epd.getProperties(); } catch (Exception e) { LOG.log(Level.SEVERE, "Problem processing Zookeeper callback: " + e.getMessage(), e); } return null; } private void notifyListeners(EndpointDescription epd, boolean isRemoval) { System.out.println("**************** notifyListeners("+epd+" , "+isRemoval+")"); for (ServiceReference sref : discoveredServiceTracker.relatedServiceListeners) { if (bctx.getService(sref) instanceof EndpointListener) { final EndpointListener epl = (EndpointListener)bctx.getService(sref); String[] scopes = Util.getScopes(sref); for (final String currentScope : scopes) { Filter f; try { f = FrameworkUtil.createFilter(currentScope); Dictionary d = new Properties(); Map<String, Object> props = epd.getProperties(); Set<Map.Entry<String, Object>> entries = props.entrySet(); for (Map.Entry<String, Object> entry : entries) { d.put(entry.getKey(), entry.getValue()); } LOG.fine("matching " + epd + " against " + currentScope); if (f.match(d)) { LOG.fine("MATCHED " + epd + "against " + currentScope); LOG.info("scheduling EndpointListener call for listener ; " + epl + " from bundle " + sref.getBundle().getSymbolicName() + " based on scope [" + currentScope + "]"); if (isRemoval) epl.endpointRemoved(epd, currentScope); else epl.endpointAdded(epd, currentScope); break; } } catch (InvalidSyntaxException e) { LOG.warning("skipping scope [" + currentScope + "] of endpoint listener from bundle "+sref.getBundle().getSymbolicName()+" becaue it is invalid: " + e.getMessage()); } } } }
3.org.apache.cxf.dosgi.topologymanager.EndpointListenerImpl
public void endpointRemoved(EndpointDescription epd, String filter) { LOG.info("EndpointListenerImpl: EndpointRemoved() -> "+epd); topManager.removeImportableService(filter, epd); }
4.以下同 LocalDiscovery初始化。
总结：
类似服务的暴露，都是通过bundle change,service change,以及初始化来完成remote service 的import 和delete，将proxy service register到本地osgi容器中。
类似资料：