aboutsummaryrefslogtreecommitdiffstats
path: root/feature-pooling-dmaap/src/main/java
diff options
context:
space:
mode:
Diffstat (limited to 'feature-pooling-dmaap/src/main/java')
-rw-r--r--feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/DmaapManager.java27
-rw-r--r--feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingFeature.java59
-rw-r--r--feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingManagerImpl.java4
-rw-r--r--feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/message/Forward.java2
4 files changed, 46 insertions, 46 deletions
diff --git a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/DmaapManager.java b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/DmaapManager.java
index eb41f803..8780eefc 100644
--- a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/DmaapManager.java
+++ b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/DmaapManager.java
@@ -21,11 +21,13 @@
package org.onap.policy.drools.pooling;
import java.util.List;
-import org.onap.policy.drools.event.comm.FilterableTopicSource;
-import org.onap.policy.drools.event.comm.TopicEndpoint;
-import org.onap.policy.drools.event.comm.TopicListener;
-import org.onap.policy.drools.event.comm.TopicSink;
-import org.onap.policy.drools.event.comm.TopicSource;
+
+import org.onap.policy.common.endpoints.event.comm.FilterableTopicSource;
+import org.onap.policy.common.endpoints.event.comm.TopicEndpoint;
+import org.onap.policy.common.endpoints.event.comm.TopicListener;
+import org.onap.policy.common.endpoints.event.comm.TopicSink;
+import org.onap.policy.common.endpoints.event.comm.TopicSource;
+import org.onap.policy.common.endpoints.event.comm.impl.ProxyTopicEndpointManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -97,8 +99,7 @@ public class DmaapManager {
}
/**
- * Used by junit tests to set the factory used to create various objects used by this
- * class.
+ * Used by junit tests to set the factory used to create various objects used by this class.
*
* @param factory the new factory
*/
@@ -162,8 +163,8 @@ public class DmaapManager {
/**
* Stops the publisher.
*
- * @param waitMs time, in milliseconds, to wait for the sink to transmit any queued
- * messages and close
+ * @param waitMs time, in milliseconds, to wait for the sink to transmit any queued messages and
+ * close
*/
public void stopPublisher(long waitMs) {
if (!publishing) {
@@ -171,8 +172,8 @@ public class DmaapManager {
}
/*
- * Give the sink a chance to transmit messages in the queue. It would be better if
- * "waitMs" could be passed to sink.stop(), but that isn't an option at this time.
+ * Give the sink a chance to transmit messages in the queue. It would be better if "waitMs"
+ * could be passed to sink.stop(), but that isn't an option at this time.
*/
try {
Thread.sleep(waitMs);
@@ -262,14 +263,14 @@ public class DmaapManager {
* @return the topic sources
*/
public List<TopicSource> getTopicSources() {
- return TopicEndpoint.manager.getTopicSources();
+ return ProxyTopicEndpointManager.getInstance().getTopicSources();
}
/**
* @return the topic sinks
*/
public List<TopicSink> getTopicSinks() {
- return TopicEndpoint.manager.getTopicSinks();
+ return ProxyTopicEndpointManager.getInstance().getTopicSinks();
}
}
diff --git a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingFeature.java b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingFeature.java
index bf35bcf5..815dc548 100644
--- a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingFeature.java
+++ b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingFeature.java
@@ -25,13 +25,14 @@ import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
+
+import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
+import org.onap.policy.common.endpoints.event.comm.TopicSink;
+import org.onap.policy.common.endpoints.event.comm.TopicSource;
+import org.onap.policy.common.endpoints.event.comm.impl.ProxyTopicEndpointManager;
import org.onap.policy.common.utils.properties.SpecProperties;
import org.onap.policy.common.utils.properties.exception.PropertyException;
import org.onap.policy.drools.controller.DroolsController;
-import org.onap.policy.drools.event.comm.TopicEndpoint;
-import org.onap.policy.drools.event.comm.TopicSink;
-import org.onap.policy.drools.event.comm.TopicSource;
-import org.onap.policy.drools.event.comm.Topic.CommInfrastructure;
import org.onap.policy.drools.features.DroolsControllerFeatureAPI;
import org.onap.policy.drools.features.PolicyControllerFeatureAPI;
import org.onap.policy.drools.features.PolicyEngineFeatureAPI;
@@ -44,11 +45,10 @@ import org.slf4j.LoggerFactory;
/**
* Controller/session pooling. Multiple hosts may be launched, all servicing the same
- * controllers/sessions. When this feature is enabled, the requests are divided across the
- * different hosts, instead of all running on a single, active host.
- * <p>
- * With each controller, there is an associated DMaaP topic that is used for internal
- * communication between the different hosts serving the controller.
+ * controllers/sessions. When this feature is enabled, the requests are divided across the different
+ * hosts, instead of all running on a single, active host. <p> With each controller, there is an
+ * associated DMaaP topic that is used for internal communication between the different hosts
+ * serving the controller.
*/
public class PoolingFeature implements PolicyEngineFeatureAPI, PolicyControllerFeatureAPI, DroolsControllerFeatureAPI {
@@ -80,9 +80,9 @@ public class PoolingFeature implements PolicyEngineFeatureAPI, PolicyControllerF
private final CountDownLatch activeLatch = new CountDownLatch(1);
/**
- * Arguments passed to beforeOffer(), which are saved for when the beforeInsert() is
- * called later. As multiple threads can be active within the methods at the same
- * time, we must keep this in thread local storage.
+ * Arguments passed to beforeOffer(), which are saved for when the beforeInsert() is called
+ * later. As multiple threads can be active within the methods at the same time, we must keep
+ * this in thread local storage.
*/
private ThreadLocal<OfferArgs> offerArgs = new ThreadLocal<>();
@@ -128,13 +128,13 @@ public class PoolingFeature implements PolicyEngineFeatureAPI, PolicyControllerF
public boolean beforeStart(PolicyEngine engine) {
logger.info("initializing " + PoolingProperties.FEATURE_NAME);
featProps = factory.getProperties(PoolingProperties.FEATURE_NAME);
-
+
// remove any generic pooling topic - always use controller-specific property
featProps.remove(PoolingProperties.POOLING_TOPIC);
-
+
factory.initTopicSources(featProps);
factory.initTopicSinks(featProps);
-
+
return false;
}
@@ -154,7 +154,7 @@ public class PoolingFeature implements PolicyEngineFeatureAPI, PolicyControllerF
String name = controller.getName();
SpecProperties specProps = new SpecProperties(PoolingProperties.PREFIX, name, featProps);
-
+
if (FeatureEnabledChecker.isFeatureEnabled(specProps, PoolingProperties.FEATURE_ENABLED)) {
try {
// get & validate the properties
@@ -239,8 +239,8 @@ public class PoolingFeature implements PolicyEngineFeatureAPI, PolicyControllerF
@Override
public boolean beforeOffer(PolicyController controller, CommInfrastructure protocol, String topic2, String event) {
/*
- * As this is invoked a lot, we'll directly call the manager's method instead of
- * using the functional interface via doManager().
+ * As this is invoked a lot, we'll directly call the manager's method instead of using the
+ * functional interface via doManager().
*/
PoolingManagerImpl mgr = ctlr2pool.get(controller.getName());
if (mgr == null) {
@@ -270,20 +270,20 @@ public class PoolingFeature implements PolicyEngineFeatureAPI, PolicyControllerF
} catch (IllegalArgumentException | IllegalStateException e) {
logger.warn("cannot get controller for {} {}", droolsController.getGroupId(),
- droolsController.getArtifactId(), e);
+ droolsController.getArtifactId(), e);
return false;
}
if (controller == null) {
logger.warn("cannot determine controller for {} {}", droolsController.getGroupId(),
- droolsController.getArtifactId());
+ droolsController.getArtifactId());
return false;
}
/*
- * As this is invoked a lot, we'll directly call the manager's method instead of
- * using the functional interface via doManager().
+ * As this is invoked a lot, we'll directly call the manager's method instead of using the
+ * functional interface via doManager().
*/
PoolingManagerImpl mgr = ctlr2pool.get(controller.getName());
if (mgr == null) {
@@ -295,7 +295,7 @@ public class PoolingFeature implements PolicyEngineFeatureAPI, PolicyControllerF
@Override
public boolean afterOffer(PolicyController controller, CommInfrastructure protocol, String topic, String event,
- boolean success) {
+ boolean success) {
// clear any stored arguments
offerArgs.set(null);
@@ -304,8 +304,8 @@ public class PoolingFeature implements PolicyEngineFeatureAPI, PolicyControllerF
}
/**
- * Executes a function using the manager associated with the controller. Catches any
- * exceptions from the function and re-throws it as a runtime exception.
+ * Executes a function using the manager associated with the controller. Catches any exceptions
+ * from the function and re-throws it as a runtime exception.
*
* @param controller
* @param func function to be executed
@@ -349,8 +349,7 @@ public class PoolingFeature implements PolicyEngineFeatureAPI, PolicyControllerF
/**
*
* @param mgr
- * @return {@code true} if the request was handled by the manager, {@code false}
- * otherwise
+ * @return {@code true} if the request was handled by the manager, {@code false} otherwise
* @throws PoolingFeatureException
*/
public boolean apply(PoolingManagerImpl mgr) throws PoolingFeatureException;
@@ -412,7 +411,7 @@ public class PoolingFeature implements PolicyEngineFeatureAPI, PolicyControllerF
* @return a new pooling manager
*/
public PoolingManagerImpl makeManager(String host, PolicyController controller, PoolingProperties props,
- CountDownLatch activeLatch) {
+ CountDownLatch activeLatch) {
return new PoolingManagerImpl(host, controller, props, activeLatch);
}
@@ -433,7 +432,7 @@ public class PoolingFeature implements PolicyEngineFeatureAPI, PolicyControllerF
* @return the topic sources
*/
public List<TopicSource> initTopicSources(Properties props) {
- return TopicEndpoint.manager.addTopicSources(props);
+ return ProxyTopicEndpointManager.getInstance().addTopicSources(props);
}
/**
@@ -443,7 +442,7 @@ public class PoolingFeature implements PolicyEngineFeatureAPI, PolicyControllerF
* @return the topic sinks
*/
public List<TopicSink> initTopicSinks(Properties props) {
- return TopicEndpoint.manager.addTopicSinks(props);
+ return ProxyTopicEndpointManager.getInstance().addTopicSinks(props);
}
}
}
diff --git a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingManagerImpl.java b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingManagerImpl.java
index 68dfee14..02ba4ec9 100644
--- a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingManagerImpl.java
+++ b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingManagerImpl.java
@@ -31,8 +31,8 @@ import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.onap.policy.common.utils.properties.SpecProperties;
import org.onap.policy.drools.controller.DroolsController;
-import org.onap.policy.drools.event.comm.Topic.CommInfrastructure;
-import org.onap.policy.drools.event.comm.TopicListener;
+import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
+import org.onap.policy.common.endpoints.event.comm.TopicListener;
import org.onap.policy.drools.pooling.extractor.ClassExtractors;
import org.onap.policy.drools.pooling.message.BucketAssignments;
import org.onap.policy.drools.pooling.message.Forward;
diff --git a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/message/Forward.java b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/message/Forward.java
index 6122d361..fb3d4eb2 100644
--- a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/message/Forward.java
+++ b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/message/Forward.java
@@ -20,7 +20,7 @@
package org.onap.policy.drools.pooling.message;
-import org.onap.policy.drools.event.comm.Topic.CommInfrastructure;
+import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
import org.onap.policy.drools.pooling.PoolingFeatureException;
import com.fasterxml.jackson.annotation.JsonIgnore;