* ONAP
* ================================================================================
* Copyright (C) 2018-2019 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2019-2020 Nordix Foundation
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.lang3.StringUtils;
-import org.onap.policy.drools.utils.Pair;
+import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
* Extractors for each object class. Properties define how the data is to be
* extracted for a given class, where the properties are similar to the
* following:
- *
+ *
* <pre>
* <code><a.prefix>.<class.name> = ${event.reqid}</code>
* </pre>
- *
+ *
* <p>For any given field name (e.g., "reqid"), it first looks for a public "getXxx()"
* method to extract the specified field. If that fails, then it looks for a public field
* by the given name. If that also fails, and the object is a <i>Map</i> subclass, then it
/**
* Constructor.
- *
+ *
* @param props properties that specify how the data is to be extracted from
* a given class
* @param prefix property name prefix, prepended before the class name
/**
* Gets the number of extractors in the map.
- *
+ *
* @return gets the number of extractors in the map
*/
protected int size() {
/**
* Extracts the desired data item from an object.
- *
+ *
* @param object object from which to extract the data item
* @return the extracted item, or {@code null} if it could not be extracted
*/
/**
* Gets the extractor for the given type of object, creating one if it
* doesn't exist yet.
- *
+ *
* @param object object whose extracted is desired
* @return an extractor for the object
*/
/**
* Builds an extractor for the class.
- *
+ *
* @param clazz class for which the extractor should be built
- *
+ *
* @return a new extractor
*/
private Extractor buildExtractor(Class<?> clazz) {
/**
* Builds an extractor for the class, based on the config value extracted
* from the corresponding property.
- *
+ *
* @param clazz class for which the extractor should be built
* @param value config value (e.g., "${event.request.id}"
* @return a new extractor
/**
* Gets the extractor for a class, examining all super classes and
* interfaces.
- *
+ *
* @param clazz class whose extractor is desired
* @param addOk {@code true} if the extractor may be added, provided the
* property is defined, {@code false} otherwise
* hierarchically, where each name identifies a particular component within
* the hierarchy. Supports retrieval from {@link Map} objects, as well as
* via getXxx() methods, or by direct field retrieval.
- *
+ *
* <p>Note: this will <i>not</i> work if POJOs are contained within a Map.
*/
private class ComponetizedExtractor implements Extractor {
/**
* Constructor.
- *
+ *
* @param clazz the class associated with the object at the root of the
* hierarchy
* @param names name associated with each component
Pair<Extractor, Class<?>> pair = buildExtractor(clz, comp);
- extractors[x] = pair.first();
- clz = pair.second();
+ extractors[x] = pair.getLeft();
+ clz = pair.getRight();
}
}
/**
* Builds an extractor for the given component of an object.
- *
+ *
* @param clazz type of object from which the component will be
* extracted
* @param comp name of the component to extract
* @throws ExtractorException extrator exception
*/
private Pair<Extractor, Class<?>> buildExtractor(Class<?> clazz, String comp) throws ExtractorException {
-
+
Pair<Extractor, Class<?>> pair = getMethodExtractor(clazz, comp);
-
+
if (pair == null) {
pair = getFieldExtractor(clazz, comp);
}
/**
* Gets an extractor that invokes a getXxx() method to retrieve the
* object.
- *
+ *
* @param clazz container's class
* @param name name of the property to be retrieved
* @return a new extractor, or {@code null} if the class does not
return null;
}
- return new Pair<>(new MethodExtractor(meth), retType);
+ return Pair.of(new MethodExtractor(meth), retType);
} catch (NoSuchMethodException expected) {
// no getXxx() method, maybe there's a field by this name
/**
* Gets an extractor for a field within the object.
- *
+ *
* @param clazz container's class
* @param name name of the field whose value is to be extracted
* @return a new extractor, or {@code null} if the class does not
return null;
}
- return new Pair<>(new FieldExtractor(field), field.getType());
+ return Pair.of(new FieldExtractor(field), field.getType());
}
/**
* Gets an extractor for an item within a Map object.
- *
+ *
* @param clazz container's class
* @param key item key within the map
* @return a new extractor, or {@code null} if the class is not a Map
* If the object turns out not to be a map, then the MapExtractor
* for the next component will just return null.
*/
- return new Pair<>(new MapExtractor(key), Map.class);
+ return Pair.of(new MapExtractor(key), Map.class);
}
/**
* Gets field within a class, examining all super classes and
* interfaces.
- *
+ *
* @param clazz class whose field is desired
* @param name name of the desired field
* @return the field within the class, or {@code null} if the field does
* ONAP
* ================================================================================
* Copyright (C) 2018-2019 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2020 Nordix Foundation
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
+import org.apache.commons.lang3.tuple.Pair;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.onap.policy.drools.pooling.message.Message;
import org.onap.policy.drools.system.PolicyController;
import org.onap.policy.drools.system.PolicyEngine;
-import org.onap.policy.drools.utils.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
*
* <p>Invoke {@link #runSlow()}, before the test, to slow things down.
*/
-public class FeatureTest {
+public class FeatureTest {
private static final Logger logger = LoggerFactory.getLogger(FeatureTest.class);
-
/**
* Name of the topic used for inter-host communication.
*/
private static final String INTERNAL_TOPIC = "my.internal.topic";
-
/**
* Name of the topic from which "external" events "arrive".
*/
private static final String EXTERNAL_TOPIC = "my.external.topic";
-
/**
* Name of the controller.
*/
private static final String CONTROLLER1 = "controller.one";
-
private static long stdReactivateWaitMs = 200;
private static long stdIdentificationMs = 60;
private static long stdStartHeartbeatMs = 60;
* Used to decode events from the external topic.
*/
private static final Gson mapper = new Gson();
-
/**
* Used to identify the current context.
*/
private static final ThreadLocal<Context> currentContext = new ThreadLocal<Context>();
-
/**
* Context for the current test case.
*/
private Context ctx;
-
/**
* Setup.
*/
+
@Before
public void setUp() {
ctx = null;
}
-
/**
* Tear down.
*/
+
@After
public void tearDown() {
if (ctx != null) {
private void run(int nmessages, int nhosts) throws Exception {
ctx = new Context(nmessages);
-
for (int x = 0; x < nhosts; ++x) {
ctx.addHost();
}
-
ctx.startHosts();
-
for (int x = 0; x < nmessages; ++x) {
ctx.offerExternal(makeMessage(x));
}
-
ctx.awaitEvents(stdEventWaitSec, TimeUnit.SECONDS);
-
assertEquals(0, ctx.getDecodeErrors());
assertEquals(0, ctx.getRemainingEvents());
ctx.checkAllSawAMsg();
private String makeMessage(int reqnum) {
return "{\"reqid\":\"req" + reqnum + "\", \"data\":\"hello " + reqnum + "\"}";
}
-
/**
* Invoke this to slow the timers down.
*/
+
protected static void runSlow() {
stdReactivateWaitMs = 10000;
stdIdentificationMs = 10000;
stdInterPollMs = 2000;
stdEventWaitSec = 1000;
}
-
/**
* Decodes an event.
*
* @param event event
* @return the decoded event, or {@code null} if it cannot be decoded
*/
+
private static Object decodeEvent(String event) {
try {
return mapper.fromJson(event, TreeMap.class);
-
} catch (JsonParseException e) {
logger.warn("cannot decode external event", e);
return null;
}
}
-
/**
* Context used for a single test case.
*/
- private static class Context {
+ private static class Context {
/**
* Hosts that have been added to this context.
*/
private final Deque<Host> hosts = new LinkedList<>();
-
/**
* Maps a drools controller to its policy controller.
*/
private final IdentityHashMap<DroolsController, PolicyController> drools2policy = new IdentityHashMap<>();
-
/**
* Maps a channel to its queue. Does <i>not</i> include the "admin" channel.
*/
private final ConcurrentMap<String, BlockingQueue<String>> channel2queue = new ConcurrentHashMap<>(7);
-
/**
* Queue for the external "DMaaP" topic.
*/
private final BlockingQueue<String> externalTopic = new LinkedBlockingQueue<String>();
-
/**
* Counts the number of decode errors.
*/
private final AtomicInteger numDecodeErrors = new AtomicInteger(0);
-
/**
* Number of events we're still waiting to receive.
*/
private final CountDownLatch eventCounter;
-
/**
* The current host. Set by {@link #withHost(Host, VoidFunction)} and used by
* {@link #getCurrentHost()}.
*
* @param nEvents number of events to be processed
*/
+
public Context(int events) {
eventCounter = new CountDownLatch(events);
}
-
/**
* Destroys the context, stopping any hosts that remain.
*/
+
public void destroy() {
stopHosts();
hosts.clear();
*
* @return the new Host
*/
+
public Host addHost() {
Host host = new Host(this);
hosts.add(host);
-
return host;
}
/**
* Starts the hosts.
*/
+
public void startHosts() {
hosts.forEach(host -> host.start());
}
/**
* Stops the hosts.
*/
+
public void stopHosts() {
hosts.forEach(host -> host.stop());
}
/**
* Verifies that all hosts processed at least one message.
*/
+
public void checkAllSawAMsg() {
int msgs = 0;
for (Host host : hosts) {
* @param host host
* @param func function to invoke
*/
+
public void withHost(Host host, VoidFunction func) {
currentHost = host;
func.apply();
*
* @param event event
*/
+
public void offerExternal(String event) {
externalTopic.offer(event);
}
* @param channel channel
* @param queue the channel's queue
*/
+
public void addInternal(String channel, BlockingQueue<String> queue) {
channel2queue.put(channel, queue);
}
*
* @param message message
*/
+
public void offerInternal(String message) {
channel2queue.values().forEach(queue -> queue.offer(message));
}
* @param channel channel
* @param message message
*/
+
public void offerInternal(String channel, String message) {
BlockingQueue<String> queue = channel2queue.get(channel);
if (queue != null) {
* @param controller controller
* @param droolsController drools controller
*/
+
public void addController(PolicyController controller, DroolsController droolsController) {
drools2policy.put(droolsController, controller);
}
* @return the controller associated with a drools controller, or {@code null} if
* it has no associated controller
*/
+
public PolicyController getController(DroolsController droolsController) {
return drools2policy.get(droolsController);
}
*
* @return queue for the external topic
*/
+
public BlockingQueue<String> getExternalTopic() {
return externalTopic;
}
*
* @return the number of decode errors so far
*/
+
public int getDecodeErrors() {
return numDecodeErrors.get();
}
/**
* Increments the count of decode errors.
*/
+
public void bumpDecodeErrors() {
numDecodeErrors.incrementAndGet();
}
*
* @return the number of events that haven't been processed
*/
+
public long getRemainingEvents() {
return eventCounter.getCount();
}
/**
* Adds an event to the counter.
*/
+
public void addEvent() {
eventCounter.countDown();
}
* @return {@code true} if all events have been processed, {@code false} otherwise
* @throws InterruptedException throws interrupted
*/
+
public boolean awaitEvents(long time, TimeUnit units) throws InterruptedException {
return eventCounter.await(time, units);
}
*
* @return the current host, or {@code null} if there is no current host
*/
+
public Host getCurrentHost() {
return currentHost;
}
/**
* Simulates a single "host".
*/
- private static class Host {
+ private static class Host {
private final Context context;
-
private final PoolingFeature feature;
/**
* {@code True} if this host has processed a message, {@code false} otherwise.
*/
+
private final AtomicBoolean sawMsg = new AtomicBoolean(false);
/**
* This host's internal "DMaaP" topic.
*/
+
private final BlockingQueue<String> msgQueue = new LinkedBlockingQueue<>();
/**
* Source that reads from the external topic and posts to the listener.
*/
+
private TopicSource externalSource;
// mock objects
*
* @param context context
*/
+
public Host(Context context) {
this.context = context;
-
when(controller.getName()).thenReturn(CONTROLLER1);
when(controller.getDrools()).thenReturn(drools);
-
// stop consuming events if the controller stops
when(controller.stop()).thenAnswer(args -> {
externalSource.unregister(controller);
return true;
});
-
doAnswer(new MyExternalTopicListener(context, this)).when(controller).onTopicEvent(any(), any(), any());
-
context.addController(controller, drools);
-
// arrange to read from the external topic
externalSource = new TopicSourceImpl(context, false);
-
feature = new PoolingFeatureImpl(context);
}
*
* @return the host name
*/
+
public String getName() {
return feature.getHost();
}
* Starts threads for the host so that it begins consuming from both the external
* "DMaaP" topic and its own internal "DMaaP" topic.
*/
- public void start() {
+ public void start() {
context.withHost(this, () -> {
-
feature.beforeStart(engine);
feature.afterCreate(controller);
-
// assign the queue for this host's internal topic
context.addInternal(getName(), msgQueue);
-
feature.beforeStart(controller);
-
// start consuming events from the external topic
externalSource.register(controller);
-
feature.afterStart(controller);
});
}
/**
* Stops the host's threads.
*/
+
public void stop() {
feature.beforeStop(controller);
externalSource.unregister(controller);
* @param event event
* @return {@code true} if the event was handled, {@code false} otherwise
*/
+
public boolean beforeOffer(CommInfrastructure protocol, String topic2, String event) {
return feature.beforeOffer(controller, protocol, topic2, event);
}
* @param success success
* @return {@code true} if the event was handled, {@code false} otherwise
*/
- public boolean afterOffer(CommInfrastructure protocol, String topic, String event, boolean success) {
+ public boolean afterOffer(CommInfrastructure protocol, String topic, String event, boolean success) {
return feature.afterOffer(controller, protocol, topic, event, success);
}
* @param fact fact
* @return {@code true} if the event was handled, {@code false} otherwise
*/
+
public boolean beforeInsert(Object fact) {
return feature.beforeInsert(drools, fact);
}
* controller, {@code false} otherwise
* @return {@code true} if the event was handled, {@code false} otherwise
*/
+
public boolean afterInsert(Object fact, boolean successInsert) {
return feature.afterInsert(drools, fact, successInsert);
}
/**
* Indicates that a message was seen for this host.
*/
+
public void sawMessage() {
sawMsg.set(true);
}
* @return {@code true} if a message was seen for this host, {@code false}
* otherwise
*/
+
public boolean messageSeen() {
return sawMsg.get();
}
*
* @return the queue associated with this host's internal topic
*/
+
public BlockingQueue<String> getInternalQueue() {
return msgQueue;
}
* Listener for the external topic. Simulates the actions taken by
* <i>AggregatedPolicyController.onTopicEvent</i>.
*/
- private static class MyExternalTopicListener implements Answer<Void> {
+ private static class MyExternalTopicListener implements Answer<Void> {
private final Context context;
private final Host host;
CommInfrastructure commType = args.getArgument(index++);
String topic = args.getArgument(index++);
String event = args.getArgument(index++);
-
if (host.beforeOffer(commType, topic, event)) {
return null;
}
-
boolean result;
Object fact = decodeEvent(event);
-
if (fact == null) {
result = false;
context.bumpDecodeErrors();
-
} else {
result = true;
-
if (!host.beforeInsert(fact)) {
// feature did not handle it so we handle it here
host.afterInsert(fact, result);
-
host.sawMessage();
context.addEvent();
}
}
-
host.afterOffer(commType, topic, event, result);
return null;
}
* <i>channel</i> embedded within the message. If it's the "admin" channel, then the
* message is placed on all queues.
*/
- private static class TopicSinkImpl extends TopicImpl implements TopicSink {
+ private static class TopicSinkImpl extends TopicImpl implements TopicSink {
private final Context context;
-
/**
* Used to decode the messages so that the channel can be extracted.
*/
*
* @param context context
*/
+
public TopicSinkImpl(Context context) {
this.context = context;
}
if (!isAlive()) {
return false;
}
-
try {
Message msg = serializer.decodeMsg(message);
String channel = msg.getChannel();
-
if (Message.ADMIN.equals(channel)) {
// add to every queue
context.offerInternal(message);
-
} else {
// add to a specific queue
context.offerInternal(channel, message);
}
-
return true;
-
} catch (JsonParseException e) {
logger.warn("could not decode message: {}", message);
context.bumpDecodeErrors();
/**
* Source implementation that reads from a queue associated with a topic.
*/
+
private static class TopicSourceImpl extends TopicImpl implements FilterableTopicSource {
private final String topic;
-
/**
* Queue from which to retrieve messages.
*/
private final BlockingQueue<String> queue;
-
/**
* Manages the current consumer thread. The "first" item is used as a trigger to
* tell the thread to stop processing, while the "second" item is triggered <i>by
* @param internal {@code true} if to read from the internal topic, {@code false}
* to read from the external topic
*/
+
public TopicSourceImpl(Context context, boolean internal) {
if (internal) {
this.topic = INTERNAL_TOPIC;
this.queue = context.getCurrentHost().getInternalQueue();
-
} else {
this.topic = EXTERNAL_TOPIC;
this.queue = context.getExternalTopic();
* Starts a thread that takes messages from the queue and gives them to the
* listener. Stops the thread of any previously registered listener.
*/
+
@Override
public void register(TopicListener listener) {
- Pair<CountDownLatch, CountDownLatch> newPair = new Pair<>(new CountDownLatch(1), new CountDownLatch(1));
-
+ Pair<CountDownLatch, CountDownLatch> newPair = Pair.of(new CountDownLatch(1), new CountDownLatch(1));
reregister(newPair);
-
Thread thread = new Thread(() -> {
-
try {
do {
- processMessages(newPair.first(), listener);
- } while (!newPair.first().await(stdInterPollMs, TimeUnit.MILLISECONDS));
-
+ processMessages(newPair.getLeft(), listener);
+ } while (!newPair.getLeft().await(stdInterPollMs, TimeUnit.MILLISECONDS));
logger.info("topic source thread completed");
-
} catch (InterruptedException e) {
logger.warn("topic source thread aborted", e);
Thread.currentThread().interrupt();
-
} catch (RuntimeException e) {
logger.warn("topic source thread aborted", e);
}
-
- newPair.second().countDown();
-
+ newPair.getRight().countDown();
});
-
thread.setDaemon(true);
thread.start();
}
/**
* Stops the thread of <i>any</i> currently registered listener.
*/
+
@Override
public void unregister(TopicListener listener) {
reregister(null);
*
* @param newPair the new "pair", or {@code null} to unregister
*/
+
private void reregister(Pair<CountDownLatch, CountDownLatch> newPair) {
try {
Pair<CountDownLatch, CountDownLatch> oldPair = pair.getAndSet(newPair);
// unregister was invoked twice in a row
logger.warn("re-unregister for topic source");
}
-
// no previous thread to stop
return;
}
-
// need to stop the previous thread
-
// tell it to stop
- oldPair.first().countDown();
-
+ oldPair.getLeft().countDown();
// wait for it to stop
- if (!oldPair.second().await(2, TimeUnit.SECONDS)) {
+ if (!oldPair.getRight().await(2, TimeUnit.SECONDS)) {
logger.warn("old topic registration is still running");
}
-
} catch (InterruptedException e) {
logger.warn("old topic registration may still be running", e);
Thread.currentThread().interrupt();
}
-
if (newPair != null) {
// register was invoked twice in a row
logger.warn("re-register for topic source");
* @param listener listener
* @throws InterruptedException throws interrupted exception
*/
- private void processMessages(CountDownLatch stopped, TopicListener listener) throws InterruptedException {
+ private void processMessages(CountDownLatch stopped, TopicListener listener) throws InterruptedException {
for (int x = 0; x < 5 && stopped.getCount() > 0; ++x) {
-
String msg = queue.poll(stdPollMs, TimeUnit.MILLISECONDS);
if (msg == null) {
return;
}
-
listener.onTopicEvent(CommInfrastructure.UEB, topic, msg);
}
}
* Topic implementation. Most methods just throw
* {@link UnsupportedOperationException}.
*/
+
private static class TopicImpl implements Topic {
/**
* Constructor.
*/
+
public TopicImpl() {
super();
}
/**
* Feature with overrides.
*/
- private static class PoolingFeatureImpl extends PoolingFeature {
+ private static class PoolingFeatureImpl extends PoolingFeature {
private final Context context;
/**
*
* @param context context
*/
+
public PoolingFeatureImpl(Context context) {
this.context = context;
-
/*
* Note: do NOT extract anything from "context" at this point, because it
* hasn't been fully initialized yet
@Override
public Properties getProperties(String featName) {
Properties props = new Properties();
-
props.setProperty(PoolingProperties.PROP_EXTRACTOR_PREFIX + ".java.util.Map", "${reqid}");
-
props.setProperty(specialize(PoolingProperties.FEATURE_ENABLED, CONTROLLER1), "true");
props.setProperty(specialize(PoolingProperties.POOLING_TOPIC, CONTROLLER1), INTERNAL_TOPIC);
props.setProperty(specialize(PoolingProperties.OFFLINE_LIMIT, CONTROLLER1), "10000");
"" + stdActiveHeartbeatMs);
props.setProperty(specialize(PoolingProperties.INTER_HEARTBEAT_MS, CONTROLLER1),
"" + stdInterHeartbeatMs);
-
return props;
}
* @param spec specializer to be embedded
* @return the property name, with the specializer embedded within it
*/
+
private String specialize(String propnm, String spec) {
String suffix = propnm.substring(PREFIX.length());
return PREFIX + spec + "." + suffix;
@Override
protected PoolingManagerImpl makeManager(String host, PolicyController controller, PoolingProperties props,
CountDownLatch activeLatch) {
-
currentContext.set(context);
-
return new PoolingManagerTest(host, controller, props, activeLatch);
}
}
/**
* Pooling Manager with overrides.
*/
+
private static class PoolingManagerTest extends PoolingManagerImpl {
/**
* @param props the properties
* @param activeLatch the latch
*/
+
public PoolingManagerTest(String host, PolicyController controller, PoolingProperties props,
CountDownLatch activeLatch) {
-
super(host, controller, props, activeLatch);
}
/**
* DMaaP Manager with overrides.
*/
+
private static class DmaapManagerImpl extends DmaapManager {
/**
* @param topic the topic
* @throws PoolingFeatureException if an error occurs
*/
+
public DmaapManagerImpl(String topic) throws PoolingFeatureException {
super(topic);
}
/**
* Controller that also implements the {@link TopicListener} interface.
*/
- private static interface ListenerController extends PolicyController, TopicListener {
+ private static interface ListenerController extends PolicyController, TopicListener {
}
/**
* Simple function that takes no arguments and returns nothing.
*/
+
@FunctionalInterface
private static interface VoidFunction {
-
void apply();
}
}
* ONAP
* ================================================================================
* Copyright (C) 2018 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2020 Nordix Foundation
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
import java.util.List;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
+import org.apache.commons.lang3.tuple.Pair;
import org.junit.Before;
import org.junit.Test;
import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
import org.onap.policy.drools.controller.DroolsController;
import org.onap.policy.drools.system.PolicyController;
import org.onap.policy.drools.system.PolicyEngine;
-import org.onap.policy.drools.utils.Pair;
public class PoolingFeatureTest {
/**
* Setup.
- *
+ *
* @throws Exception exception
*/
@Before
pool.afterCreate(controller1);
pool.afterCreate(controller2);
- mgr1 = managers.get(0).first();
- mgr2 = managers.get(1).first();
+ mgr1 = managers.get(0).getLeft();
+ mgr2 = managers.get(1).getLeft();
}
@Test
verify(mgr1).afterStop();
assertFalse(pool.afterStop(controllerDisabled));
-
+
// count should be unchanged
verify(mgr1).afterStop();
}
public void testAfterHalt() {
assertFalse(pool.afterHalt(controller1));
assertFalse(pool.afterHalt(controller1));
-
+
verify(mgr1, never()).afterStop();
assertFalse(pool.afterStop(controllerDisabled));
public void testAfterShutdown() {
assertFalse(pool.afterShutdown(controller1));
assertFalse(pool.afterShutdown(controller1));
-
+
verify(mgr1, never()).afterStop();
assertFalse(pool.afterStop(controllerDisabled));
PoolingManagerImpl mgr = mock(PoolingManagerImpl.class);
- managers.add(new Pair<>(mgr, props));
+ managers.add(Pair.of(mgr, props));
return mgr;
}
* ONAP
* ================================================================================
* Copyright (C) 2018 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2020 Nordix Foundation
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
import java.util.Arrays;
import java.util.Map;
+import org.apache.commons.lang3.tuple.Pair;
import org.junit.Before;
import org.junit.Test;
import org.onap.policy.drools.pooling.message.BucketAssignments;
import org.onap.policy.drools.pooling.message.Message;
import org.onap.policy.drools.pooling.message.Offline;
import org.onap.policy.drools.pooling.message.Query;
-import org.onap.policy.drools.utils.Pair;
import org.onap.policy.drools.utils.Triple;
public class ActiveStateTest extends SupportBasicStateTester {
/**
* Setup.
*/
+ @Override
@Before
public void setUp() throws Exception {
super.setUp();
// ensure a heart beat was generated
Pair<String, Heartbeat> msg = capturePublishedMessage(Heartbeat.class);
- assertEquals(MY_HOST, msg.second().getSource());
+ assertEquals(MY_HOST, msg.getRight().getSource());
}
@Test
state = new ActiveState(mgr);
/*
- *
+ *
* PREV_HOST2 has buckets and is my predecessor, but it isn't the leader thus
* should be ignored.
*/
verify(mgr, times(2)).publish(anyString(), any(Heartbeat.class));
Pair<String, Heartbeat> msg = capturePublishedMessage(Heartbeat.class);
- assertEquals(MY_HOST, msg.first());
- assertEquals(MY_HOST, msg.second().getSource());
+ assertEquals(MY_HOST, msg.getLeft());
+ assertEquals(MY_HOST, msg.getRight().getSource());
}
@Test
verify(mgr, times(1)).publish(any(), any());
Pair<String, Heartbeat> msg = capturePublishedMessage(Heartbeat.class);
- assertEquals(MY_HOST, msg.first());
- assertEquals(MY_HOST, msg.second().getSource());
+ assertEquals(MY_HOST, msg.getLeft());
+ assertEquals(MY_HOST, msg.getRight().getSource());
}
@Test
// this message should go to itself
msg = capturePublishedMessage(Heartbeat.class, index++);
- assertEquals(MY_HOST, msg.first());
- assertEquals(MY_HOST, msg.second().getSource());
+ assertEquals(MY_HOST, msg.getLeft());
+ assertEquals(MY_HOST, msg.getRight().getSource());
// this message should go to its successor
msg = capturePublishedMessage(Heartbeat.class, index++);
- assertEquals(HOST1, msg.first());
- assertEquals(MY_HOST, msg.second().getSource());
+ assertEquals(HOST1, msg.getLeft());
+ assertEquals(MY_HOST, msg.getRight().getSource());
}
}
* ONAP
* ================================================================================
* Copyright (C) 2018 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2020 Nordix Foundation
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
import static org.mockito.Mockito.when;
import java.util.Map;
+import org.apache.commons.lang3.tuple.Pair;
import org.junit.Before;
import org.junit.Test;
import org.onap.policy.drools.pooling.message.BucketAssignments;
import org.onap.policy.drools.pooling.message.Leader;
import org.onap.policy.drools.pooling.message.Message;
import org.onap.policy.drools.pooling.message.Query;
-import org.onap.policy.drools.utils.Pair;
public class InactiveStateTest extends SupportBasicStateTester {
/**
* Setup.
- *
+ *
*/
+ @Override
@Before
public void setUp() throws Exception {
super.setUp();
Pair<Long, StateTimerTask> timer = onceTasks.remove();
- assertEquals(STD_REACTIVATE_WAIT_MS, timer.first().longValue());
+ assertEquals(STD_REACTIVATE_WAIT_MS, timer.getLeft().longValue());
// invoke the task - it should go to the state returned by the mgr
State next = mock(State.class);
when(mgr.goStart()).thenReturn(next);
- assertEquals(next, timer.second().fire());
+ assertEquals(next, timer.getRight().fire());
}
@Test
* ONAP
* ================================================================================
* Copyright (C) 2018 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2020 Nordix Foundation
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
import static org.mockito.Mockito.when;
import java.util.Map;
+import org.apache.commons.lang3.tuple.Pair;
import org.junit.Before;
import org.junit.Test;
import org.onap.policy.drools.pooling.message.BucketAssignments;
import org.onap.policy.drools.pooling.message.Leader;
import org.onap.policy.drools.pooling.message.Message;
import org.onap.policy.drools.pooling.message.Offline;
-import org.onap.policy.drools.utils.Pair;
public class QueryStateTest extends SupportBasicStateTester {
/**
* Setup.
*/
+ @Override
@Before
public void setUp() throws Exception {
super.setUp();
Pair<Long, StateTimerTask> timer = onceTasks.remove();
- assertEquals(STD_IDENTIFICATION_MS, timer.first().longValue());
- assertNotNull(timer.second());
+ assertEquals(STD_IDENTIFICATION_MS, timer.getLeft().longValue());
+ assertNotNull(timer.getRight());
}
@Test
Pair<Long, StateTimerTask> timer = onceTasks.remove();
- assertEquals(STD_IDENTIFICATION_MS, timer.first().longValue());
- assertNotNull(timer.second());
+ assertEquals(STD_IDENTIFICATION_MS, timer.getLeft().longValue());
+ assertNotNull(timer.getRight());
// should published an Offline message and go inactive
State next = mock(State.class);
when(mgr.goStart()).thenReturn(next);
- assertEquals(next, timer.second().fire());
+ assertEquals(next, timer.getRight().fire());
// should continue distributing
verify(mgr, never()).startDistributing(null);
Pair<Long, StateTimerTask> timer = onceTasks.remove();
- assertEquals(STD_IDENTIFICATION_MS, timer.first().longValue());
- assertNotNull(timer.second());
+ assertEquals(STD_IDENTIFICATION_MS, timer.getLeft().longValue());
+ assertNotNull(timer.getRight());
State next = mock(State.class);
when(mgr.goActive()).thenReturn(next);
- assertEquals(next, timer.second().fire());
+ assertEquals(next, timer.getRight().fire());
// should have published a Leader message
Leader msg = captureAdminMessage(Leader.class);
Pair<Long, StateTimerTask> timer = onceTasks.remove();
- assertEquals(STD_IDENTIFICATION_MS, timer.first().longValue());
- assertNotNull(timer.second());
+ assertEquals(STD_IDENTIFICATION_MS, timer.getLeft().longValue());
+ assertNotNull(timer.getRight());
// set up active state, as that's what it should return
State next = mock(State.class);
when(mgr.goActive()).thenReturn(next);
- assertEquals(next, timer.second().fire());
+ assertEquals(next, timer.getRight().fire());
// should NOT have published a Leader message
assertTrue(admin.isEmpty());
Pair<Long, StateTimerTask> timer = onceTasks.remove();
- assertEquals(STD_IDENTIFICATION_MS, timer.first().longValue());
- assertNotNull(timer.second());
+ assertEquals(STD_IDENTIFICATION_MS, timer.getLeft().longValue());
+ assertNotNull(timer.getRight());
// set up inactive state, as that's what it should return
State next = mock(State.class);
when(mgr.goInactive()).thenReturn(next);
- assertEquals(next, timer.second().fire());
+ assertEquals(next, timer.getRight().fire());
// should NOT have published a Leader message
assertTrue(admin.isEmpty());
* ONAP
* ================================================================================
* Copyright (C) 2018 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2020 Nordix Foundation
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
import static org.mockito.Mockito.when;
import java.util.Map;
+import org.apache.commons.lang3.tuple.Pair;
import org.junit.Before;
import org.junit.Test;
import org.onap.policy.drools.pooling.message.Forward;
import org.onap.policy.drools.pooling.message.Message;
import org.onap.policy.drools.pooling.message.Offline;
import org.onap.policy.drools.pooling.message.Query;
-import org.onap.policy.drools.utils.Pair;
import org.onap.policy.drools.utils.Triple;
public class StartStateTest extends SupportBasicStateTester {
/**
* Setup.
*/
+ @Override
@Before
public void setUp() throws Exception {
super.setUp();
Pair<String, Heartbeat> msg = capturePublishedMessage(Heartbeat.class);
- assertEquals(MY_HOST, msg.first());
- assertEquals(state.getHbTimestampMs(), msg.second().getTimestampMs());
+ assertEquals(MY_HOST, msg.getLeft());
+ assertEquals(state.getHbTimestampMs(), msg.getRight().getTimestampMs());
/*
// invoke the task - it should generate another heartbeat
assertEquals(null, generator.third().fire());
- verify(mgr, times(2)).publish(MY_HOST, msg.second());
+ verify(mgr, times(2)).publish(MY_HOST, msg.getRight());
// and again
assertEquals(null, generator.third().fire());
- verify(mgr, times(3)).publish(MY_HOST, msg.second());
+ verify(mgr, times(3)).publish(MY_HOST, msg.getRight());
/*
*/
Pair<Long, StateTimerTask> checker = onceTasks.removeFirst();
- assertEquals(STD_HEARTBEAT_WAIT_MS, checker.first().longValue());
+ assertEquals(STD_HEARTBEAT_WAIT_MS, checker.getLeft().longValue());
// invoke the task - it should go to the state returned by the mgr
State next = mock(State.class);
when(mgr.goInactive()).thenReturn(next);
- assertEquals(next, checker.second().fire());
+ assertEquals(next, checker.getRight().fire());
verify(mgr).startDistributing(null);
}
* ONAP
* ================================================================================
* Copyright (C) 2018 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2020 Nordix Foundation
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicReference;
+import org.apache.commons.lang3.tuple.Pair;
import org.onap.policy.drools.pooling.CancellableScheduledTask;
import org.onap.policy.drools.pooling.PoolingManager;
import org.onap.policy.drools.pooling.PoolingProperties;
import org.onap.policy.drools.pooling.message.BucketAssignments;
import org.onap.policy.drools.pooling.message.Leader;
import org.onap.policy.drools.pooling.message.Message;
-import org.onap.policy.drools.utils.Pair;
import org.onap.policy.drools.utils.Triple;
/**
/**
* Setup.
- *
+ *
* @throws Exception throws exception
*/
public void setUp() throws Exception {
// capture publish() arguments
doAnswer(invocation -> {
Object[] args = invocation.getArguments();
- published.add(new Pair<>((String) args[0], (Message) args[1]));
+ published.add(Pair.of((String) args[0], (Message) args[1]));
return null;
}).when(mgr).publish(anyString(), any(Message.class));
// capture schedule() arguments, and return a new future
when(mgr.schedule(anyLong(), any(StateTimerTask.class))).thenAnswer(invocation -> {
Object[] args = invocation.getArguments();
- onceTasks.add(new Pair<>((Long) args[0], (StateTimerTask) args[1]));
+ onceTasks.add(Pair.of((Long) args[0], (StateTimerTask) args[1]));
CancellableScheduledTask sched = mock(CancellableScheduledTask.class);
onceSchedules.add(sched);
/**
* Makes a sorted set of hosts.
- *
+ *
* @param hosts the hosts to be sorted
* @return the set of hosts, sorted
*/
/**
* Captures the host array from the Leader message published to the admin channel.
- *
+ *
* @return the host array, as a list
*/
protected List<String> captureHostList() {
/**
* Captures the host array from the Leader message published to the admin channel.
- *
+ *
* @return the host array
*/
protected String[] captureHostArray() {
/**
* Captures the assignments from the Leader message published to the admin channel.
- *
+ *
* @return the bucket assignments
*/
protected BucketAssignments captureAssignments() {
/**
* Captures the message published to the admin channel.
- *
+ *
* @param clazz type of {@link Message} to capture
* @return the message that was published
*/
/**
* Captures the message published to the admin channel.
- *
+ *
* @param clazz type of {@link Message} to capture
* @param index index of the item to be captured
* @return the message that was published
/**
* Captures the message published to the non-admin channels.
- *
+ *
* @param clazz type of {@link Message} to capture
* @return the (channel,message) pair that was published
*/
/**
* Captures the message published to the non-admin channels.
- *
+ *
* @param clazz type of {@link Message} to capture
* @param index index of the item to be captured
* @return the (channel,message) pair that was published
*/
protected <T extends Message> Pair<String, T> capturePublishedMessage(Class<T> clazz, int index) {
Pair<String, Message> msg = published.get(index);
- return new Pair<>(msg.first(), clazz.cast(msg.second()));
+ return Pair.of(msg.getLeft(), clazz.cast(msg.getRight()));
}
}
* feature-server-pool
* ================================================================================
* Copyright (C) 2020 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2020 Nordix Foundation
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import lombok.AllArgsConstructor;
+import org.apache.commons.lang3.tuple.Pair;
import org.kie.api.runtime.KieSession;
import org.kie.api.runtime.rule.FactHandle;
import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
import org.onap.policy.drools.system.PolicyControllerConstants;
import org.onap.policy.drools.system.PolicyEngine;
import org.onap.policy.drools.system.PolicyEngineConstants;
-import org.onap.policy.drools.utils.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
kieSession.insert(backupAndRemove);
// add pending operation to the list
- pendingData.add(new Pair<>(droolsObjectsWrapper, session));
+ pendingData.add(Pair.of(droolsObjectsWrapper, session));
}
}
for (Pair<CompletableFuture<List<Object>>, PolicySession> pair :
pendingData) {
- PolicySession session = pair.second();
+ PolicySession session = pair.getRight();
long delay = endTime - System.currentTimeMillis();
if (delay < 0) {
/*
}
try {
List<Object> droolsObjects =
- pair.first().get(delay, TimeUnit.MILLISECONDS);
+ pair.getLeft().get(delay, TimeUnit.MILLISECONDS);
// if we reach this point, session data read has completed
logger.info("{}: session={}, got {} object(s)",
+++ /dev/null
-/*-
- * ============LICENSE_START=======================================================
- * policy-utils
- * ================================================================================
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.policy.drools.utils;
-
-public class Pair<F, S> {
-
- protected F first;
- protected S second;
-
- public Pair(F first, S second) {
- this.first = first;
- this.second = second;
- }
-
- public F first() {
- return this.first;
- }
-
- public void first(F first) {
- this.first = first;
- }
-
- public S second() {
- return this.second;
- }
-
- public void second(S second) {
- this.second = second;
- }
-
- public F getFirst() {
- return this.first;
- }
-
- public S getSecond() {
- return this.second;
- }
-
- @Override
- public String toString() {
- StringBuilder builder = new StringBuilder();
- builder.append("Pair [first=").append(first).append(", second=").append(second).append("]");
- return builder.toString();
- }
-}
+++ /dev/null
-/*-
- * ============LICENSE_START=======================================================
- * policy-utils
- * ================================================================================
- * Copyright (C) 2017-2018, 2020 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.policy.drools.utils;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-
-import org.junit.Test;
-
-public class PairTripleTest {
-
- @Test
- public void testPair() {
- Pair<String, String> pair = new Pair<String, String>("foo", "bar");
-
- assertEquals("foo", pair.first());
- assertEquals("bar", pair.second());
- assertEquals("foo", pair.getFirst());
- assertEquals("bar", pair.getSecond());
-
- pair.first("one");
- pair.second("two");
-
- assertEquals("one", pair.first());
- assertEquals("two", pair.second());
- assertEquals("one", pair.getFirst());
- assertEquals("two", pair.getSecond());
-
- assertNotNull(pair.toString());
-
- }
-
- @Test
- public void testTriple() {
- Triple<String, String, String> triple = new Triple<String, String, String>("foo", "bar", "fiz");
-
- assertEquals("foo", triple.first());
- assertEquals("bar", triple.second());
- assertEquals("fiz", triple.third());
-
- triple.first("one");
- triple.second("two");
- triple.third("three");
-
- assertEquals("one", triple.first());
- assertEquals("two", triple.second());
- assertEquals("three", triple.third());
- }
-
-}
* ONAP
* ================================================================================
* Copyright (C) 2018 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2020 Nordix Foundation
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
triple.setThird("III");
Assert.assertTrue("III".equals(triple.third()));
+
}
}
\ No newline at end of file
Code Review / policy / drools-pdp.git / commitdiff