You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@karaf.apache.org by jb...@apache.org on 2015/09/14 14:53:57 UTC
[2/2] karaf-cellar git commit: [KARAF-3981] Improve synchronizers
[KARAF-3981] Improve synchronizers
Project: http://git-wip-us.apache.org/repos/asf/karaf-cellar/repo
Commit: http://git-wip-us.apache.org/repos/asf/karaf-cellar/commit/4a2996b8
Tree: http://git-wip-us.apache.org/repos/asf/karaf-cellar/tree/4a2996b8
Diff: http://git-wip-us.apache.org/repos/asf/karaf-cellar/diff/4a2996b8
Branch: refs/heads/cellar-2.3.x
Commit: 4a2996b889a766bd7937b394cd8b42fbbc906455
Parents: 470dd22
Author: Jean-Baptiste Onofré <jb...@apache.org>
Authored: Mon Sep 14 14:53:22 2015 +0200
Committer: Jean-Baptiste Onofré <jb...@apache.org>
Committed: Mon Sep 14 14:53:22 2015 +0200
----------------------------------------------------------------------
assembly/src/main/resources/features.xml | 2 +-
assembly/src/main/resources/groups.cfg | 18 +--
.../karaf/cellar/bundle/BundleEventHandler.java | 93 ++++++-----
.../karaf/cellar/bundle/BundleSupport.java | 76 +++++----
.../karaf/cellar/bundle/BundleSynchronizer.java | 157 ++++++++++++-------
.../cellar/bundle/LocalBundleListener.java | 33 ++--
.../bundle/shell/InstallBundleCommand.java | 11 +-
.../cellar/bundle/shell/ListBundleCommand.java | 17 +-
.../cellar/bundle/shell/StartBundleCommand.java | 6 +-
.../cellar/bundle/shell/StopBundleCommand.java | 6 +-
.../bundle/shell/UninstallBundleCommand.java | 4 +-
.../resources/OSGI-INF/blueprint/blueprint.xml | 1 +
.../config/ConfigurationEventHandler.java | 9 +-
.../config/ConfigurationSynchronizer.java | 71 +++++++--
.../config/LocalConfigurationListener.java | 2 +-
.../resources/OSGI-INF/blueprint/blueprint.xml | 1 +
.../cellar/features/FeaturesEventHandler.java | 9 +-
.../cellar/features/FeaturesSynchronizer.java | 149 +++++++++++++-----
.../cellar/features/LocalFeaturesListener.java | 12 +-
.../resources/OSGI-INF/blueprint/blueprint.xml | 1 +
.../karaf/cellar/hazelcast/QueueProducer.java | 4 +-
.../karaf/cellar/hazelcast/TopicProducer.java | 4 +-
.../internal/CellarBundleMBeanImpl.java | 31 ++--
.../karaf/cellar/obr/ObrBundleEventHandler.java | 13 +-
.../karaf/cellar/obr/ObrUrlEventHandler.java | 9 +-
.../karaf/cellar/obr/ObrUrlSynchronizer.java | 50 ++++--
.../resources/OSGI-INF/blueprint/blueprint.xml | 1 +
27 files changed, 519 insertions(+), 271 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/assembly/src/main/resources/features.xml
----------------------------------------------------------------------
diff --git a/assembly/src/main/resources/features.xml b/assembly/src/main/resources/features.xml
index 29f596e..554f0de 100644
--- a/assembly/src/main/resources/features.xml
+++ b/assembly/src/main/resources/features.xml
@@ -26,7 +26,7 @@
<feature name="hazelcast" description="In memory data grid" version="${hazelcast.version}" resolver="(obr)">
<configfile finalname="/etc/hazelcast.xml">mvn:org.apache.karaf.cellar/apache-karaf-cellar/${project.version}/xml/hazelcast</configfile>
- <bundle start-level="70">mvn:com.hazelcast/hazelcast-all/${hazelcast.version}</bundle>
+ <bundle start-level="70">mvn:com.hazelcast/hazelcast/${hazelcast.version}</bundle>
</feature>
<feature name="cellar-hazelcast" description="Cellar implementation based on Hazelcast" version="${project.version}" resolver="(obr)">
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/assembly/src/main/resources/groups.cfg
----------------------------------------------------------------------
diff --git a/assembly/src/main/resources/groups.cfg b/assembly/src/main/resources/groups.cfg
index 0f397c0..53d247d 100644
--- a/assembly/src/main/resources/groups.cfg
+++ b/assembly/src/main/resources/groups.cfg
@@ -17,17 +17,13 @@ default.bundle.blacklist.outbound = *.xml
default.config.whitelist.inbound = *
default.config.whitelist.outbound = *
default.config.blacklist.inbound = org.apache.felix.fileinstall*, \
- org.apache.karaf.cellar*, \
org.apache.karaf.management, \
org.apache.karaf.shell, \
- org.ops4j.pax.logging, \
org.ops4j.pax.web, \
org.apache.aries.transaction
default.config.blacklist.outbound = org.apache.felix.fileinstall*, \
- org.apache.karaf.cellar*, \
org.apache.karaf.management, \
org.apache.karaf.shell, \
- org.ops4j.pax.logging, \
org.ops4j.pax.web, \
org.apache.aries.transaction
@@ -43,11 +39,15 @@ default.features.blacklist.outbound = none
# The following properties define the behavior to use when the node joins the cluster (the usage of the bootstrap
# synchronizer), per cluster group and per resource.
# The following values are accepted:
-# disabled: means that the synchronizer is not used, meaning the node or the cluster are not updated at all
-# cluster: if the node is the first one in the cluster, it pushes its local state to the cluster, else it's not the
-# first node of the cluster, the node will update its local state with the cluster one (meaning that the cluster
-# is the master)
-# node: in this case, the node is the master, it means that the cluster state will be overwritten by the node state.
+# disabled: means that the synchronizer doesn't sync cluster group and node states
+# cluster: the synchronizer retrieves the state from the cluster group first (pull first), and push the node the state
+# to the cluster group after (push after)
+# node: the synchronizer push the node state to the cluster group (push first), and pull the state from the cluster group
+# after (pull after)
+# clusterOnly: the cluster is the "master", the node only retrieves and applies the cluster group state, nothing is
+# pushed to the cluster group
+# nodeOnly: the node is the "master", the node pushes his state to the cluster group, nothing is pulled from the
+# cluster group
#
default.bundle.sync = cluster
default.config.sync = cluster
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleEventHandler.java
----------------------------------------------------------------------
diff --git a/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleEventHandler.java b/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleEventHandler.java
index 1f2ae28..ee34cf2 100644
--- a/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleEventHandler.java
+++ b/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleEventHandler.java
@@ -13,8 +13,6 @@
*/
package org.apache.karaf.cellar.bundle;
-import java.util.List;
-
import org.apache.karaf.cellar.core.Configurations;
import org.apache.karaf.cellar.core.control.BasicSwitch;
import org.apache.karaf.cellar.core.control.Switch;
@@ -22,74 +20,89 @@ import org.apache.karaf.cellar.core.control.SwitchStatus;
import org.apache.karaf.cellar.core.event.EventHandler;
import org.apache.karaf.cellar.core.event.EventType;
import org.apache.karaf.features.Feature;
-import org.osgi.framework.BundleEvent;
+import org.osgi.framework.Bundle;
import org.osgi.framework.BundleException;
import org.osgi.service.cm.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.List;
+
/**
- * This class handles the received cluster bundle event.
+ * The BundleEventHandler is responsible to process received cluster event for bundles.
*/
public class BundleEventHandler extends BundleSupport implements EventHandler<ClusterBundleEvent> {
- static final Logger LOGGER = LoggerFactory.getLogger(BundleEventHandler.class);
+ private static final transient Logger LOGGER = LoggerFactory.getLogger(BundleEventHandler.class);
public static final String SWITCH_ID = "org.apache.karaf.cellar.bundle.handler";
private final Switch eventSwitch = new BasicSwitch(SWITCH_ID);
-
+
/**
- * Handle a received cluster bundle event.
+ * Handle received bundle cluster events.
*
- * @param event the received cluster bundle event.
+ * @param event the received bundle cluster event.
*/
@Override
public void handle(ClusterBundleEvent event) {
+
// check if the handler switch is ON
if (this.getSwitch().getStatus().equals(SwitchStatus.OFF)) {
LOGGER.debug("CELLAR BUNDLE: {} switch is OFF, cluster event is not handled", SWITCH_ID);
return;
}
- // check if the node is local
+ if (groupManager == null) {
+ //in rare cases for example right after installation this happens!
+ LOGGER.error("CELLAR BUNDLE: retrieved event {} while groupManager is not available yet!", event);
+ return;
+ }
+
+ // check if the group is local
if (!groupManager.isLocalGroup(event.getSourceGroup().getName())) {
- LOGGER.debug("CELLAR BUNDLE: node is not part of the event cluster group");
+ LOGGER.debug("CELLAR BUNDLE: node is not part of the event cluster group {}", event.getSourceGroup().getName());
return;
}
try {
- // check if the cluster bundle event is allowed inbound
+ // check if it's not a "local" event
+ if (event.getSourceNode() != null && event.getSourceNode().getId().equalsIgnoreCase(clusterManager.getNode().getId())) {
+ LOGGER.trace("CELLAR BUNDLE: cluster event is local (coming from local synchronizer or listener)");
+ return;
+ }
+ // check if the pid is marked as local.
if (isAllowed(event.getSourceGroup(), Constants.CATEGORY, event.getLocation(), EventType.INBOUND)) {
- // check the features first
- List<Feature> matchingFeatures = retrieveFeature(event.getLocation());
- for (Feature feature : matchingFeatures) {
- if (!isAllowed(event.getSourceGroup(), "features", feature.getName(), EventType.INBOUND)) {
- LOGGER.debug("CELLAR BUNDLE: bundle {} is contained in the feature {} marked as BLOCKED INBOUND in cluster group {}", event.getLocation(), feature.getName(), event.getSourceGroup().getName());
- return;
- }
- }
- if (event.getType() == BundleEvent.INSTALLED) {
- LOGGER.debug("CELLAR BUNDLE: installing bundle {} from {}", event.getId(), event.getLocation());
+ // check the features first
+ List<Feature> matchingFeatures = retrieveFeature(event.getLocation());
+ for (Feature feature : matchingFeatures) {
+ if (!isAllowed(event.getSourceGroup(), "features", feature.getName(), EventType.INBOUND)) {
+ LOGGER.trace("CELLAR BUNDLE: bundle {} is contained in feature {} marked BLOCKED INBOUND for cluster group {}", event.getLocation(), feature.getName(), event.getSourceGroup().getName());
+ return;
+ }
+ }
+ if (event.getType() == Bundle.INSTALLED) {
installBundleFromLocation(event.getLocation());
- } else if (event.getType() == BundleEvent.UNINSTALLED) {
- LOGGER.debug("CELLAR BUNDLE: un-installing bundle {}/{}", event.getSymbolicName(), event.getVersion());
+ LOGGER.debug("CELLAR BUNDLE: installing {}/{}", event.getSymbolicName(), event.getVersion());
+ } else if (event.getType() == Bundle.UNINSTALLED) {
uninstallBundle(event.getSymbolicName(), event.getVersion());
- } else if (event.getType() == BundleEvent.STARTED) {
- LOGGER.debug("CELLAR BUNDLE: starting bundle {}/{}", event.getSymbolicName(), event.getVersion());
+ LOGGER.debug("CELLAR BUNDLE: uninstalling {}/{}", event.getSymbolicName(), event.getVersion());
+ } else if (event.getType() == Bundle.ACTIVE) {
+ if (!isInstalled(event.getLocation())) {
+ installBundleFromLocation(event.getLocation());
+ }
startBundle(event.getSymbolicName(), event.getVersion());
- } else if (event.getType() == BundleEvent.STOPPED) {
- LOGGER.debug("CELLAR BUNDLE: stopping bundle {}/{}", event.getSymbolicName(), event.getVersion());
+ LOGGER.debug("CELLAR BUNDLE: starting {}/{}", event.getSymbolicName(), event.getVersion());
+ } else if (event.getType() == Bundle.RESOLVED) {
stopBundle(event.getSymbolicName(), event.getVersion());
- } else if (event.getType() == BundleEvent.UPDATED) {
- LOGGER.debug("CELLAR BUNDLE: updating bundle {}/{}", event.getSymbolicName(), event.getVersion());
- updateBundle(event.getSymbolicName(), event.getVersion());
+ LOGGER.debug("CELLAR BUNDLE: stopping {}/{}", event.getSymbolicName(), event.getVersion());
}
- } else LOGGER.warn("CELLAR BUNDLE: bundle {} is marked BLOCKED INBOUND in cluster group {}", event.getLocation(), event.getSourceGroup().getName());
+ } else
+ LOGGER.trace("CELLAR BUNDLE: bundle {} is marked BLOCKED INBOUND for cluster group {}", event.getSymbolicName(), event.getSourceGroup().getName());
} catch (BundleException e) {
- LOGGER.error("CELLAR BUNDLE: failed to handle bundle event", e);
+ LOGGER.error("CELLAR BUNDLE: failed to install bundle {}/{}.", new Object[]{event.getSymbolicName(), event.getVersion()}, e);
} catch (Exception e) {
- LOGGER.error("CELLAR BUNDLE: failed to handle bundle event", e);
+ LOGGER.error("CELLAR BUNDLE: failed to handle bundle event", e);
}
}
@@ -102,15 +115,15 @@ public class BundleEventHandler extends BundleSupport implements EventHandler<Cl
}
/**
- * Get the current handler switch.
+ * Get the cluster bundle event handler switch.
*
- * @return the bundle event handler switch.
+ * @return the cluster bundle event handler switch.
*/
@Override
public Switch getSwitch() {
// load the switch status from the config
try {
- Configuration configuration = configurationAdmin.getConfiguration(Configurations.NODE);
+ Configuration configuration = configurationAdmin.getConfiguration(Configurations.NODE, null);
if (configuration != null) {
Boolean status = new Boolean((String) configuration.getProperties().get(Configurations.HANDLER + "." + this.getClass().getName()));
if (status) {
@@ -126,13 +139,13 @@ public class BundleEventHandler extends BundleSupport implements EventHandler<Cl
}
/**
- * Get the cluster event type handled by this handler.
+ * Get the cluster event type.
*
- * @return the cluster bundle event managed by this handler.
+ * @return the cluster bundle event type.
*/
@Override
public Class<ClusterBundleEvent> getType() {
return ClusterBundleEvent.class;
}
-
-}
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleSupport.java
----------------------------------------------------------------------
diff --git a/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleSupport.java b/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleSupport.java
index 91b8bd9..e27a53b 100644
--- a/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleSupport.java
+++ b/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleSupport.java
@@ -25,29 +25,49 @@ import java.util.ArrayList;
import java.util.List;
/**
- * Generic class to manage bundles.
+ * Generic Cellar bundle support.
*/
public class BundleSupport extends CellarSupport {
protected BundleContext bundleContext;
- private FeaturesService featuresService;
+ private FeaturesService featuresService;
/**
* Locally install a bundle.
*
* @param location the bundle location.
- * @throws BundleException in case of install failure.
+ * @throws BundleException in case of installation failure.
*/
public void installBundleFromLocation(String location) throws BundleException {
getBundleContext().installBundle(location);
}
+ public boolean isInstalled(String location) {
+ Bundle[] bundles = getBundleContext().getBundles();
+ for (Bundle bundle : bundles) {
+ if (bundle.getLocation().equals(location)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ public boolean isStarted(String location) {
+ Bundle[] bundles = getBundleContext().getBundles();
+ for (Bundle bundle : bundles) {
+ if (bundle.getLocation().equals(location) && (bundle.getState() == Bundle.ACTIVE)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
/**
* Locally uninstall a bundle.
*
* @param symbolicName the bundle symbolic name.
* @param version the bundle version.
- * @throws BundleException in case of uninstall failure.
+ * @throws BundleException in case of un-installation failure.
*/
public void uninstallBundle(String symbolicName, String version) throws BundleException {
Bundle[] bundles = getBundleContext().getBundles();
@@ -115,27 +135,27 @@ public class BundleSupport extends CellarSupport {
}
/**
- * Get the features where a bundle is belonging.
+ * Get the list of features where the bundle is belonging.
*
* @param bundleLocation the bundle location.
- * @return the list of features where the bundle is present.
+ * @return the list of feature where the bundle is present.
* @throws Exception in case of retrieval failure.
*/
- protected List<Feature> retrieveFeature(String bundleLocation) throws Exception {
- Feature[] features = featuresService.listFeatures();
- List<Feature> matchingFeatures = new ArrayList<Feature>();
- for (Feature feature : features) {
- List<BundleInfo> bundles = feature.getBundles();
- for (BundleInfo bundleInfo : bundles) {
- String location = bundleInfo.getLocation();
- if (location.equalsIgnoreCase(bundleLocation)) {
- matchingFeatures.add(feature);
- LOGGER.debug("CELLAR BUNDLE: found feature {} containing bundle {}", feature.getName(), bundleLocation);
- }
- }
- }
- return matchingFeatures;
- }
+ protected List<Feature> retrieveFeature(String bundleLocation) throws Exception {
+ Feature[] features = featuresService.listFeatures();
+ List<Feature> matchingFeatures = new ArrayList<Feature>();
+ for (Feature feature : features) {
+ List<BundleInfo> bundles = feature.getBundles();
+ for (BundleInfo bundleInfo : bundles) {
+ String location = bundleInfo.getLocation();
+ if (location.equalsIgnoreCase(bundleLocation)) {
+ matchingFeatures.add(feature);
+ LOGGER.debug("CELLAR BUNDLE: found a feature {} containing bundle {}", feature.getName(), bundleLocation);
+ }
+ }
+ }
+ return matchingFeatures;
+ }
public BundleContext getBundleContext() {
return this.bundleContext;
@@ -145,12 +165,12 @@ public class BundleSupport extends CellarSupport {
this.bundleContext = bundleContext;
}
- public FeaturesService getFeaturesService() {
- return featuresService;
- }
+ public FeaturesService getFeaturesService() {
+ return featuresService;
+ }
- public void setFeaturesService(FeaturesService featureService) {
- this.featuresService = featureService;
- }
+ public void setFeaturesService(FeaturesService featureService) {
+ this.featuresService = featureService;
+ }
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleSynchronizer.java
----------------------------------------------------------------------
diff --git a/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleSynchronizer.java b/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleSynchronizer.java
index 9c18ab6..362606e 100644
--- a/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleSynchronizer.java
+++ b/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleSynchronizer.java
@@ -21,7 +21,6 @@ import org.apache.karaf.cellar.core.event.EventProducer;
import org.apache.karaf.cellar.core.event.EventType;
import org.osgi.framework.Bundle;
import org.osgi.framework.BundleContext;
-import org.osgi.framework.BundleEvent;
import org.osgi.framework.BundleException;
import org.osgi.framework.BundleReference;
import org.osgi.service.cm.Configuration;
@@ -34,14 +33,22 @@ import java.util.Map;
import java.util.Set;
/**
- * BundleSynchronizer is used when Cellar starts or when a node joins a cluster group.
- * The purpose is synchronize bundles local state with the states in cluster groups.
+ * The BundleSynchronizer is called when Cellar starts or a node joins a cluster group.
+ * The purpose is to synchronize bundles local state with the states in the cluster groups.
*/
public class BundleSynchronizer extends BundleSupport implements Synchronizer {
private static final transient Logger LOGGER = LoggerFactory.getLogger(BundleSynchronizer.class);
+ private EventProducer eventProducer;
+
+ public void setEventProducer(EventProducer eventProducer) {
+ this.eventProducer = eventProducer;
+ }
+
public void init() {
+ if (groupManager == null)
+ return;
Set<Group> groups = groupManager.listLocalGroups();
if (groups != null && !groups.isEmpty()) {
for (Group group : groups) {
@@ -62,19 +69,32 @@ public class BundleSynchronizer extends BundleSupport implements Synchronizer {
@Override
public void sync(Group group) {
String policy = getSyncPolicy(group);
- if (policy != null && policy.equalsIgnoreCase("cluster")) {
- LOGGER.debug("CELLAR BUNDLE: sync policy is set as 'cluster' for cluster group " + group.getName());
- if (clusterManager.listNodesByGroup(group).size() == 1 && clusterManager.listNodesByGroup(group).contains(clusterManager.getNode())) {
- LOGGER.debug("CELLAR BUNDLE: node is the first and only member of the group, pushing state");
- push(group);
- } else {
- LOGGER.debug("CELLAR BUNDLE: pulling state");
- pull(group);
- }
+ if (policy == null) {
+ LOGGER.warn("CELLAR BUNDLE: sync policy is not defined for cluster group {}", group.getName());
}
- if (policy != null && policy.equalsIgnoreCase("node")) {
- LOGGER.debug("CELLAR BUNDLE: sync policy is set as 'node' for cluster group " + group.getName());
+ if (policy.equalsIgnoreCase("cluster")) {
+ LOGGER.debug("CELLAR BUNDLE: sync policy set as 'cluster' for cluster group {}", group.getName());
+ LOGGER.debug("CELLAR BUNDLE: updating node from the cluster (pull first)");
+ pull(group);
+ LOGGER.debug("CELLAR BUNDLE: updating cluster from the local node (push after)");
+ push(group);
+ } else if (policy.equalsIgnoreCase("node")) {
+ LOGGER.debug("CELLAR BUNDLE: sync policy set as 'node' for cluster group {}", group.getName());
+ LOGGER.debug("CELLAR BUNDLE: updating cluster from the local node (push first)");
+ push(group);
+ LOGGER.debug("CELLAR BUNDLE: updating node from the cluster (pull after)");
+ pull(group);
+ } else if (policy.equalsIgnoreCase("clusterOnly")) {
+ LOGGER.debug("CELLAR BUNDLE: sync policy set as 'clusterOnly' for cluster group {}", group.getName());
+ LOGGER.debug("CELLAR BUNDLE: updating node from the cluster (pull only)");
+ pull(group);
+ } else if (policy.equalsIgnoreCase("nodeOnly")) {
+ LOGGER.debug("CELLAR BUNDLE: sync policy set as 'nodeOnly' for cluster group {}", group.getName());
+ LOGGER.debug("CELLAR BUNDLE: updating cluster from the local node (push only)");
push(group);
+ } else {
+ LOGGER.debug("CELLAR BUNDLE: sync policy set as 'disabled' for cluster group {}", group.getName());
+ LOGGER.debug("CELLAR BUNDLE: no sync");
}
}
@@ -91,6 +111,7 @@ public class BundleSynchronizer extends BundleSupport implements Synchronizer {
Map<String, BundleState> clusterBundles = clusterManager.getMap(Constants.BUNDLE_MAP + Configurations.SEPARATOR + groupName);
ClassLoader originalClassLoader = Thread.currentThread().getContextClassLoader();
+
try {
Thread.currentThread().setContextClassLoader(getClass().getClassLoader());
for (Map.Entry<String, BundleState> entry : clusterBundles.entrySet()) {
@@ -105,17 +126,30 @@ public class BundleSynchronizer extends BundleSupport implements Synchronizer {
String bundleLocation = state.getLocation();
if (isAllowed(group, Constants.CATEGORY, bundleLocation, EventType.INBOUND)) {
try {
- if (state.getStatus() == BundleEvent.INSTALLED) {
- installBundleFromLocation(state.getLocation());
- } else if (state.getStatus() == BundleEvent.STARTED) {
- installBundleFromLocation(state.getLocation());
- startBundle(symbolicName, version);
+ if (state.getStatus() == Bundle.INSTALLED) {
+ if (!isInstalled(state.getLocation())) {
+ LOGGER.debug("CELLAR BUNDLE: installing bundle located {} on node", state.getLocation());
+ installBundleFromLocation(state.getLocation());
+ } else {
+ LOGGER.debug("CELLAR BUNDLE: bundle located {} already installed on node", state.getLocation());
+ }
+ } else if (state.getStatus() == Bundle.ACTIVE) {
+ if (!isInstalled(state.getLocation())) {
+ LOGGER.debug("CELLAR BUNDLE: installing bundle located {} on node", state.getLocation());
+ installBundleFromLocation(state.getLocation());
+ }
+ if (!isStarted(state.getLocation())) {
+ LOGGER.debug("CELLAR BUNDLE: starting bundle {}/{} on node", symbolicName, version);
+ startBundle(symbolicName, version);
+ } else {
+ LOGGER.debug("CELLAR BUNDLE: bundle located {} already started on node", state.getLocation());
+ }
}
} catch (BundleException e) {
LOGGER.error("CELLAR BUNDLE: failed to pull bundle {}", id, e);
}
} else
- LOGGER.debug("CELLAR BUNDLE: bundle {} is marked BLOCKED INBOUND for cluster group {}", bundleLocation, groupName);
+ LOGGER.trace("CELLAR BUNDLE: bundle {} is marked BLOCKED INBOUND for cluster group {}", bundleLocation, groupName);
}
}
}
@@ -133,6 +167,11 @@ public class BundleSynchronizer extends BundleSupport implements Synchronizer {
@Override
public void push(Group group) {
+ if (eventProducer.getSwitch().getStatus().equals(SwitchStatus.OFF)) {
+ LOGGER.warn("CELLAR BUNDLE: cluster event producer is OFF");
+ return;
+ }
+
if (group != null) {
String groupName = group.getName();
LOGGER.debug("CELLAR BUNDLE: pushing bundles to cluster group {}", groupName);
@@ -148,52 +187,52 @@ public class BundleSynchronizer extends BundleSupport implements Synchronizer {
for (Bundle bundle : bundles) {
long bundleId = bundle.getBundleId();
String symbolicName = bundle.getSymbolicName();
- String version = bundle.getHeaders().get("Bundle-Version").toString();
+ String version = bundle.getHeaders().get(org.osgi.framework.Constants.BUNDLE_VERSION);
String bundleLocation = bundle.getLocation();
int status = bundle.getState();
String id = symbolicName + "/" + version;
- // check if the bundle is allowed outbound
+ // check if the pid is marked as local.
if (isAllowed(group, Constants.CATEGORY, bundleLocation, EventType.OUTBOUND)) {
-
- BundleState bundleState = new BundleState();
- bundleState.setId(bundleId);
- // get the bundle name or location.
- String name = (String) bundle.getHeaders().get(org.osgi.framework.Constants.BUNDLE_NAME);
- // if there is no name, then default to symbolic name.
- name = (name == null) ? symbolicName : name;
- // if there is no symbolic name, resort to location.
- name = (name == null) ? bundle.getLocation() : name;
- bundleState.setName(name);
- bundleState.setSymbolicName(symbolicName);
- bundleState.setVersion(version);
-
- bundleState.setLocation(bundleLocation);
-
- if (status == Bundle.ACTIVE)
- status = BundleEvent.STARTED;
- if (status == Bundle.INSTALLED)
- status = BundleEvent.INSTALLED;
- if (status == Bundle.RESOLVED)
- status = BundleEvent.RESOLVED;
- if (status == Bundle.STARTING)
- status = BundleEvent.STARTING;
- if (status == Bundle.UNINSTALLED)
- status = BundleEvent.UNINSTALLED;
- if (status == Bundle.STOPPING)
- status = BundleEvent.STARTED;
-
- bundleState.setStatus(status);
-
- BundleState existingState = clusterBundles.get(id);
-
- if (existingState == null || !existingState.getLocation().equals(bundleState.getLocation()) || existingState.getStatus() != bundleState.getStatus()) {
- // update the bundle in the cluster group
+ if (!clusterBundles.containsKey(id)) {
+ LOGGER.debug("CELLAR BUNDLE: deploying bundle {} on the cluster", id);
+ BundleState bundleState = new BundleState();
+ // get the bundle name or location.
+ String name = (String) bundle.getHeaders().get(org.osgi.framework.Constants.BUNDLE_NAME);
+ // if there is no name, then default to symbolic name.
+ name = (name == null) ? symbolicName : name;
+ // if there is no symbolic name, resort to location.
+ name = (name == null) ? bundle.getLocation() : name;
+ bundleState.setId(bundleId);
+ bundleState.setName(name);
+ bundleState.setSymbolicName(symbolicName);
+ bundleState.setVersion(version);
+ bundleState.setLocation(bundleLocation);
+ bundleState.setStatus(status);
+ // update cluster state
clusterBundles.put(id, bundleState);
+ // send cluster event
+ ClusterBundleEvent clusterEvent = new ClusterBundleEvent(symbolicName, version, bundleLocation, status);
+ clusterEvent.setSourceGroup(group);
+ clusterEvent.setSourceNode(clusterManager.getNode());
+ eventProducer.produce(clusterEvent);
+ } else {
+ BundleState bundleState = clusterBundles.get(id);
+ if (bundleState.getStatus() != status) {
+ LOGGER.debug("CELLAR BUNDLE: updating bundle {} on the cluster", id);
+ // update cluster state
+ bundleState.setStatus(status);
+ clusterBundles.put(id, bundleState);
+ // send cluster event
+ ClusterBundleEvent clusterEvent = new ClusterBundleEvent(symbolicName, version, bundleLocation, status);
+ clusterEvent.setSourceGroup(group);
+ clusterEvent.setSourceNode(clusterManager.getNode());
+ eventProducer.produce(clusterEvent);
+ }
}
} else
- LOGGER.debug("CELLAR BUNDLE: bundle {} is marked as BLOCKED OUTBOUND for cluster group {}", bundleLocation, groupName);
+ LOGGER.trace("CELLAR BUNDLE: bundle {} is marked BLOCKED OUTBOUND for cluster group {}", bundleLocation, groupName);
}
} finally {
Thread.currentThread().setContextClassLoader(originalClassLoader);
@@ -210,7 +249,6 @@ public class BundleSynchronizer extends BundleSupport implements Synchronizer {
@Override
public String getSyncPolicy(Group group) {
String groupName = group.getName();
-
try {
Configuration configuration = configurationAdmin.getConfiguration(Configurations.GROUP, null);
Dictionary<String, Object> properties = configuration.getProperties();
@@ -221,7 +259,8 @@ public class BundleSynchronizer extends BundleSupport implements Synchronizer {
} catch (IOException e) {
LOGGER.error("CELLAR BUNDLE: error while retrieving the sync policy", e);
}
+
return "disabled";
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/bundle/src/main/java/org/apache/karaf/cellar/bundle/LocalBundleListener.java
----------------------------------------------------------------------
diff --git a/bundle/src/main/java/org/apache/karaf/cellar/bundle/LocalBundleListener.java b/bundle/src/main/java/org/apache/karaf/cellar/bundle/LocalBundleListener.java
index e7489f6..9d9cfcf 100644
--- a/bundle/src/main/java/org/apache/karaf/cellar/bundle/LocalBundleListener.java
+++ b/bundle/src/main/java/org/apache/karaf/cellar/bundle/LocalBundleListener.java
@@ -55,7 +55,7 @@ public class LocalBundleListener extends BundleSupport implements SynchronousBun
}
if (!isEnabled()) {
- LOGGER.debug("CELLAR BUNDLE: local listener is disabled");
+ LOGGER.trace("CELLAR BUNDLE: local listener is disabled");
return;
}
@@ -70,12 +70,11 @@ public class LocalBundleListener extends BundleSupport implements SynchronousBun
}
if (event != null && event.getBundle() != null) {
-
Set<Group> groups = null;
try {
groups = groupManager.listLocalGroups();
} catch (Exception ex) {
- LOGGER.warn("CELLAR BUNDLE: failed to list local groups. Is Cellar uninstalling ?");
+ LOGGER.warn("Failed to list local groups. Is Cellar uninstalling ?");
}
if (groups != null && !groups.isEmpty()) {
@@ -90,16 +89,17 @@ public class LocalBundleListener extends BundleSupport implements SynchronousBun
String symbolicName = event.getBundle().getSymbolicName();
String version = event.getBundle().getHeaders().get(org.osgi.framework.Constants.BUNDLE_VERSION);
String bundleLocation = event.getBundle().getLocation();
- int type = event.getType();
+ int status = event.getBundle().getState();
+
if (isAllowed(group, Constants.CATEGORY, bundleLocation, EventType.OUTBOUND)) {
ClassLoader originalClassLoader = Thread.currentThread().getContextClassLoader();
Thread.currentThread().setContextClassLoader(getClass().getClassLoader());
try {
- // update bundle in the cluster group
+ // update bundles in the cluster group
Map<String, BundleState> clusterBundles = clusterManager.getMap(Constants.BUNDLE_MAP + Configurations.SEPARATOR + group.getName());
- if (type == BundleEvent.UNINSTALLED) {
+ if (event.getType() == BundleEvent.UNINSTALLED) {
clusterBundles.remove(symbolicName + "/" + version);
} else {
BundleState state = clusterBundles.get(symbolicName + "/" + version);
@@ -109,7 +109,8 @@ public class LocalBundleListener extends BundleSupport implements SynchronousBun
state.setId(event.getBundle().getBundleId());
state.setName(name);
state.setVersion(version);
- state.setStatus(type);
+ state.setSymbolicName(symbolicName);
+ state.setStatus(status);
state.setLocation(bundleLocation);
clusterBundles.put(symbolicName + "/" + version, state);
}
@@ -118,30 +119,30 @@ public class LocalBundleListener extends BundleSupport implements SynchronousBun
List<Feature> matchingFeatures = retrieveFeature(bundleLocation);
for (Feature feature : matchingFeatures) {
if (!isAllowed(group, "features", feature.getName(), EventType.OUTBOUND)) {
- LOGGER.debug("CELLAR BUNDLE: bundle {} is contained in feature {} marked BLOCKED OUTBOUND for cluster group {}", bundleLocation, feature.getName(), group.getName());
+ LOGGER.trace("CELLAR BUNDLE: bundle {} is contained in feature {} marked BLOCKED OUTBOUND for cluster group {}", bundleLocation, feature.getName(), group.getName());
return;
}
}
- // broadcast the cluster bundle event
- ClusterBundleEvent clusterBundleEvent = new ClusterBundleEvent(symbolicName, version, bundleLocation, type);
+ // broadcast the cluster event
+ ClusterBundleEvent clusterBundleEvent = new ClusterBundleEvent(symbolicName, version, bundleLocation, status);
clusterBundleEvent.setSourceGroup(group);
+ clusterBundleEvent.setSourceNode(clusterManager.getNode());
eventProducer.produce(clusterBundleEvent);
} catch (Exception e) {
- LOGGER.error("CELLAR BUNDLE: failed to create cluster bundle event", e);
+ LOGGER.error("CELLAR BUNDLE: failed to create bundle event", e);
} finally {
Thread.currentThread().setContextClassLoader(originalClassLoader);
}
- } else
- LOGGER.debug("CELLAR BUNDLE: bundle {} is marked BLOCKED OUTBOUND for cluster group {}", bundleLocation, group.getName());
+ } else LOGGER.trace("CELLAR BUNDLE: bundle {} is marked BLOCKED OUTBOUND for cluster group {}", bundleLocation, group.getName());
}
}
}
}
/**
- * Check if the local node bundle listener is enabled in the etc/org.apache.karaf.cellar.groups.cfg.
+ * Check if the local bundle listener is enabled in the etc/org.apache.karaf.cellar.groups.cfg.
*
* @return true if enabled, false else.
*/
@@ -160,7 +161,7 @@ public class LocalBundleListener extends BundleSupport implements SynchronousBun
}
public void init() {
- bundleContext.addBundleListener(this);
+ getBundleContext().addBundleListener(this);
}
public void destroy() {
@@ -175,4 +176,4 @@ public class LocalBundleListener extends BundleSupport implements SynchronousBun
this.eventProducer = eventProducer;
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/InstallBundleCommand.java
----------------------------------------------------------------------
diff --git a/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/InstallBundleCommand.java b/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/InstallBundleCommand.java
index f5e96c4..3ef71d9 100644
--- a/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/InstallBundleCommand.java
+++ b/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/InstallBundleCommand.java
@@ -26,7 +26,7 @@ import org.apache.karaf.cellar.core.control.SwitchStatus;
import org.apache.karaf.cellar.core.event.EventProducer;
import org.apache.karaf.cellar.core.event.EventType;
import org.apache.karaf.cellar.core.shell.CellarCommandSupport;
-import org.osgi.framework.BundleEvent;
+import org.osgi.framework.Bundle;
import java.net.URL;
import java.util.List;
@@ -103,9 +103,9 @@ public class InstallBundleCommand extends CellarCommandSupport {
state.setLocation(url);
state.setId(clusterBundles.size());
if (start) {
- state.setStatus(BundleEvent.STARTED);
+ state.setStatus(Bundle.ACTIVE);
} else {
- state.setStatus(BundleEvent.INSTALLED);
+ state.setStatus(Bundle.INSTALLED);
}
clusterBundles.put(symbolicName + "/" + version, state);
} finally {
@@ -113,7 +113,10 @@ public class InstallBundleCommand extends CellarCommandSupport {
}
// broadcast the cluster event
- ClusterBundleEvent event = new ClusterBundleEvent(symbolicName, version, url, BundleEvent.INSTALLED);
+ ClusterBundleEvent event;
+ if (start)
+ event = new ClusterBundleEvent(symbolicName, version, url, Bundle.ACTIVE);
+ else event = new ClusterBundleEvent(symbolicName, version, url, Bundle.INSTALLED);
event.setSourceGroup(group);
eventProducer.produce(event);
} else {
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/ListBundleCommand.java
----------------------------------------------------------------------
diff --git a/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/ListBundleCommand.java b/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/ListBundleCommand.java
index 6eef376..5c1d145 100644
--- a/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/ListBundleCommand.java
+++ b/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/ListBundleCommand.java
@@ -21,7 +21,7 @@ import org.apache.karaf.cellar.bundle.Constants;
import org.apache.karaf.cellar.core.Configurations;
import org.apache.karaf.cellar.core.Group;
import org.apache.karaf.cellar.core.shell.CellarCommandSupport;
-import org.osgi.framework.BundleEvent;
+import org.osgi.framework.Bundle;
import java.util.*;
@@ -64,25 +64,22 @@ public class ListBundleCommand extends CellarCommandSupport {
String status;
switch (bundle.getStatus()) {
- case BundleEvent.INSTALLED:
+ case Bundle.INSTALLED:
status = "Installed";
break;
- case BundleEvent.RESOLVED:
+ case Bundle.RESOLVED:
status = "Resolved";
break;
- case BundleEvent.STARTED:
+ case Bundle.ACTIVE:
status = "Active";
break;
- case BundleEvent.STARTING:
+ case Bundle.STARTING:
status = "Starting";
break;
- case BundleEvent.STOPPED:
- status = "Resolved";
- break;
- case BundleEvent.STOPPING:
+ case Bundle.STOPPING:
status = "Stopping";
break;
- case BundleEvent.UNINSTALLED:
+ case Bundle.UNINSTALLED:
status = "Uninstalled";
break;
default:
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/StartBundleCommand.java
----------------------------------------------------------------------
diff --git a/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/StartBundleCommand.java b/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/StartBundleCommand.java
index 3c431eb..9fbc759 100644
--- a/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/StartBundleCommand.java
+++ b/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/StartBundleCommand.java
@@ -23,7 +23,7 @@ import org.apache.karaf.cellar.core.Group;
import org.apache.karaf.cellar.core.control.SwitchStatus;
import org.apache.karaf.cellar.core.event.EventProducer;
import org.apache.karaf.cellar.core.event.EventType;
-import org.osgi.framework.BundleEvent;
+import org.osgi.framework.Bundle;
import java.util.Map;
@@ -79,7 +79,7 @@ public class StartBundleCommand extends BundleCommandSupport {
return null;
}
- state.setStatus(BundleEvent.STARTED);
+ state.setStatus(Bundle.ACTIVE);
clusterBundles.put(key, state);
} finally {
Thread.currentThread().setContextClassLoader(originalClassLoader);
@@ -87,7 +87,7 @@ public class StartBundleCommand extends BundleCommandSupport {
// broadcast the cluster event
String[] split = key.split("/");
- ClusterBundleEvent event = new ClusterBundleEvent(split[0], split[1], location, BundleEvent.STARTED);
+ ClusterBundleEvent event = new ClusterBundleEvent(split[0], split[1], location, Bundle.ACTIVE);
event.setSourceGroup(group);
eventProducer.produce(event);
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/StopBundleCommand.java
----------------------------------------------------------------------
diff --git a/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/StopBundleCommand.java b/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/StopBundleCommand.java
index 70a7c00..2cdb559 100644
--- a/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/StopBundleCommand.java
+++ b/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/StopBundleCommand.java
@@ -23,7 +23,7 @@ import org.apache.karaf.cellar.core.Group;
import org.apache.karaf.cellar.core.control.SwitchStatus;
import org.apache.karaf.cellar.core.event.EventProducer;
import org.apache.karaf.cellar.core.event.EventType;
-import org.osgi.framework.BundleEvent;
+import org.osgi.framework.Bundle;
import java.util.Map;
@@ -67,7 +67,7 @@ public class StopBundleCommand extends BundleCommandSupport {
System.err.println("Bundle " + key + " not found in cluster group " + groupName);
return null;
}
- state.setStatus(BundleEvent.STOPPED);
+ state.setStatus(Bundle.RESOLVED);
location = state.getLocation();
// check if the bundle is allowed
@@ -87,7 +87,7 @@ public class StopBundleCommand extends BundleCommandSupport {
// broadcast the cluster event
String[] split = key.split("/");
- ClusterBundleEvent event = new ClusterBundleEvent(split[0], split[1], location, BundleEvent.STOPPED);
+ ClusterBundleEvent event = new ClusterBundleEvent(split[0], split[1], location, Bundle.RESOLVED);
event.setSourceGroup(group);
eventProducer.produce(event);
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/UninstallBundleCommand.java
----------------------------------------------------------------------
diff --git a/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/UninstallBundleCommand.java b/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/UninstallBundleCommand.java
index f69b295..7cde315 100644
--- a/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/UninstallBundleCommand.java
+++ b/bundle/src/main/java/org/apache/karaf/cellar/bundle/shell/UninstallBundleCommand.java
@@ -23,7 +23,7 @@ import org.apache.karaf.cellar.core.Group;
import org.apache.karaf.cellar.core.control.SwitchStatus;
import org.apache.karaf.cellar.core.event.EventProducer;
import org.apache.karaf.cellar.core.event.EventType;
-import org.osgi.framework.BundleEvent;
+import org.osgi.framework.Bundle;
import java.util.Map;
@@ -87,7 +87,7 @@ public class UninstallBundleCommand extends BundleCommandSupport {
// broadcast the cluster event
String[] split = key.split("/");
- ClusterBundleEvent event = new ClusterBundleEvent(split[0], split[1], location, BundleEvent.UNINSTALLED);
+ ClusterBundleEvent event = new ClusterBundleEvent(split[0], split[1], location, Bundle.UNINSTALLED);
event.setSourceGroup(group);
eventProducer.produce(event);
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/bundle/src/main/resources/OSGI-INF/blueprint/blueprint.xml
----------------------------------------------------------------------
diff --git a/bundle/src/main/resources/OSGI-INF/blueprint/blueprint.xml b/bundle/src/main/resources/OSGI-INF/blueprint/blueprint.xml
index cb84006..6948b4f 100644
--- a/bundle/src/main/resources/OSGI-INF/blueprint/blueprint.xml
+++ b/bundle/src/main/resources/OSGI-INF/blueprint/blueprint.xml
@@ -32,6 +32,7 @@
<property name="groupManager" ref="groupManager"/>
<property name="clusterManager" ref="clusterManager"/>
<property name="bundleContext" ref="blueprintBundleContext"/>
+ <property name="eventProducer" ref="eventProducer"/>
<property name="featuresService" ref="featuresService"/>
</bean>
<service ref="synchronizer" interface="org.apache.karaf.cellar.core.Synchronizer">
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/config/src/main/java/org/apache/karaf/cellar/config/ConfigurationEventHandler.java
----------------------------------------------------------------------
diff --git a/config/src/main/java/org/apache/karaf/cellar/config/ConfigurationEventHandler.java b/config/src/main/java/org/apache/karaf/cellar/config/ConfigurationEventHandler.java
index 9340343..fd0c49c 100644
--- a/config/src/main/java/org/apache/karaf/cellar/config/ConfigurationEventHandler.java
+++ b/config/src/main/java/org/apache/karaf/cellar/config/ConfigurationEventHandler.java
@@ -55,6 +55,12 @@ public class ConfigurationEventHandler extends ConfigurationSupport implements E
return;
}
+ // check if it's not a "local" event
+ if (event.getSourceNode() != null && event.getSourceNode().getId().equalsIgnoreCase(clusterManager.getNode().getId())) {
+ LOGGER.trace("CELLAR CONFIG: cluster event is local (coming from local synchronizer or listener)");
+ return;
+ }
+
Group group = event.getSourceGroup();
String groupName = group.getName();
@@ -90,7 +96,8 @@ public class ConfigurationEventHandler extends ConfigurationSupport implements E
} catch (Exception ex) {
LOGGER.error("CELLAR CONFIG: failed to update local configuration", ex);
}
- } else LOGGER.debug("CELLAR CONFIG: configuration with PID {} is marked BLOCKED INBOUND for cluster group {}", pid, groupName);
+ } else
+ LOGGER.debug("CELLAR CONFIG: configuration with PID {} is marked BLOCKED INBOUND for cluster group {}", pid, groupName);
}
public void init() {
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/config/src/main/java/org/apache/karaf/cellar/config/ConfigurationSynchronizer.java
----------------------------------------------------------------------
diff --git a/config/src/main/java/org/apache/karaf/cellar/config/ConfigurationSynchronizer.java b/config/src/main/java/org/apache/karaf/cellar/config/ConfigurationSynchronizer.java
index c257961..dc5f7ae 100644
--- a/config/src/main/java/org/apache/karaf/cellar/config/ConfigurationSynchronizer.java
+++ b/config/src/main/java/org/apache/karaf/cellar/config/ConfigurationSynchronizer.java
@@ -38,6 +38,12 @@ public class ConfigurationSynchronizer extends ConfigurationSupport implements S
private static final transient Logger LOGGER = LoggerFactory.getLogger(ConfigurationSynchronizer.class);
+ private EventProducer eventProducer;
+
+ public void setEventProducer(EventProducer eventProducer) {
+ this.eventProducer = eventProducer;
+ }
+
public ConfigurationSynchronizer() {
// nothing to do
}
@@ -63,19 +69,32 @@ public class ConfigurationSynchronizer extends ConfigurationSupport implements S
@Override
public void sync(Group group) {
String policy = getSyncPolicy(group);
- if (policy != null && policy.equalsIgnoreCase("cluster")) {
- LOGGER.debug("CELLAR CONFIG: sync policy is set as 'cluster' for cluster group " + group.getName());
- if (clusterManager.listNodesByGroup(group).size() == 1 && clusterManager.listNodesByGroup(group).contains(clusterManager.getNode())) {
- LOGGER.debug("CELLAR CONFIG: node is the first and only member of the group, pushing state");
- push(group);
- } else {
- LOGGER.debug("CELLAR CONFIG: pulling state");
- pull(group);
- }
+ if (policy == null) {
+ LOGGER.warn("CELLAR CONFIG: sync policy is not defined for cluster group {}", group.getName());
}
- if (policy != null && policy.equalsIgnoreCase("node")) {
- LOGGER.debug("CELLAR CONFIG: sync policy is set as 'node' for cluster group " + group.getName());
+ if (policy.equalsIgnoreCase("cluster")) {
+ LOGGER.debug("CELLAR CONFIG: sync policy set as 'cluster' for cluster group {}", group.getName());
+ LOGGER.debug("CELLAR CONFIG: updating node from the cluster (pull first)");
+ pull(group);
+ LOGGER.debug("CELLAR CONFIG: updating cluster from the local node (push after)");
+ push(group);
+ } else if (policy.equalsIgnoreCase("node")) {
+ LOGGER.debug("CELLAR CONFIG: sync policy set as 'node' for cluster group {}", group.getName());
+ LOGGER.debug("CELLAR CONFIG: updating cluster from the local node (push first)");
push(group);
+ LOGGER.debug("CELLAR CONFIG: updating node from the cluster (pull after)");
+ pull(group);
+ } else if (policy.equalsIgnoreCase("clusterOnly")) {
+ LOGGER.debug("CELLAR CONFIG: sync policy set as 'clusterOnly' for cluster group " + group.getName());
+ LOGGER.debug("CELLAR CONFIG: updating node from the cluster (pull only)");
+ pull(group);
+ } else if (policy.equalsIgnoreCase("nodeOnly")) {
+ LOGGER.debug("CELLAR CONFIG: sync policy set as 'nodeOnly' for cluster group " + group.getName());
+ LOGGER.debug("CELLAR CONFIG: updating cluster from the local node (push only)");
+ push(group);
+ } else {
+ LOGGER.debug("CELLAR CONFIG: sync policy set as 'disabled' for cluster group " + group.getName());
+ LOGGER.debug("CELLAR CONFIG: no sync");
}
}
@@ -107,6 +126,7 @@ public class ConfigurationSynchronizer extends ConfigurationSupport implements S
localDictionary = filter(localDictionary);
if (!equals(clusterDictionary, localDictionary)) {
+ LOGGER.debug("CELLAR CONFIG: updating configration {} on node", pid);
localConfiguration.update((Dictionary) clusterDictionary);
persistConfiguration(configurationAdmin, pid, clusterDictionary);
}
@@ -129,6 +149,11 @@ public class ConfigurationSynchronizer extends ConfigurationSupport implements S
*/
public void push(Group group) {
+ if (eventProducer.getSwitch().getStatus().equals(SwitchStatus.OFF)) {
+ LOGGER.warn("CELLAR CONFIG: cluster event producer is OFF");
+ return;
+ }
+
if (group != null) {
String groupName = group.getName();
Map<String, Properties> clusterConfigurations = clusterManager.getMap(Constants.CONFIGURATION_MAP + Configurations.SEPARATOR + groupName);
@@ -144,8 +169,28 @@ public class ConfigurationSynchronizer extends ConfigurationSupport implements S
if (isAllowed(group, Constants.CATEGORY, pid, EventType.OUTBOUND)) {
Dictionary localDictionary = localConfiguration.getProperties();
localDictionary = filter(localDictionary);
- // update the configurations in the cluster group
- clusterConfigurations.put(pid, dictionaryToProperties(localDictionary));
+ if (!clusterConfigurations.containsKey(pid)) {
+ LOGGER.debug("CELLAR CONFIG: creating configuration pid {} on the cluster", pid);
+ // update cluster configurations
+ clusterConfigurations.put(pid, dictionaryToProperties(localDictionary));
+ // send cluster event
+ ClusterConfigurationEvent event = new ClusterConfigurationEvent(pid);
+ event.setSourceGroup(group);
+ event.setSourceNode(clusterManager.getNode());
+ eventProducer.produce(event);
+ } else {
+ Dictionary clusterDictionary = clusterConfigurations.get(pid);
+ if (!equals(clusterDictionary, localDictionary)) {
+ LOGGER.debug("CELLAR CONFIG: updating configuration pid {} on the cluster", pid);
+ // update cluster configurations
+ clusterConfigurations.put(pid, dictionaryToProperties(localDictionary));
+ // send cluster event
+ ClusterConfigurationEvent event = new ClusterConfigurationEvent(pid);
+ event.setSourceGroup(group);
+ event.setSourceNode(clusterManager.getNode());
+ eventProducer.produce(event);
+ }
+ }
} else
LOGGER.debug("CELLAR CONFIG: configuration with PID {} is marked BLOCKED OUTBOUND for cluster group {}", pid, groupName);
}
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/config/src/main/java/org/apache/karaf/cellar/config/LocalConfigurationListener.java
----------------------------------------------------------------------
diff --git a/config/src/main/java/org/apache/karaf/cellar/config/LocalConfigurationListener.java b/config/src/main/java/org/apache/karaf/cellar/config/LocalConfigurationListener.java
index a74cf92..4385bf9 100644
--- a/config/src/main/java/org/apache/karaf/cellar/config/LocalConfigurationListener.java
+++ b/config/src/main/java/org/apache/karaf/cellar/config/LocalConfigurationListener.java
@@ -48,7 +48,7 @@ public class LocalConfigurationListener extends ConfigurationSupport implements
public void configurationEvent(ConfigurationEvent event) {
if (!isEnabled()) {
- LOGGER.debug("CELLAR CONFIG: local listener is disabled");
+ LOGGER.trace("CELLAR CONFIG: local listener is disabled");
return;
}
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/config/src/main/resources/OSGI-INF/blueprint/blueprint.xml
----------------------------------------------------------------------
diff --git a/config/src/main/resources/OSGI-INF/blueprint/blueprint.xml b/config/src/main/resources/OSGI-INF/blueprint/blueprint.xml
index 337f0d9..d83f7e9 100644
--- a/config/src/main/resources/OSGI-INF/blueprint/blueprint.xml
+++ b/config/src/main/resources/OSGI-INF/blueprint/blueprint.xml
@@ -33,6 +33,7 @@
<property name="configurationAdmin" ref="configurationAdmin"/>
<property name="groupManager" ref="groupManager"/>
<property name="clusterManager" ref="clusterManager"/>
+ <property name="eventProducer" ref="eventProducer"/>
<property name="storage" value="${storage}"/>
</bean>
<service ref="synchronizer" interface="org.apache.karaf.cellar.core.Synchronizer">
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/features/src/main/java/org/apache/karaf/cellar/features/FeaturesEventHandler.java
----------------------------------------------------------------------
diff --git a/features/src/main/java/org/apache/karaf/cellar/features/FeaturesEventHandler.java b/features/src/main/java/org/apache/karaf/cellar/features/FeaturesEventHandler.java
index dd48a3d..f212dcf 100644
--- a/features/src/main/java/org/apache/karaf/cellar/features/FeaturesEventHandler.java
+++ b/features/src/main/java/org/apache/karaf/cellar/features/FeaturesEventHandler.java
@@ -68,6 +68,12 @@ public class FeaturesEventHandler extends FeaturesSupport implements EventHandle
return;
}
+ // check if it's not a "local" event
+ if (event.getSourceNode() != null && event.getSourceNode().getId().equalsIgnoreCase(clusterManager.getNode().getId())) {
+ LOGGER.trace("CELLAR FEATURE: cluster event is local (coming from local synchronizer or listener)");
+ return;
+ }
+
String name = event.getName();
String version = event.getVersion();
if (isAllowed(event.getSourceGroup(), Constants.CATEGORY, name, EventType.INBOUND) || event.getForce()) {
@@ -103,7 +109,8 @@ public class FeaturesEventHandler extends FeaturesSupport implements EventHandle
} catch (Exception e) {
LOGGER.error("CELLAR FEATURES_MAP: failed to handle event", e);
}
- } else LOGGER.debug("CELLAR FEATURES_MAP: feature {} is marked BLOCKED INBOUND for cluster group {}", name, event.getSourceGroup().getName());
+ } else
+ LOGGER.debug("CELLAR FEATURES_MAP: feature {} is marked BLOCKED INBOUND for cluster group {}", name, event.getSourceGroup().getName());
}
@Override
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/features/src/main/java/org/apache/karaf/cellar/features/FeaturesSynchronizer.java
----------------------------------------------------------------------
diff --git a/features/src/main/java/org/apache/karaf/cellar/features/FeaturesSynchronizer.java b/features/src/main/java/org/apache/karaf/cellar/features/FeaturesSynchronizer.java
index 073dc19..a605184 100644
--- a/features/src/main/java/org/apache/karaf/cellar/features/FeaturesSynchronizer.java
+++ b/features/src/main/java/org/apache/karaf/cellar/features/FeaturesSynchronizer.java
@@ -16,11 +16,13 @@ package org.apache.karaf.cellar.features;
import org.apache.karaf.cellar.core.Configurations;
import org.apache.karaf.cellar.core.Group;
import org.apache.karaf.cellar.core.Synchronizer;
+import org.apache.karaf.cellar.core.control.SwitchStatus;
import org.apache.karaf.cellar.core.event.EventProducer;
import org.apache.karaf.cellar.core.event.EventType;
import org.apache.karaf.features.Feature;
import org.apache.karaf.features.FeatureEvent;
import org.apache.karaf.features.Repository;
+import org.apache.karaf.features.RepositoryEvent;
import org.osgi.service.cm.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,6 +42,12 @@ public class FeaturesSynchronizer extends FeaturesSupport implements Synchronize
private static final transient Logger LOGGER = LoggerFactory.getLogger(FeaturesSynchronizer.class);
+ private EventProducer eventProducer;
+
+ public void setEventProducer(EventProducer eventProducer) {
+ this.eventProducer = eventProducer;
+ }
+
@Override
public void init() {
Set<Group> groups = groupManager.listLocalGroups();
@@ -63,19 +71,32 @@ public class FeaturesSynchronizer extends FeaturesSupport implements Synchronize
@Override
public void sync(Group group) {
String policy = getSyncPolicy(group);
- if (policy != null && policy.equalsIgnoreCase("cluster")) {
- LOGGER.debug("CELLAR FEATURE: sync policy is set as 'cluster' for cluster group " + group.getName());
- if (clusterManager.listNodesByGroup(group).size() == 1 && clusterManager.listNodesByGroup(group).contains(clusterManager.getNode())) {
- LOGGER.debug("CELLAR FEATURE: node is the first and only member of the group, pushing state");
- push(group);
- } else {
- LOGGER.debug("CELLAR FEATURE: pulling state");
- pull(group);
- }
+ if (policy == null) {
+ LOGGER.warn("CELLAR FEATURE: sync policy is not defined for cluster group {}", group.getName());
}
- if (policy != null && policy.equalsIgnoreCase("node")) {
- LOGGER.debug("CELLAR FEATURE: sync policy is set as 'node' for cluster group " + group.getName());
+ if (policy.equalsIgnoreCase("cluster")) {
+ LOGGER.debug("CELLAR FEATURE: sync policy set as 'cluster' for cluster group {}", group.getName());
+ LOGGER.debug("CELLAR FEATURE: updating node from the cluster (pull first)");
+ pull(group);
+ LOGGER.debug("CELLAR FEATURE: updating cluster from the local node (push after)");
+ push(group);
+ } else if (policy.equalsIgnoreCase("node")) {
+ LOGGER.debug("CELLAR FEATURE: sync policy set as 'node' for cluster group {}", group.getName());
+ LOGGER.debug("CELLAR FEATURE: updating cluster from the local node (push first)");
+ push(group);
+ LOGGER.debug("CELLAR FEATURE: updating node from the cluster (pull after)");
+ pull(group);
+ } else if (policy.equalsIgnoreCase("clusterOnly")) {
+ LOGGER.debug("CELLAR FEATURE: sync policy set as 'clusterOnly' for cluster group " + group.getName());
+ LOGGER.debug("CELLAR FEATURE: updating node from the cluster (pull only)");
+ pull(group);
+ } else if (policy.equalsIgnoreCase("nodeOnly")) {
+ LOGGER.debug("CELLAR FEATURE: sync policy set as 'nodeOnly' for cluster group " + group.getName());
+ LOGGER.debug("CELLAR FEATURE: updating cluster from the local node (push only)");
push(group);
+ } else {
+ LOGGER.debug("CELLAR FEATURE: sync policy set as 'disabled' for cluster group " + group.getName());
+ LOGGER.debug("CELLAR FEATURE: no sync");
}
}
@@ -154,52 +175,98 @@ public class FeaturesSynchronizer extends FeaturesSupport implements Synchronize
*/
@Override
public void push(Group group) {
+
+ if (eventProducer.getSwitch().getStatus().equals(SwitchStatus.OFF)) {
+ LOGGER.warn("CELLAR FEATURE: cluster event producer is OFF");
+ return;
+ }
+
if (group != null) {
String groupName = group.getName();
LOGGER.debug("CELLAR FEATURES_MAP: pushing features repositories and features in cluster group {}.", groupName);
- List<String> clusterRepositories = clusterManager.getList(Constants.REPOSITORIES_LIST + Configurations.SEPARATOR + groupName);
- Map<String, FeatureState> clusterFeatures = clusterManager.getMap(Constants.FEATURES_MAP + Configurations.SEPARATOR + groupName);
+ List<String> clusterRepositories = clusterManager.getList(Constants.REPOSITORIES_LIST + Configurations.SEPARATOR + groupName);
+ Map<String, FeatureState> clusterFeatures = clusterManager.getMap(Constants.FEATURES_MAP + Configurations.SEPARATOR + groupName);
- Repository[] repositoryList = new Repository[0];
- Feature[] featuresList = new Feature[0];
+ Repository[] repositoryList = new Repository[0];
+ Feature[] featuresList = new Feature[0];
- try {
- repositoryList = featuresService.listRepositories();
- featuresList = featuresService.listFeatures();
- } catch (Exception e) {
- LOGGER.warn("CELLAR FEATURES: unable to list features", e);
- }
+ try {
+ repositoryList = featuresService.listRepositories();
+ featuresList = featuresService.listFeatures();
+ } catch (Exception e) {
+ LOGGER.warn("CELLAR FEATURES: unable to list features", e);
+ }
- // push the local features repositories to the cluster group
- if (repositoryList != null && repositoryList.length > 0) {
- for (Repository repository : repositoryList) {
- if (!clusterRepositories.contains(repository.getURI().toString())) {
- clusterRepositories.add(repository.getURI().toString());
- LOGGER.debug("CELLAR FEATURES: pushing repository {} in cluster group {}", repository.getName(), groupName);
- } else {
- LOGGER.debug("CELLAR FEATURES: repository {} is already in cluster group {}", repository.getName(), groupName);
- }
+ // push the local features repositories to the cluster group
+ if (repositoryList != null && repositoryList.length > 0) {
+ for (Repository repository : repositoryList) {
+ if (!clusterRepositories.contains(repository.getURI().toString())) {
+ LOGGER.debug("CELLAR FEATURES: pushing repository {} in cluster group {}", repository.getName(), groupName);
+ // updating cluster state
+ clusterRepositories.add(repository.getURI().toString());
+ // sending cluster event
+ ClusterRepositoryEvent event = new ClusterRepositoryEvent(repository.getURI().toString(), RepositoryEvent.EventType.RepositoryAdded);
+ event.setSourceGroup(group);
+ event.setSourceNode(clusterManager.getNode());
+ eventProducer.produce(event);
+ } else {
+ LOGGER.debug("CELLAR FEATURES: repository {} is already in cluster group {}", repository.getName(), groupName);
}
}
+ }
+
+ if (featuresList != null && featuresList.length > 0) {
+ for (Feature feature : featuresList) {
+ if (isAllowed(group, Constants.CATEGORY, feature.getName(), EventType.OUTBOUND)) {
+ boolean installed = featuresService.isInstalled(feature);
+ String key = feature.getName() + "/" + feature.getVersion();
+ FeatureState clusterFeature = clusterFeatures.get(key);
+ if (clusterFeature == null) {
+ LOGGER.debug("CELLAR FEATURE: adding feature {} to cluster group {}", key, groupName);
+ // updating cluster state
+ clusterFeature = new FeatureState();
+ clusterFeature.setName(feature.getName());
+ clusterFeature.setVersion(feature.getVersion());
+ clusterFeature.setInstalled(installed);
+ clusterFeatures.put(key, clusterFeature);
+ // sending cluster event
+ ClusterFeaturesEvent event;
+ if (installed) {
+ event = new ClusterFeaturesEvent(feature.getName(), feature.getVersion(), FeatureEvent.EventType.FeatureInstalled);
+ } else {
+ event = new ClusterFeaturesEvent(feature.getName(), feature.getVersion(), FeatureEvent.EventType.FeatureUninstalled);
+ }
+ event.setSourceGroup(group);
+ event.setSourceNode(clusterManager.getNode());
+ eventProducer.produce(event);
- if (featuresList != null && featuresList.length > 0) {
- for (Feature feature : featuresList) {
- if (isAllowed(group, Constants.CATEGORY, feature.getName(), EventType.OUTBOUND)) {
- LOGGER.debug("CELLAR FEATURES: pushing feature {} in cluster group {}", feature.getName(), group.getName());
- // push the local features status to the cluster group
- FeatureState clusterFeatureState = new FeatureState();
- clusterFeatureState.setName(feature.getName());
- clusterFeatureState.setVersion(feature.getVersion());
- clusterFeatureState.setInstalled(featuresService.isInstalled(feature));
- clusterFeatures.put(feature.getName() + "/" + feature.getVersion(), clusterFeatureState);
} else {
- LOGGER.debug("CELLAR FEATURES: feature {} is marked BLOCKED OUTBOUND for cluster group {}", feature.getName(), group.getName());
+ if (clusterFeature.getInstalled() != installed) {
+ // updating cluster state
+ clusterFeature.setInstalled(installed);
+ clusterFeatures.put(key, clusterFeature);
+ // sending cluster event
+ ClusterFeaturesEvent event;
+ if (installed) {
+ event = new ClusterFeaturesEvent(feature.getName(), feature.getVersion(), FeatureEvent.EventType.FeatureInstalled);
+ } else {
+ event = new ClusterFeaturesEvent(feature.getName(), feature.getVersion(), FeatureEvent.EventType.FeatureUninstalled);
+ }
+ event.setSourceGroup(group);
+ event.setSourceNode(clusterManager.getNode());
+ eventProducer.produce(event);
+ } else {
+ LOGGER.debug("CELLAR FEATURE: feature {} already sync on the cluster group {}", key, groupName);
+ }
}
+ } else {
+ LOGGER.debug("CELLAR FEATURES: feature {} is marked BLOCKED OUTBOUND for cluster group {}", feature.getName(), group.getName());
}
}
}
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/features/src/main/java/org/apache/karaf/cellar/features/LocalFeaturesListener.java
----------------------------------------------------------------------
diff --git a/features/src/main/java/org/apache/karaf/cellar/features/LocalFeaturesListener.java b/features/src/main/java/org/apache/karaf/cellar/features/LocalFeaturesListener.java
index a9dd88a..25efd52 100644
--- a/features/src/main/java/org/apache/karaf/cellar/features/LocalFeaturesListener.java
+++ b/features/src/main/java/org/apache/karaf/cellar/features/LocalFeaturesListener.java
@@ -58,13 +58,13 @@ public class LocalFeaturesListener extends FeaturesSupport implements org.apache
public void featureEvent(FeatureEvent event) {
if (!isEnabled()) {
- LOGGER.debug("CELLAR FEATURES_MAP: local listener is disabled");
+ LOGGER.trace("CELLAR FEATURES: local listener is disabled");
return;
}
// check if the producer is ON
if (eventProducer.getSwitch().getStatus().equals(SwitchStatus.OFF)) {
- LOGGER.debug("CELLAR FEATURES_MAP: cluster event producer is OFF");
+ LOGGER.debug("CELLAR FEATURES: cluster event producer is OFF");
return;
}
@@ -96,9 +96,10 @@ public class LocalFeaturesListener extends FeaturesSupport implements org.apache
// broadcast the cluster event
ClusterFeaturesEvent featureEvent = new ClusterFeaturesEvent(name, version, type);
featureEvent.setSourceGroup(group);
+ featureEvent.setSourceNode(clusterManager.getNode());
eventProducer.produce(featureEvent);
} else
- LOGGER.debug("CELLAR FEATURES_MAP: feature {} is marked BLOCKED OUTBOUND for cluster group {}", name, group.getName());
+ LOGGER.debug("CELLAR FEATURES: feature {} is marked BLOCKED OUTBOUND for cluster group {}", name, group.getName());
}
}
}
@@ -113,13 +114,13 @@ public class LocalFeaturesListener extends FeaturesSupport implements org.apache
public void repositoryEvent(RepositoryEvent event) {
if (!isEnabled()) {
- LOGGER.debug("CELLAR FEATURES_MAP: local listener is disabled");
+ LOGGER.trace("CELLAR FEATURES: local listener is disabled");
return;
}
// check if the producer is ON
if (eventProducer.getSwitch().getStatus().equals(SwitchStatus.OFF)) {
- LOGGER.debug("CELLAR FEATURES_MAP: cluster event producer is OFF");
+ LOGGER.debug("CELLAR FEATURES: cluster event producer is OFF");
return;
}
@@ -133,6 +134,7 @@ public class LocalFeaturesListener extends FeaturesSupport implements org.apache
for (Group group : groups) {
ClusterRepositoryEvent repositoryEvent = new ClusterRepositoryEvent(event.getRepository().getURI().toString(), event.getType());
repositoryEvent.setSourceGroup(group);
+ repositoryEvent.setSourceNode(clusterManager.getNode());
RepositoryEvent.EventType type = event.getType();
List<String> clusterRepositories = clusterManager.getList(Constants.REPOSITORIES_LIST + Configurations.SEPARATOR + group.getName());
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/features/src/main/resources/OSGI-INF/blueprint/blueprint.xml
----------------------------------------------------------------------
diff --git a/features/src/main/resources/OSGI-INF/blueprint/blueprint.xml b/features/src/main/resources/OSGI-INF/blueprint/blueprint.xml
index bab0aca..76c9b58 100644
--- a/features/src/main/resources/OSGI-INF/blueprint/blueprint.xml
+++ b/features/src/main/resources/OSGI-INF/blueprint/blueprint.xml
@@ -31,6 +31,7 @@
<property name="clusterManager" ref="clusterManager"/>
<property name="groupManager" ref="groupManager"/>
<property name="configurationAdmin" ref="configurationAdmin"/>
+ <property name="eventProducer" ref="eventProducer"/>
<property name="featuresService" ref="featuresService"/>
</bean>
<service ref="synchronizer" interface="org.apache.karaf.cellar.core.Synchronizer">
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/hazelcast/src/main/java/org/apache/karaf/cellar/hazelcast/QueueProducer.java
----------------------------------------------------------------------
diff --git a/hazelcast/src/main/java/org/apache/karaf/cellar/hazelcast/QueueProducer.java b/hazelcast/src/main/java/org/apache/karaf/cellar/hazelcast/QueueProducer.java
index ad332c8..bb4a08f 100644
--- a/hazelcast/src/main/java/org/apache/karaf/cellar/hazelcast/QueueProducer.java
+++ b/hazelcast/src/main/java/org/apache/karaf/cellar/hazelcast/QueueProducer.java
@@ -31,6 +31,7 @@ import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IQueue;
import org.apache.karaf.cellar.core.Configurations;
import org.apache.karaf.cellar.core.Node;
+import org.apache.karaf.cellar.core.command.Command;
import org.apache.karaf.cellar.core.command.Result;
import org.apache.karaf.cellar.core.control.BasicSwitch;
import org.apache.karaf.cellar.core.control.Switch;
@@ -76,7 +77,8 @@ public class QueueProducer<E extends Event> implements EventProducer<E> {
@Override
public void produce(E event) {
if (this.getSwitch().getStatus().equals(SwitchStatus.ON) || event.getForce() || event instanceof Result) {
- event.setSourceNode(node);
+ if (event instanceof Result || event instanceof Command)
+ event.setSourceNode(node);
try {
queue.put(event);
} catch (InterruptedException e) {
http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/4a2996b8/hazelcast/src/main/java/org/apache/karaf/cellar/hazelcast/TopicProducer.java
----------------------------------------------------------------------
diff --git a/hazelcast/src/main/java/org/apache/karaf/cellar/hazelcast/TopicProducer.java b/hazelcast/src/main/java/org/apache/karaf/cellar/hazelcast/TopicProducer.java
index bd8d551..d18cdbe 100644
--- a/hazelcast/src/main/java/org/apache/karaf/cellar/hazelcast/TopicProducer.java
+++ b/hazelcast/src/main/java/org/apache/karaf/cellar/hazelcast/TopicProducer.java
@@ -17,6 +17,7 @@ import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.ITopic;
import org.apache.karaf.cellar.core.Configurations;
import org.apache.karaf.cellar.core.Node;
+import org.apache.karaf.cellar.core.command.Command;
import org.apache.karaf.cellar.core.command.Result;
import org.apache.karaf.cellar.core.control.BasicSwitch;
import org.apache.karaf.cellar.core.control.Switch;
@@ -62,7 +63,8 @@ public class TopicProducer<E extends Event> implements EventProducer<E> {
@Override
public void produce(E event) {
if (this.getSwitch().getStatus().equals(SwitchStatus.ON) || event.getForce() || event instanceof Result) {
- event.setSourceNode(node);
+ if (event instanceof Result || event instanceof Command)
+ event.setSourceNode(node);
topic.publish(event);
} else {
if (eventSwitch.getStatus().equals(SwitchStatus.OFF)) {