diff options
Diffstat (limited to 'models-sim/models-sim-dmaap/src')
35 files changed, 2793 insertions, 415 deletions
diff --git a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/parameters/DmaapSimParameterGroup.java b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/parameters/DmaapSimParameterGroup.java index caae287b8..11da57582 100644 --- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/parameters/DmaapSimParameterGroup.java +++ b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/parameters/DmaapSimParameterGroup.java @@ -1,6 +1,7 @@ /*- * ============LICENSE_START======================================================= * Copyright (C) 2019 Nordix Foundation. + * Modifications Copyright (C) 2019 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. @@ -21,8 +22,8 @@ package org.onap.policy.models.sim.dmaap.parameters; import lombok.Getter; - import org.onap.policy.common.parameters.ParameterGroupImpl; +import org.onap.policy.common.parameters.annotations.Min; import org.onap.policy.common.parameters.annotations.NotBlank; import org.onap.policy.common.parameters.annotations.NotNull; @@ -36,6 +37,14 @@ public class DmaapSimParameterGroup extends ParameterGroupImpl { private RestServerParameters restServerParameters; /** + * Frequency, in seconds, with which to sweep the topics of idle consumers. On each + * sweep cycle, if a consumer group has had no new poll requests since the last sweep + * cycle, it is removed. + */ + @Min(1) + private long topicSweepSec; + + /** * Create the DMaaP simulator parameter group. * * @param name the parameter group name diff --git a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/parameters/DmaapSimParameterHandler.java b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/parameters/DmaapSimParameterHandler.java index 8eb76ec00..252054504 100644 --- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/parameters/DmaapSimParameterHandler.java +++ b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/parameters/DmaapSimParameterHandler.java @@ -1,6 +1,7 @@ /*- * ============LICENSE_START======================================================= * Copyright (C) 2019 Nordix Foundation. + * Modifications Copyright (C) 2019 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. @@ -38,7 +39,7 @@ public class DmaapSimParameterHandler { private static final Logger LOGGER = LoggerFactory.getLogger(DmaapSimParameterHandler.class); - private static final Coder CODER = new StandardCoder(); + private final Coder coder = new StandardCoder(); /** * Read the parameters from the parameter file. @@ -54,7 +55,7 @@ public class DmaapSimParameterHandler { try { // Read the parameters from JSON File file = new File(arguments.getFullConfigurationFilePath()); - dmaapSimParameterGroup = CODER.decode(file, DmaapSimParameterGroup.class); + dmaapSimParameterGroup = coder.decode(file, DmaapSimParameterGroup.class); } catch (final CoderException e) { final String errorMessage = "error reading parameters from \"" + arguments.getConfigurationFilePath() + "\"\n" + "(" + e.getClass().getSimpleName() + "):" + e.getMessage(); diff --git a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/parameters/RestServerParameters.java b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/parameters/RestServerParameters.java index c7269f66d..8fb2cffa3 100644 --- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/parameters/RestServerParameters.java +++ b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/parameters/RestServerParameters.java @@ -1,6 +1,7 @@ /*- * ============LICENSE_START======================================================= * Copyright (C) 2019 Nordix Foundation. + * Modifications Copyright (C) 2019 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. @@ -21,13 +22,14 @@ package org.onap.policy.models.sim.dmaap.parameters; import lombok.Getter; + import org.onap.policy.common.parameters.ParameterGroupImpl; import org.onap.policy.common.parameters.annotations.Min; import org.onap.policy.common.parameters.annotations.NotBlank; import org.onap.policy.common.parameters.annotations.NotNull; /** - * Class to hold all parameters needed for DMaaP simulator rest server. + * Class to hold all parameters needed for rest server. */ @NotNull @NotBlank @@ -39,6 +41,6 @@ public class RestServerParameters extends ParameterGroupImpl { private int port; public RestServerParameters() { - super("RestServerParameters"); + super(RestServerParameters.class.getSimpleName()); } } diff --git a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/ConsumerGroupData.java b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/ConsumerGroupData.java new file mode 100644 index 000000000..737151339 --- /dev/null +++ b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/ConsumerGroupData.java @@ -0,0 +1,190 @@ +/*- + * ============LICENSE_START======================================================= + * ONAP Policy Models + * ================================================================================ + * Copyright (C) 2019 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.models.sim.dmaap.provider; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Data associated with a consumer group. All consumer instances within a group share the + * same data object. + */ +public class ConsumerGroupData { + private static final Logger logger = LoggerFactory.getLogger(ConsumerGroupData.class); + + /** + * Returned when messages can no longer be read from this consumer group object, + * because it is being removed from the topic. {@link #UNREADABLE_LIST} must not be + * the same list as Collections.emptyList(), thus we wrap it. + */ + public static final List<String> UNREADABLE_LIST = Collections.unmodifiableList(Collections.emptyList()); + + /** + * Returned when there are no messages read. Collections.emptyList() is already + * unmodifiable, thus no need to wrap it. + */ + private static final List<String> EMPTY_LIST = Collections.emptyList(); + + /** + * This is locked while fields other than {@link #messageQueue} are updated. + */ + private final Object lockit = new Object(); + + /** + * Number of sweep cycles that have occurred since a consumer has attempted to read + * from the queue. This consumer group should be removed once this count exceeds + * {@code 1}, provided {@link #nreaders} is zero. + */ + private int nsweeps = 0; + + /** + * Number of consumers that are currently attempting to read from the queue. This + * consumer group should not be removed as long as this is non-zero. + */ + private int nreaders = 0; + + /** + * Message queue. + */ + private final BlockingQueue<String> messageQueue = new LinkedBlockingQueue<>(); + + + /** + * Constructs the object. + * + * @param topicName name of the topic with which this object is associated + * @param groupName name of the consumer group with which this object is associated + */ + public ConsumerGroupData(String topicName, String groupName) { + logger.info("Topic {}: add consumer group: {}", topicName, groupName); + } + + /** + * Determines if this consumer group should be removed. This should be invoked once + * during each sweep cycle. When this returns {@code true}, this consumer group should + * be immediately discarded, as any readers will sit in a spin loop waiting for it to + * be discarded. + * + * @return {@code true} if this consumer group should be removed, {@code false} + * otherwise + */ + public boolean shouldRemove() { + synchronized (lockit) { + return (nreaders == 0 && ++nsweeps > 1); + } + } + + /** + * Reads messages from the queue, blocking if necessary. + * + * @param maxRead maximum number of messages to read + * @param waitMs time to wait, in milliseconds, if the queue is currently empty + * @return a list of messages read from the queue, empty if no messages became + * available before the wait time elapsed, or {@link #UNREADABLE_LIST} if this + * consumer group object is no longer active + * @throws InterruptedException if this thread was interrupted while waiting for the + * first message + */ + public List<String> read(int maxRead, long waitMs) throws InterruptedException { + + synchronized (lockit) { + if (nsweeps > 1 && nreaders == 0) { + // cannot use this consumer group object anymore + return UNREADABLE_LIST; + } + + ++nreaders; + } + + /* + * Note: do EVERYTHING inside of the "try" block, so that the "finally" block can + * update the reader count. + * + * Do NOT hold the lockit while we're polling, as poll() may block for a while. + */ + try { + // always read at least one message + int maxRead2 = Math.max(1, maxRead); + long waitMs2 = Math.max(0, waitMs); + + // perform a blocking read of the queue + String obj = getNextMessage(waitMs2); + if (obj == null) { + return EMPTY_LIST; + } + + /* + * List should hold all messages from the queue PLUS the one we already have. + * Note: it's possible for additional messages to be added to the queue while + * we're reading from it. In that case, the list will grow as needed. + */ + List<String> lst = new ArrayList<>(Math.min(maxRead2, messageQueue.size() + 1)); + lst.add(obj); + + // perform NON-blocking read of subsequent messages + for (int x = 1; x < maxRead2; ++x) { + if ((obj = messageQueue.poll()) == null) { + break; + } + + lst.add(obj); + } + + return lst; + + } finally { + synchronized (lockit) { + --nreaders; + nsweeps = 0; + } + } + } + + /** + * Writes messages to the queue. + * + * @param messages messages to be written to the queue + */ + public void write(List<String> messages) { + messageQueue.addAll(messages); + } + + // the following methods may be overridden by junit tests + + /** + * Gets the next message from the queue. + * + * @param waitMs time to wait, in milliseconds, if the queue is currently empty + * @return the next message, or {@code null} if no messages became available before + * the wait time elapsed + * @throws InterruptedException if this thread was interrupted while waiting for the + * message + */ + protected String getNextMessage(long waitMs) throws InterruptedException { + return messageQueue.poll(waitMs, TimeUnit.MILLISECONDS); + } +} diff --git a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/DmaapSimProvider.java b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/DmaapSimProvider.java index 9de29cdac..d11d1b397 100644 --- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/DmaapSimProvider.java +++ b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/DmaapSimProvider.java @@ -1,6 +1,7 @@ /*- * ============LICENSE_START======================================================= * Copyright (C) 2019 Nordix Foundation. + * Modifications Copyright (C) 2019 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. @@ -20,18 +21,20 @@ package org.onap.policy.models.sim.dmaap.provider; +import java.util.Collections; import java.util.LinkedHashMap; +import java.util.List; import java.util.Map; -import java.util.SortedMap; -import java.util.TreeMap; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; - import javax.ws.rs.core.Response; - -import org.apache.commons.lang3.tuple.MutablePair; -import org.onap.policy.common.utils.coder.CoderException; -import org.onap.policy.common.utils.coder.StandardCoder; -import org.onap.policy.models.sim.dmaap.DmaapSimRuntimeException; +import javax.ws.rs.core.Response.Status; +import lombok.Getter; +import lombok.Setter; +import org.onap.policy.common.utils.services.ServiceManagerContainer; +import org.onap.policy.models.sim.dmaap.parameters.DmaapSimParameterGroup; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -40,50 +43,70 @@ import org.slf4j.LoggerFactory; * * @author Liam Fallon (liam.fallon@est.tech) */ -public class DmaapSimProvider { +public class DmaapSimProvider extends ServiceManagerContainer { private static final Logger LOGGER = LoggerFactory.getLogger(DmaapSimProvider.class); - // Recurring string constants - private static final String TOPIC_TAG = "Topic:"; + @Getter + @Setter + private static DmaapSimProvider instance; - // Time for a get to wait before checking of a message has come - private static final long DMAAP_SIM_WAIT_TIME = 50; + /** + * Maps a topic name to its data. + */ + private final Map<String, TopicData> topic2data = new ConcurrentHashMap<>(); - // recurring constants - private static final String WITH_TIMEOUT = " with timeout "; + /** + * Thread used to remove idle consumers from the topics. + */ + private ScheduledExecutorService timerPool; - // The map of topic messages - private static final Map<String, SortedMap<Integer, Object>> topicMessageMap = new LinkedHashMap<>(); - // The map of topic messages - private static final Map<String, Map<String, MutablePair<Integer, String>>> consumerGroupsMap = - new LinkedHashMap<>(); + /** + * Constructs the object. + * + * @param params parameters + */ + public DmaapSimProvider(DmaapSimParameterGroup params) { + addAction("Topic Sweeper", () -> { + timerPool = makeTimerPool(); + timerPool.scheduleWithFixedDelay(new SweeperTask(), params.getTopicSweepSec(), params.getTopicSweepSec(), + TimeUnit.SECONDS); + }, () -> timerPool.shutdown()); + } /** * Process a DMaaP message. * - * @param topicName The topic name + * @param topicName the topic name * @param dmaapMessage the message to process * @return a response to the message */ + @SuppressWarnings("unchecked") public Response processDmaapMessagePut(final String topicName, final Object dmaapMessage) { - LOGGER.debug(TOPIC_TAG + topicName + ", Received DMaaP message: " + dmaapMessage); - - synchronized (topicMessageMap) { - SortedMap<Integer, Object> messageMap = topicMessageMap.get(topicName); - if (messageMap == null) { - messageMap = new TreeMap<>(); - topicMessageMap.put(topicName, messageMap); - LOGGER.debug(TOPIC_TAG + topicName + ", created topic message map"); - } + LOGGER.debug("Topic: {}, Received DMaaP message(s): {}", topicName, dmaapMessage); - int nextKey = (messageMap.isEmpty() ? 0 : messageMap.lastKey() + 1); + List<Object> lst; - messageMap.put(nextKey, dmaapMessage); - LOGGER.debug(TOPIC_TAG + topicName + ", cached DMaaP message " + nextKey + ": " + dmaapMessage); + if (dmaapMessage instanceof List) { + lst = (List<Object>) dmaapMessage; + } else { + lst = Collections.singletonList(dmaapMessage); } - return Response.status(Response.Status.OK).entity("{\n \"serverTimeMs\": 0,\n \"count\": 1\n}").build(); + TopicData topic = topic2data.get(topicName); + + /* + * Write all messages and return the count. If the topic doesn't exist yet, then + * there are no subscribers to receive the messages, thus treat it as if all + * messages were published. + */ + int nmessages = (topic != null ? topic.write(lst) : lst.size()); + + Map<String, Object> map = new LinkedHashMap<>(); + map.put("serverTimeMs", 0); + map.put("count", nmessages); + + return Response.status(Response.Status.OK).entity(map).build(); } /** @@ -92,102 +115,66 @@ public class DmaapSimProvider { * @param topicName The topic to wait on * @param consumerGroup the consumer group that is waiting * @param consumerId the consumer ID that is waiting - * @param timeout the length of time to wait for + * @param limit the maximum number of messages to get + * @param timeoutMs the length of time to wait for * @return the DMaaP message or */ public Response processDmaapMessageGet(final String topicName, final String consumerGroup, final String consumerId, - final int timeout) { + final int limit, final long timeoutMs) { - LOGGER.debug(TOPIC_TAG + topicName + ", Request for DMaaP message: " + consumerGroup + ":" + consumerId - + WITH_TIMEOUT + timeout); + LOGGER.debug("Topic: {}, Request for DMaaP message: {}: {} with limit={} timeout={}", topicName, consumerGroup, + consumerId, limit, timeoutMs); - MutablePair<Integer, String> consumerGroupPair = null; - - synchronized (consumerGroupsMap) { - Map<String, MutablePair<Integer, String>> consumerGroupMap = consumerGroupsMap.get(topicName); - if (consumerGroupMap == null) { - consumerGroupMap = new LinkedHashMap<>(); - consumerGroupsMap.put(topicName, consumerGroupMap); - LOGGER.trace( - TOPIC_TAG + topicName + ", Created consumer map entry for consumer group " + consumerGroup); - } - - consumerGroupPair = consumerGroupMap.get(consumerGroup); - if (consumerGroupPair == null) { - consumerGroupPair = new MutablePair<>(-1, consumerId); - consumerGroupMap.put(consumerGroup, consumerGroupPair); - LOGGER.trace(TOPIC_TAG + topicName + ", Created consumer group entry for consumer group " - + consumerGroup + ":" + consumerId); - } - } + try { + List<String> lst = topic2data.computeIfAbsent(topicName, this::makeTopicData).read(consumerGroup, limit, + timeoutMs); - long timeOfTimeout = System.currentTimeMillis() + timeout; + if (lst.isEmpty() && timeoutMs > 0) { + LOGGER.debug("Topic: {}, Timed out waiting for messages: {}: {}", topicName, consumerGroup, consumerId); + return Response.status(Status.REQUEST_TIMEOUT).entity(lst).build(); - do { - Object waitingMessages = getWaitingMessages(topicName, consumerGroupPair); - if (waitingMessages != null) { - LOGGER.debug(TOPIC_TAG + topicName + ", Request for DMaaP message: " + consumerGroup + ":" + consumerId - + WITH_TIMEOUT + timeout + ", returning messages " + waitingMessages); - return Response.status(Response.Status.OK).entity(waitingMessages).build(); + } else { + LOGGER.debug("Topic: {}, Retrieved {} messages for: {}: {}", topicName, consumerGroup, lst.size(), + consumerId); + return Response.status(Status.OK).entity(lst).build(); } - try { - TimeUnit.MILLISECONDS.sleep(DMAAP_SIM_WAIT_TIME); - } catch (InterruptedException ie) { - String errorMessage = "Interrupt on wait on simulation of DMaaP topic " + topicName + " for request ID " - + consumerGroup + ":" + consumerId + WITH_TIMEOUT + timeout; - LOGGER.warn(errorMessage, ie); - Thread.currentThread().interrupt(); - throw new DmaapSimRuntimeException(errorMessage, ie); - } + } catch (InterruptedException e) { + LOGGER.warn("Topic: {}, Request for DMaaP message interrupted: {}: {}", topicName, consumerGroup, + consumerId, e); + Thread.currentThread().interrupt(); + return Response.status(Status.GONE).entity(Collections.emptyList()).build(); } - while (timeOfTimeout > System.currentTimeMillis()); - - LOGGER.trace(TOPIC_TAG + topicName + ", timed out waiting for messages : " + consumerGroup + ":" + consumerId - + WITH_TIMEOUT + timeout); - return Response.status(Response.Status.REQUEST_TIMEOUT).build(); } /** - * Return any messages on this topic with a message number greater than the supplied message number. - * - * @param topicName the topic name to check - * @param consumerGroupPair the pair with the information on the last message retrieved - * @return the messages or null if there are none + * Task to remove idle consumers from each topic. */ - private Object getWaitingMessages(final String topicName, final MutablePair<Integer, String> consumerGroupPair) { - String foundMessageList = "["; - - synchronized (topicMessageMap) { - SortedMap<Integer, Object> messageMap = topicMessageMap.get(topicName); - if (messageMap == null || messageMap.lastKey() <= consumerGroupPair.getLeft()) { - return null; - } + private class SweeperTask implements Runnable { + @Override + public void run() { + topic2data.values().forEach(TopicData::removeIdleConsumers); + } + } - boolean first = true; - for (Object dmaapMessage : messageMap.tailMap(consumerGroupPair.getLeft() + 1).values()) { - if (first) { - first = false; - } else { - foundMessageList += ","; - } - try { - foundMessageList += new StandardCoder().encode(dmaapMessage); - } catch (CoderException ce) { - String errorMessage = "Encoding error on message on DMaaP topic " + topicName; - LOGGER.warn(errorMessage, ce); - return null; - } - } - foundMessageList += ']'; + // the following methods may be overridden by junit tests - LOGGER.debug(TOPIC_TAG + topicName + ", returning DMaaP messages from " + consumerGroupPair.getLeft() - + " to " + messageMap.lastKey()); - synchronized (consumerGroupsMap) { - consumerGroupPair.setLeft(messageMap.lastKey()); - } - } + /** + * Makes a new timer pool. + * + * @return a new timer pool + */ + protected ScheduledExecutorService makeTimerPool() { + return Executors.newScheduledThreadPool(1); + } - return (foundMessageList.length() < 3 ? null : foundMessageList); + /** + * Makes a new topic. + * + * @param topicName topic name + * @return a new topic + */ + protected TopicData makeTopicData(String topicName) { + return new TopicData(topicName); } } diff --git a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/TopicData.java b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/TopicData.java new file mode 100644 index 000000000..2737f455b --- /dev/null +++ b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/TopicData.java @@ -0,0 +1,201 @@ +/*- + * ============LICENSE_START======================================================= + * ONAP Policy Models + * ================================================================================ + * Copyright (C) 2019 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.models.sim.dmaap.provider; + +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Function; +import org.onap.policy.common.utils.coder.Coder; +import org.onap.policy.common.utils.coder.CoderException; +import org.onap.policy.common.utils.coder.StandardCoder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Data associated with a topic. + * + * <p/> + * Note: for ease of implementation, this adds a topic when a consumer polls it rather + * than when a publisher writes to it. This is the opposite of how the real DMaaP works. + * As a result, this will never return a topic-not-found message to the consumer. + */ +public class TopicData { + private static final Logger logger = LoggerFactory.getLogger(TopicData.class); + + /** + * Name of the topic with which this data is associated. + */ + private final String topicName; + + /** + * Maps a consumer group name to its associated data. + */ + private final Map<String, ConsumerGroupData> group2data = new ConcurrentHashMap<>(); + + + /** + * Constructs the object. + * + * @param topicName name of the topic with which this object is associated + */ + public TopicData(String topicName) { + logger.info("Topic {}: added", topicName); + this.topicName = topicName; + } + + /** + * Removes idle consumers from the topic. This is typically called once during each + * sweep cycle. + */ + public void removeIdleConsumers() { + Iterator<Entry<String, ConsumerGroupData>> iter = group2data.entrySet().iterator(); + while (iter.hasNext()) { + Entry<String, ConsumerGroupData> ent = iter.next(); + if (ent.getValue().shouldRemove()) { + /* + * We want the minimum amount of time to elapse between invoking + * shouldRemove() and iter.remove(), thus all other statements (e.g., + * logging) should be done AFTER iter.remove(). + */ + iter.remove(); + + logger.info("Topic {}: removed consumer group: {}", topicName, ent.getKey()); + } + } + } + + /** + * Reads from a particular consumer group's queue. + * + * @param consumerGroup name of the consumer group from which to read + * @param maxRead maximum number of messages to read + * @param waitMs time to wait, in milliseconds, if the queue is currently empty + * @return a list of messages read from the queue, empty if no messages became + * available before the wait time elapsed + * @throws InterruptedException if this thread was interrupted while waiting for the + * first message + */ + public List<String> read(String consumerGroup, int maxRead, long waitMs) throws InterruptedException { + /* + * It's possible that this thread may spin several times while waiting for + * removeIdleConsumers() to complete its call to iter.remove(), thus we create + * this closure once, rather than each time through the loop. + */ + Function<String, ConsumerGroupData> maker = this::makeData; + + // loop until we get a readable list + List<String> result; + + // @formatter:off + + do { + result = group2data.computeIfAbsent(consumerGroup, maker).read(maxRead, waitMs); + } + while (result == ConsumerGroupData.UNREADABLE_LIST); + + // @formatter:on + + return result; + } + + /** + * Writes messages to the queues of every consumer group. + * + * @param messages messages to be written to the queues + * @return the number of messages enqueued + */ + public int write(List<Object> messages) { + List<String> list = convertMessagesToStrings(messages); + + /* + * We don't care if a consumer group is deleted from the map while we're adding + * messages to it, as those messages will simply be ignored (and discarded by the + * garbage collector). + */ + for (ConsumerGroupData data : group2data.values()) { + data.write(list); + } + + return list.size(); + } + + /** + * Converts a list of message objects to a list of message strings. If a message + * cannot be converted for some reason, then it is not added to the result list, thus + * the result list may be shorted than the original input list. + * + * @param messages objects to be converted + * @return a list of message strings + */ + protected List<String> convertMessagesToStrings(List<Object> messages) { + Coder coder = new StandardCoder(); + List<String> list = new ArrayList<>(messages.size()); + + for (Object msg : messages) { + String str = convertMessageToString(msg, coder); + if (str != null) { + list.add(str); + } + } + + return list; + } + + /** + * Converts a message object to a message string. + * + * @param message message to be converted + * @param coder used to encode the message as a string + * @return the message string, or {@code null} if it cannot be converted + */ + protected String convertMessageToString(Object message, Coder coder) { + if (message == null) { + return null; + } + + if (message instanceof String) { + return message.toString(); + } + + try { + return coder.encode(message); + } catch (CoderException e) { + logger.warn("cannot encode {}", message, e); + return null; + } + } + + // this may be overridden by junit tests + + /** + * Makes data for a consumer group. + * + * @param consumerGroup name of the consumer group to make + * @return new consumer group data + */ + protected ConsumerGroupData makeData(String consumerGroup) { + return new ConsumerGroupData(topicName, consumerGroup); + } +} diff --git a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/CambriaMessageBodyHandler.java b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/CambriaMessageBodyHandler.java index e269ac00b..dbd9422ef 100644 --- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/CambriaMessageBodyHandler.java +++ b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/CambriaMessageBodyHandler.java @@ -1,66 +1,178 @@ -/* - * ============LICENSE_START======================================================= ONAP - * ================================================================================ Copyright (C) 2019 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 +/*- + * ============LICENSE_START======================================================= + * ONAP Policy Models + * ================================================================================ + * Copyright (C) 2019 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 + * 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. + * 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.models.sim.dmaap.rest; import java.io.BufferedReader; +import java.io.EOFException; import java.io.IOException; import java.io.InputStream; import java.io.InputStreamReader; +import java.io.Reader; import java.lang.annotation.Annotation; import java.lang.reflect.Type; +import java.nio.charset.StandardCharsets; +import java.util.LinkedList; +import java.util.List; import javax.ws.rs.Consumes; -import javax.ws.rs.Produces; import javax.ws.rs.core.MediaType; import javax.ws.rs.core.MultivaluedMap; import javax.ws.rs.ext.MessageBodyReader; import javax.ws.rs.ext.Provider; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import org.apache.commons.io.IOUtils; /** - * Provider that serializes and de-serializes JSON via gson. + * Provider that decodes "application/cambria" messages. */ @Provider @Consumes(CambriaMessageBodyHandler.MEDIA_TYPE_APPLICATION_CAMBRIA) -@Produces(CambriaMessageBodyHandler.MEDIA_TYPE_APPLICATION_CAMBRIA) public class CambriaMessageBodyHandler implements MessageBodyReader<Object> { - // Media type for Cambria public static final String MEDIA_TYPE_APPLICATION_CAMBRIA = "application/cambria"; - public static final Logger logger = LoggerFactory.getLogger(CambriaMessageBodyHandler.class); + /** + * Maximum length of a message or partition. + */ + private static final int MAX_LEN = 10000000; + + /** + * Maximum digits in a length field. + */ + private static final int MAX_DIGITS = 10; @Override public boolean isReadable(Class<?> type, Type genericType, Annotation[] annotations, MediaType mediaType) { - return MEDIA_TYPE_APPLICATION_CAMBRIA.equals(mediaType.toString()); + return (mediaType != null && MEDIA_TYPE_APPLICATION_CAMBRIA.equals(mediaType.toString())); } @Override - public String readFrom(Class<Object> type, Type genericType, Annotation[] annotations, MediaType mediaType, - MultivaluedMap<String, String> httpHeaders, InputStream entityStream) - throws IOException { - - String cambriaString = ""; - try (BufferedReader bufferedReader = new BufferedReader( - new InputStreamReader(entityStream))) { - String line; - while ((line = bufferedReader.readLine()) != null) { - cambriaString += line; + public List<Object> readFrom(Class<Object> type, Type genericType, Annotation[] annotations, MediaType mediaType, + MultivaluedMap<String, String> httpHeaders, InputStream entityStream) throws IOException { + + try (BufferedReader bufferedReader = + new BufferedReader(new InputStreamReader(entityStream, StandardCharsets.UTF_8))) { + List<Object> messages = new LinkedList<>(); + String msg; + while ((msg = readMessage(bufferedReader)) != null) { + messages.add(msg); + } + + return messages; + } + } + + /** + * Reads a message. + * + * @param reader source from which to read + * @return the message that was read, or {@code null} if there are no more messages + * @throws IOException if an error occurs + */ + private String readMessage(Reader reader) throws IOException { + if (!skipWhitespace(reader)) { + return null; + } + + int partlen = readLength(reader); + if (partlen > MAX_LEN) { + throw new IOException("invalid partition length"); + } + + int msglen = readLength(reader); + if (msglen > MAX_LEN) { + throw new IOException("invalid message length"); + } + + // skip over the partition + reader.skip(partlen); + + return readString(reader, msglen); + } + + /** + * Skips whitespace. + * + * @param reader source from which to read + * @return {@code true} if there is another character after the whitespace, + * {@code false} if the end of the stream has been reached + * @throws IOException if an error occurs + */ + private boolean skipWhitespace(Reader reader) throws IOException { + int chr; + + do { + reader.mark(1); + if ((chr = reader.read()) < 0) { + return false; + } + } + while (Character.isWhitespace(chr)); + + // push the last character back onto the reader + reader.reset(); + + return true; + } + + /** + * Reads a length field, which is a number followed by ".". + * + * @param reader source from which to read + * @return the length, or -1 if EOF has been reached + * @throws IOException if an error occurs + */ + private int readLength(Reader reader) throws IOException { + StringBuilder bldr = new StringBuilder(MAX_DIGITS); + + int chr; + for (int x = 0; x < MAX_DIGITS; ++x) { + if ((chr = reader.read()) < 0) { + throw new EOFException("missing '.' in 'length' field"); } - return cambriaString.substring(cambriaString.indexOf('{'), cambriaString.length()); + if (chr == '.') { + String text = bldr.toString().trim(); + return (text.isEmpty() ? 0 : Integer.valueOf(text)); + } + + if (!Character.isDigit(chr)) { + throw new IOException("invalid character in 'length' field"); + } + + bldr.append((char) chr); } + + throw new IOException("too many digits in 'length' field"); + } + + /** + * Reads a string. + * + * @param reader source from which to read + * @param len length of the string (i.e., number of characters to read) + * @return the string that was read + * @throws IOException if an error occurs + */ + private String readString(Reader reader, int len) throws IOException { + char[] buf = new char[len]; + IOUtils.readFully(reader, buf); + + return new String(buf); } } diff --git a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/DmaapSimRestControllerV1.java b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/DmaapSimRestControllerV1.java index e3fdd4884..8339d0e64 100644 --- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/DmaapSimRestControllerV1.java +++ b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/DmaapSimRestControllerV1.java @@ -1,6 +1,7 @@ /*- * ============LICENSE_START======================================================= * Copyright (C) 2019 Nordix Foundation. + * Modifications Copyright (C) 2019 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. @@ -20,24 +21,24 @@ package org.onap.policy.models.sim.dmaap.rest; -import io.swagger.annotations.ApiOperation; -import io.swagger.annotations.ApiResponse; -import io.swagger.annotations.ApiResponses; -import io.swagger.annotations.Authorization; +import javax.ws.rs.Consumes; +import javax.ws.rs.DefaultValue; import javax.ws.rs.GET; import javax.ws.rs.POST; import javax.ws.rs.Path; import javax.ws.rs.PathParam; +import javax.ws.rs.Produces; import javax.ws.rs.QueryParam; import javax.ws.rs.core.Response; - import org.onap.policy.models.sim.dmaap.provider.DmaapSimProvider; /** * Class to provide REST endpoints for DMaaP simulator component statistics. */ @Path("/events") +@Produces(DmaapSimRestControllerV1.MEDIA_TYPE_APPLICATION_JSON) public class DmaapSimRestControllerV1 extends BaseRestControllerV1 { + public static final String MEDIA_TYPE_APPLICATION_JSON = "application/json"; /** * Get a DMaaP message. @@ -45,72 +46,33 @@ public class DmaapSimRestControllerV1 extends BaseRestControllerV1 { * @param topicName topic to get message from * @param consumerGroup consumer group that is getting the message * @param consumerId consumer ID that is getting the message - * @param timeout timeout for the message + * @param timeoutMs timeout for the message * @return the message */ @GET @Path("{topicName}/{consumerGroup}/{consumerId}") - // @formatter:off - @ApiOperation( - value = "Get a DMaaP event on a topic", - notes = "Returns an event on a DMaaP topic", - response = Object.class, - authorizations = - @Authorization(value = AUTHORIZATION_TYPE) - ) - @ApiResponses( - value = { - @ApiResponse( - code = AUTHENTICATION_ERROR_CODE, - message = AUTHENTICATION_ERROR_MESSAGE), - @ApiResponse( - code = AUTHORIZATION_ERROR_CODE, - message = AUTHORIZATION_ERROR_MESSAGE), - @ApiResponse( - code = SERVER_ERROR_CODE, - message = SERVER_ERROR_MESSAGE) - } - ) - // @formatter:on - public Response getDmaaapMessage(@PathParam("topicName") final String topicName, - @PathParam("consumerGroup") final String consumerGroup, @PathParam("consumerId") final String consumerId, - @QueryParam("timeout") final int timeout) { + public Response getDmaapMessage(@PathParam("topicName") final String topicName, + @PathParam("consumerGroup") final String consumerGroup, + @PathParam("consumerId") final String consumerId, + @QueryParam("limit") @DefaultValue("1") final int limit, + @QueryParam("timeout") @DefaultValue("15000") final long timeoutMs) { - return new DmaapSimProvider().processDmaapMessageGet(topicName, consumerGroup, consumerId, timeout); + return DmaapSimProvider.getInstance().processDmaapMessageGet(topicName, consumerGroup, consumerId, limit, + timeoutMs); } /** * Post a DMaaP message. * - * @param topicName topic to get message from415 + * @param topicName topic to get message from * @return the response to the post */ @POST @Path("{topicName}") - // @formatter:off - @ApiOperation( - value = "Post a DMaaP event on a topic", - notes = "Returns an event on a DMaaP topic", - response = Response.class, - authorizations = - @Authorization(value = AUTHORIZATION_TYPE) - ) - @ApiResponses( - value = { - @ApiResponse( - code = AUTHENTICATION_ERROR_CODE, - message = AUTHENTICATION_ERROR_MESSAGE), - @ApiResponse( - code = AUTHORIZATION_ERROR_CODE, - message = AUTHORIZATION_ERROR_MESSAGE), - @ApiResponse( - code = SERVER_ERROR_CODE, - message = SERVER_ERROR_MESSAGE) - } - ) - // @formatter:on - public Response postDmaaapMessage(@PathParam("topicName") final String topicName, final Object dmaapMessage) { + @Consumes(value = {CambriaMessageBodyHandler.MEDIA_TYPE_APPLICATION_CAMBRIA, + TextMessageBodyHandler.MEDIA_TYPE_TEXT_PLAIN, MEDIA_TYPE_APPLICATION_JSON}) + public Response postDmaapMessage(@PathParam("topicName") final String topicName, final Object dmaapMessage) { - return new DmaapSimProvider().processDmaapMessagePut(topicName, dmaapMessage); + return DmaapSimProvider.getInstance().processDmaapMessagePut(topicName, dmaapMessage); } } diff --git a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/DmaapSimRestServer.java b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/DmaapSimRestServer.java index 28de42c21..b05a0fe1a 100644 --- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/DmaapSimRestServer.java +++ b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/DmaapSimRestServer.java @@ -21,28 +21,21 @@ package org.onap.policy.models.sim.dmaap.rest; -import java.util.ArrayList; import java.util.List; import java.util.Properties; - -import org.onap.policy.common.capabilities.Startable; import org.onap.policy.common.endpoints.http.server.HttpServletServer; import org.onap.policy.common.endpoints.http.server.HttpServletServerFactoryInstance; import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties; +import org.onap.policy.common.gson.GsonMessageBodyHandler; +import org.onap.policy.common.utils.services.ServiceManagerContainer; import org.onap.policy.models.sim.dmaap.parameters.RestServerParameters; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * Class to manage life cycle of DMaaP Simulator rest server. */ -public class DmaapSimRestServer implements Startable { - - private static final Logger LOGGER = LoggerFactory.getLogger(DmaapSimRestServer.class); +public class DmaapSimRestServer extends ServiceManagerContainer { - private List<HttpServletServer> servers = new ArrayList<>(); - - private RestServerParameters restServerParameters; + private final List<HttpServletServer> servers; /** * Constructor for instantiating DmaapSimRestServer. @@ -50,91 +43,40 @@ public class DmaapSimRestServer implements Startable { * @param restServerParameters the rest server parameters */ public DmaapSimRestServer(final RestServerParameters restServerParameters) { - this.restServerParameters = restServerParameters; - } + this.servers = HttpServletServerFactoryInstance.getServerFactory() + .build(getServerProperties(restServerParameters)); - /** - * {@inheritDoc}. - */ - @Override - public boolean start() { - try { - servers = HttpServletServerFactoryInstance.getServerFactory().build(getServerProperties()); - for (final HttpServletServer server : servers) { - server.start(); - } - } catch (final Exception exp) { - LOGGER.error("Failed to start DMaaP simulator http server", exp); - return false; + for (HttpServletServer server : this.servers) { + addAction("REST " + server.getName(), server::start, server::stop); } - return true; } /** - * Creates the server properties object using restServerParameters. + * Creates a set of properties, suitable for building a REST server, from the + * parameters. * - * @return the properties object + * @param restServerParameters parameters from which to build the properties + * @return a set of properties representing the given parameters */ - private Properties getServerProperties() { + public static Properties getServerProperties(RestServerParameters restServerParameters) { final Properties props = new Properties(); props.setProperty(PolicyEndPointProperties.PROPERTY_HTTP_SERVER_SERVICES, restServerParameters.getName()); final String svcpfx = - PolicyEndPointProperties.PROPERTY_HTTP_SERVER_SERVICES + "." + restServerParameters.getName(); + PolicyEndPointProperties.PROPERTY_HTTP_SERVER_SERVICES + "." + restServerParameters.getName(); props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_HOST_SUFFIX, restServerParameters.getHost()); props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_PORT_SUFFIX, - Integer.toString(restServerParameters.getPort())); + Integer.toString(restServerParameters.getPort())); props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_REST_CLASSES_SUFFIX, - String.join(",", DmaapSimRestControllerV1.class.getName())); + DmaapSimRestControllerV1.class.getName()); props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX, "false"); - props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_SWAGGER_SUFFIX, "true"); + props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_SWAGGER_SUFFIX, "false"); + props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_SERIALIZATION_PROVIDER, - CambriaMessageBodyHandler.class.getName() + "," + JsonMessageBodyHandler.class.getName()); + String.join(",", CambriaMessageBodyHandler.class.getName(), + GsonMessageBodyHandler.class.getName(), + TextMessageBodyHandler.class.getName())); return props; } - - /** - * {@inheritDoc}. - */ - @Override - public boolean stop() { - for (final HttpServletServer server : servers) { - try { - server.stop(); - } catch (final Exception exp) { - LOGGER.error("Failed to stop DMaaP simulator http server", exp); - } - } - return true; - } - - /** - * {@inheritDoc}. - */ - @Override - public void shutdown() { - stop(); - } - - /** - * {@inheritDoc}. - */ - @Override - public boolean isAlive() { - return !servers.isEmpty(); - } - - /** - * {@inheritDoc}. - */ - @Override - public String toString() { - final StringBuilder builder = new StringBuilder(); - builder.append("DmaapSimRestServer [servers="); - builder.append(servers); - builder.append("]"); - return builder.toString(); - } - } diff --git a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/JsonMessageBodyHandler.java b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/JsonMessageBodyHandler.java deleted file mode 100644 index a3eebda00..000000000 --- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/JsonMessageBodyHandler.java +++ /dev/null @@ -1,63 +0,0 @@ -/* - * ============LICENSE_START======================================================= ONAP - * ================================================================================ Copyright (C) 2019 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.models.sim.dmaap.rest; - -import java.io.BufferedReader; -import java.io.IOException; -import java.io.InputStream; -import java.io.InputStreamReader; -import java.lang.annotation.Annotation; -import java.lang.reflect.Type; -import javax.ws.rs.Consumes; -import javax.ws.rs.Produces; -import javax.ws.rs.core.MediaType; -import javax.ws.rs.core.MultivaluedMap; -import javax.ws.rs.ext.MessageBodyReader; -import javax.ws.rs.ext.Provider; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * Provider that serializes and de-serializes JSON via gson. - */ -@Provider -@Consumes(MediaType.APPLICATION_JSON) -@Produces(MediaType.APPLICATION_JSON) -public class JsonMessageBodyHandler implements MessageBodyReader<Object> { - public static final Logger logger = LoggerFactory.getLogger(JsonMessageBodyHandler.class); - - @Override - public boolean isReadable(Class<?> type, Type genericType, Annotation[] annotations, MediaType mediaType) { - return MediaType.APPLICATION_JSON.equals(mediaType.toString()); - } - - @Override - public String readFrom(Class<Object> type, Type genericType, Annotation[] annotations, MediaType mediaType, - MultivaluedMap<String, String> httpHeaders, InputStream entityStream) - throws IOException { - - String jsonString = ""; - try (BufferedReader bufferedReader = new BufferedReader( - new InputStreamReader(entityStream))) { - String line; - while ((line = bufferedReader.readLine()) != null) { - jsonString += line; - } - - return jsonString; - } - } -} diff --git a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/TextMessageBodyHandler.java b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/TextMessageBodyHandler.java new file mode 100644 index 000000000..3c903c82b --- /dev/null +++ b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/TextMessageBodyHandler.java @@ -0,0 +1,66 @@ +/*- + * ============LICENSE_START======================================================= + * ONAP Policy Models + * ================================================================================ + * Copyright (C) 2019 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.models.sim.dmaap.rest; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.lang.annotation.Annotation; +import java.lang.reflect.Type; +import java.nio.charset.StandardCharsets; +import java.util.LinkedList; +import java.util.List; +import javax.ws.rs.Consumes; +import javax.ws.rs.core.MediaType; +import javax.ws.rs.core.MultivaluedMap; +import javax.ws.rs.ext.MessageBodyReader; +import javax.ws.rs.ext.Provider; + +/** + * Provider that decodes "text/plain" messages. + */ +@Provider +@Consumes(TextMessageBodyHandler.MEDIA_TYPE_TEXT_PLAIN) +public class TextMessageBodyHandler implements MessageBodyReader<Object> { + public static final String MEDIA_TYPE_TEXT_PLAIN = "text/plain"; + + @Override + public boolean isReadable(Class<?> type, Type genericType, Annotation[] annotations, MediaType mediaType) { + return (mediaType != null && MEDIA_TYPE_TEXT_PLAIN.equals(mediaType.toString())); + } + + @Override + public List<Object> readFrom(Class<Object> type, Type genericType, Annotation[] annotations, MediaType mediaType, + MultivaluedMap<String, String> httpHeaders, InputStream entityStream) throws IOException { + + try (BufferedReader bufferedReader = + new BufferedReader(new InputStreamReader(entityStream, StandardCharsets.UTF_8))) { + List<Object> messages = new LinkedList<>(); + String msg; + while ((msg = bufferedReader.readLine()) != null) { + messages.add(msg); + } + + return messages; + } + } +} diff --git a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/startstop/DmaapSimActivator.java b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/startstop/DmaapSimActivator.java index 899c0e081..b9e0efaaf 100644 --- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/startstop/DmaapSimActivator.java +++ b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/startstop/DmaapSimActivator.java @@ -1,6 +1,7 @@ /*- * ============LICENSE_START======================================================= * Copyright (C) 2019 Nordix Foundation. + * Modifications Copyright (C) 2019 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. @@ -20,19 +21,15 @@ package org.onap.policy.models.sim.dmaap.startstop; -import org.onap.policy.common.parameters.ParameterService; import org.onap.policy.common.utils.services.ServiceManagerContainer; import org.onap.policy.models.sim.dmaap.parameters.DmaapSimParameterGroup; +import org.onap.policy.models.sim.dmaap.provider.DmaapSimProvider; import org.onap.policy.models.sim.dmaap.rest.DmaapSimRestServer; /** * This class activates the DMaaP simulator as a complete service. */ public class DmaapSimActivator extends ServiceManagerContainer { - /** - * The DMaaP simulator REST API server. - */ - private DmaapSimRestServer restServer; /** * Instantiate the activator for the DMaaP simulator as a complete service. @@ -42,19 +39,11 @@ public class DmaapSimActivator extends ServiceManagerContainer { public DmaapSimActivator(final DmaapSimParameterGroup dmaapSimParameterGroup) { super("DMaaP Simulator"); - // @formatter:off - addAction("DMaaP Simulator parameters", - () -> ParameterService.register(dmaapSimParameterGroup), - () -> ParameterService.deregister(dmaapSimParameterGroup.getName())); - - addAction("Create REST server", - () -> restServer = new DmaapSimRestServer(dmaapSimParameterGroup.getRestServerParameters()), - () -> restServer = null - ); + DmaapSimProvider provider = new DmaapSimProvider(dmaapSimParameterGroup); + DmaapSimProvider.setInstance(provider); + addAction("Sim Provider", provider::start, provider::stop); - addAction("REST server", - () -> restServer.start(), - () -> restServer.stop()); - // @formatter:on + DmaapSimRestServer restServer = new DmaapSimRestServer(dmaapSimParameterGroup.getRestServerParameters()); + addAction("REST server", restServer::start, restServer::stop); } } diff --git a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/startstop/DmaapSimCommandLineArguments.java b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/startstop/DmaapSimCommandLineArguments.java index cf559f712..724c3dc35 100644 --- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/startstop/DmaapSimCommandLineArguments.java +++ b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/startstop/DmaapSimCommandLineArguments.java @@ -26,13 +26,15 @@ import java.io.PrintWriter; import java.io.StringWriter; import java.net.URL; import java.util.Arrays; - +import lombok.Getter; +import lombok.Setter; import org.apache.commons.cli.CommandLine; import org.apache.commons.cli.DefaultParser; import org.apache.commons.cli.HelpFormatter; import org.apache.commons.cli.Option; import org.apache.commons.cli.Options; import org.apache.commons.cli.ParseException; +import org.apache.commons.lang3.StringUtils; import org.onap.policy.common.utils.resources.ResourceUtils; import org.onap.policy.models.sim.dmaap.DmaapSimException; import org.onap.policy.models.sim.dmaap.DmaapSimRuntimeException; @@ -46,6 +48,9 @@ public class DmaapSimCommandLineArguments { private static final int HELP_LINE_LENGTH = 120; private final Options options; + + @Getter + @Setter private String configurationFilePath = null; /** @@ -145,7 +150,7 @@ public class DmaapSimCommandLineArguments { * @throws DmaapSimException on command argument validation errors */ public void validate() throws DmaapSimException { - validateReadableFile("DMaaP simulator configuration", configurationFilePath); + validateFileExists("DMaaP simulator configuration", configurationFilePath); } /** @@ -175,15 +180,6 @@ public class DmaapSimCommandLineArguments { } /** - * Gets the configuration file path. - * - * @return the configuration file path - */ - public String getConfigurationFilePath() { - return configurationFilePath; - } - - /** * Gets the full expanded configuration file path. * * @return the configuration file path @@ -193,16 +189,6 @@ public class DmaapSimCommandLineArguments { } /** - * Sets the configuration file path. - * - * @param configurationFilePath the configuration file path - */ - public void setConfigurationFilePath(final String configurationFilePath) { - this.configurationFilePath = configurationFilePath; - - } - - /** * Check set configuration file path. * * @return true, if check set configuration file path @@ -212,14 +198,14 @@ public class DmaapSimCommandLineArguments { } /** - * Validate readable file. + * Validate file exists. * * @param fileTag the file tag * @param fileName the file name * @throws DmaapSimException on the file name passed as a parameter */ - private void validateReadableFile(final String fileTag, final String fileName) throws DmaapSimException { - if (fileName == null || fileName.length() == 0) { + private void validateFileExists(final String fileTag, final String fileName) throws DmaapSimException { + if (StringUtils.isBlank(fileName)) { throw new DmaapSimException(fileTag + " file was not specified as an argument"); } @@ -233,11 +219,5 @@ public class DmaapSimCommandLineArguments { if (!theFile.exists()) { throw new DmaapSimException(fileTag + FILE_MESSAGE_PREAMBLE + fileName + "\" does not exist"); } - if (!theFile.isFile()) { - throw new DmaapSimException(fileTag + FILE_MESSAGE_PREAMBLE + fileName + "\" is not a normal file"); - } - if (!theFile.canRead()) { - throw new DmaapSimException(fileTag + FILE_MESSAGE_PREAMBLE + fileName + "\" is ureadable"); - } } } diff --git a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/startstop/Main.java b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/startstop/Main.java index 878d008a8..7b4f41b07 100644 --- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/startstop/Main.java +++ b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/startstop/Main.java @@ -1,6 +1,7 @@ /*- * ============LICENSE_START======================================================= * Copyright (C) 2019 Nordix Foundation. + * Modifications Copyright (C) 2019 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. @@ -21,9 +22,6 @@ package org.onap.policy.models.sim.dmaap.startstop; import java.util.Arrays; - -import org.onap.policy.common.utils.services.Registry; -import org.onap.policy.models.sim.dmaap.DmaapSimConstants; import org.onap.policy.models.sim.dmaap.DmaapSimException; import org.onap.policy.models.sim.dmaap.parameters.DmaapSimParameterGroup; import org.onap.policy.models.sim.dmaap.parameters.DmaapSimParameterHandler; @@ -75,14 +73,12 @@ public class Main { // Now, create the activator for the DMaaP Simulator service activator = new DmaapSimActivator(parameterGroup); - Registry.register(DmaapSimConstants.REG_DMAAP_SIM_ACTIVATOR, activator); // Start the activator try { activator.start(); } catch (final RuntimeException e) { LOGGER.error("start of DMaaP simulator service failed, used parameters are {}", Arrays.toString(args), e); - Registry.unregister(DmaapSimConstants.REG_DMAAP_SIM_ACTIVATOR); return; } @@ -110,7 +106,7 @@ public class Main { parameterGroup = null; // clear the DMaaP simulator activator - if (activator != null) { + if (activator != null && activator.isAlive()) { activator.stop(); } } @@ -128,8 +124,9 @@ public class Main { public void run() { try { // Shutdown the DMaaP simulator service and wait for everything to stop - activator.stop(); - } catch (final RuntimeException e) { + shutdown(); + + } catch (final RuntimeException | DmaapSimException e) { LOGGER.warn("error occured during shut down of the DMaaP simulator service", e); } } diff --git a/models-sim/models-sim-dmaap/src/main/resources/etc/DefaultConfig.json b/models-sim/models-sim-dmaap/src/main/resources/etc/DefaultConfig.json index dd2477a24..e936eb034 100644 --- a/models-sim/models-sim-dmaap/src/main/resources/etc/DefaultConfig.json +++ b/models-sim/models-sim-dmaap/src/main/resources/etc/DefaultConfig.json @@ -1,5 +1,6 @@ { "name": "DMaapSim", + "topicSweepSec": 900, "restServerParameters": { "host": "0.0.0.0", "port": 3904 diff --git a/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/DmaapSimXxxExceptionTest.java b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/DmaapSimXxxExceptionTest.java new file mode 100644 index 000000000..4e37a5e36 --- /dev/null +++ b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/DmaapSimXxxExceptionTest.java @@ -0,0 +1,39 @@ +/*- + * ============LICENSE_START======================================================= + * Copyright (C) 2019 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.models.sim.dmaap; + +import static org.junit.Assert.assertEquals; + +import org.junit.Test; +import org.onap.policy.common.utils.test.ExceptionsTester; +import org.onap.policy.models.sim.dmaap.DmaapSimException; +import org.onap.policy.models.sim.dmaap.DmaapSimRuntimeException; + +public class DmaapSimXxxExceptionTest { + + @Test + public void testDmaapSimException() { + assertEquals(3, new ExceptionsTester().test(DmaapSimException.class)); + } + + @Test + public void testDmaapSimRuntimeException() { + assertEquals(3, new ExceptionsTester().test(DmaapSimRuntimeException.class)); + } +} diff --git a/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/provider/ConsumerGroupDataTest.java b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/provider/ConsumerGroupDataTest.java new file mode 100644 index 000000000..4513ffb82 --- /dev/null +++ b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/provider/ConsumerGroupDataTest.java @@ -0,0 +1,305 @@ +/*- + * ============LICENSE_START======================================================= + * ONAP Policy Models + * ================================================================================ + * Copyright (C) 2019 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.models.sim.dmaap.provider; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +public class ConsumerGroupDataTest { + private static final int WAIT_MS = 5000; + private static final int MIN_WAIT_MS = WAIT_MS / 2; + private static final String MY_TOPIC = "my-topic"; + private static final String MY_CONSUMER = "my-consumer"; + private static final String MSG1 = "hello"; + private static final String MSG2 = "there"; + private static final String MSG3 = "world"; + private static final int MAX_THREADS = 30; + + private MyData data; + private MyReader thread; + private List<MyReader> threads; + + /** + * Sets up. + */ + @Before + public void setUp() { + data = new MyData(); + thread = null; + threads = new ArrayList<>(MAX_THREADS); + } + + /** + * Stops any running thread. + */ + @After + public void tearDown() { + for (MyReader thr : threads) { + thr.interrupt(); + } + + for (MyReader thr : threads) { + thr.await(); + } + } + + @Test + public void testShouldRemove() throws InterruptedException { + assertFalse(data.shouldRemove()); + assertTrue(data.shouldRemove()); + + data = new MyData(); + + // start a reader thread and wait for it to poll its queue + startReader(0, 10); + assertTrue(data.await()); + + assertFalse(data.shouldRemove()); + } + + @Test + public void testRead() { + data.enqueue(MSG1, MSG2, MSG3, MSG1, MSG2, MSG3); + + // this reader only wants one + startReader(1, 1); + assertTrue(thread.await()); + assertEquals("[hello]", thread.result.toString()); + + // this reader wants three + startReader(3, 1); + assertTrue(thread.await()); + assertEquals("[there, world, hello]", thread.result.toString()); + + // this reader wants three, but will only get two + startReader(3, 1); + assertTrue(thread.await()); + assertEquals("[there, world]", thread.result.toString()); + } + + @Test + public void testRead_Idle() throws InterruptedException { + // force it to idle + data.shouldRemove(); + data.shouldRemove(); + + long tbeg = System.currentTimeMillis(); + assertSame(ConsumerGroupData.UNREADABLE_LIST, data.read(1, WAIT_MS)); + + // should not have waited + assertTrue(System.currentTimeMillis() < tbeg + MIN_WAIT_MS); + } + + @Test + public void testRead_NegativeCount() throws InterruptedException { + data.enqueue(MSG1, MSG2); + startReader(-1, 3); + assertTrue(data.await()); + + // wait time should be unaffected + assertEquals(3L, data.waitMs2); + + assertTrue(thread.await()); + + // should only return one message + assertEquals("[hello]", thread.result.toString()); + } + + @Test + public void testRead_NegativeWait() throws InterruptedException { + data.enqueue(MSG1, MSG2, MSG3); + startReader(2, -3); + assertTrue(data.await()); + + assertEquals(0L, data.waitMs2); + + assertTrue(thread.await()); + + // should return two messages, as requested + assertEquals("[hello, there]", thread.result.toString()); + } + + @Test + public void testRead_NoMessages() throws InterruptedException { + startReader(0, 0); + assertTrue(data.await()); + + assertTrue(thread.await()); + assertTrue(thread.result.isEmpty()); + } + + @Test + public void testRead_MultiThreaded() { + // queue up a bunch of messages + final int expected = MAX_THREADS * 3; + for (int x = 0; x < expected; ++x) { + data.enqueue(MSG1); + } + + for (int x = 0; x < MAX_THREADS; ++x) { + startReader(4, 1); + } + + int actual = 0; + for (MyReader thr : threads) { + thr.await(); + actual += thr.result.size(); + } + + assertEquals(expected, actual); + } + + + /** + * Starts a reader thread. + * + * @param limit number of messages to read at one time + * @param waitMs wait time, in milliseconds + */ + private void startReader(int limit, long waitMs) { + thread = new MyReader(limit, waitMs); + + thread.setDaemon(true); + thread.start(); + + threads.add(thread); + } + + + private class MyData extends ConsumerGroupData { + + /** + * Decremented when {@link #getNextMessage(long)} is invoked. + */ + private final CountDownLatch latch = new CountDownLatch(1); + + /** + * Messages to be added to the queue when {@link #getNextMessage(long)} is + * invoked. + */ + private final List<String> messages = new ArrayList<>(); + + /** + * Value passed to {@link #getNextMessage(long)}. + */ + private volatile long waitMs2 = -1; + + /** + * Constructs the object. + */ + public MyData() { + super(MY_TOPIC, MY_CONSUMER); + } + + /** + * Arranges for messages to be injected into the queue the next time + * {@link #getNextMessage(long)} is invoked. + * + * @param messages the messages to be injected + */ + public void enqueue(String... messages) { + this.messages.addAll(Arrays.asList(messages)); + } + + @Override + protected String getNextMessage(long waitMs) throws InterruptedException { + waitMs2 = waitMs; + + latch.countDown(); + + synchronized (messages) { + write(messages); + messages.clear(); + } + + return super.getNextMessage(waitMs); + } + + /** + * Waits for {@link #getNextMessage(long)} to be invoked. + * + * @return {@code true} if {@link #getNextMessage(long)} was invoked, + * {@code false} if the timer expired first + * @throws InterruptedException if the current thread is interrupted while waiting + */ + public boolean await() throws InterruptedException { + return latch.await(WAIT_MS, TimeUnit.MILLISECONDS); + } + } + + /** + * Thread that will invoke the consumer group's read() method one time. + */ + private class MyReader extends Thread { + private final ConsumerGroupData group = data; + private final int limit; + private final long waitMs; + + /** + * Result returned by the read() method. + */ + private List<String> result = Collections.emptyList(); + + public MyReader(int limit, long waitMs) { + this.limit = limit; + this.waitMs = waitMs; + } + + @Override + public void run() { + try { + result = group.read(limit, waitMs); + + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + /** + * Waits for the thread to complete. + * + * @return {@code true} if the thread completed, {@code false} if the thread is + * still running + */ + public boolean await() { + try { + this.join(WAIT_MS); + + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + + return !this.isAlive(); + } + } +} diff --git a/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/provider/DmaapSimProviderTest.java b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/provider/DmaapSimProviderTest.java new file mode 100644 index 000000000..f8c141614 --- /dev/null +++ b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/provider/DmaapSimProviderTest.java @@ -0,0 +1,287 @@ +/*- + * ============LICENSE_START======================================================= + * Copyright (C) 2019 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.models.sim.dmaap.provider; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyInt; +import static org.mockito.Matchers.anyLong; +import static org.mockito.Matchers.eq; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import javax.ws.rs.core.Response; +import javax.ws.rs.core.Response.Status; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Captor; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; +import org.onap.policy.common.utils.coder.CoderException; +import org.onap.policy.common.utils.coder.StandardCoder; +import org.onap.policy.common.utils.coder.StandardCoderObject; +import org.onap.policy.models.sim.dmaap.parameters.DmaapSimParameterGroup; + +public class DmaapSimProviderTest { + private static final String EXPECTED_EXCEPTION = "expected exception"; + private static final long SWEEP_SEC = 10L; + private static final String TOPIC1 = "topic-A"; + private static final String TOPIC2 = "topic-B"; + private static final String CONSUMER1 = "consumer-X"; + private static final String CONSUMER_ID1 = "id1"; + + private MyProvider prov; + + @Mock + private DmaapSimParameterGroup params; + + @Mock + private ScheduledExecutorService timer; + + @Mock + private TopicData data1; + + @Mock + private TopicData data2; + + @Captor + private ArgumentCaptor<List<Object>> listCaptor; + + /** + * Sets up. + */ + @Before + public void setUp() { + MockitoAnnotations.initMocks(this); + + when(params.getTopicSweepSec()).thenReturn(SWEEP_SEC); + + prov = new MyProvider(params); + } + + /** + * Shuts down the provider, if it's running. + */ + @After + public void tearDown() { + if (prov.isAlive()) { + prov.shutdown(); + } + } + + /** + * Verifies that the constructor adds all of the expected actions to the service + * manager container. + */ + @Test + public void testDmaapSimProvider() { + prov.start(); + verify(timer).scheduleWithFixedDelay(any(), eq(SWEEP_SEC), eq(SWEEP_SEC), eq(TimeUnit.SECONDS)); + + prov.stop(); + verify(timer).shutdown(); + } + + @Test + public void testProcessDmaapMessagePut_List() throws CoderException { + prov = spy(new MyProvider(params)); + + when(data1.write(any())).thenReturn(2); + + // force topics to exist + prov.processDmaapMessageGet(TOPIC1, CONSUMER1, CONSUMER_ID1, 1, 0); + prov.processDmaapMessageGet(TOPIC2, CONSUMER1, CONSUMER_ID1, 1, 0); + + List<Object> lst = Arrays.asList("hello", "world"); + Response resp = prov.processDmaapMessagePut(TOPIC1, lst); + assertEquals(Status.OK.getStatusCode(), resp.getStatus()); + StandardCoderObject sco = new StandardCoder().decode(resp.getEntity().toString(), StandardCoderObject.class); + assertEquals("2", sco.getString("count")); + + List<Object> lst2 = Arrays.asList("helloB", "worldB"); + prov.processDmaapMessagePut(TOPIC1, lst2); + prov.processDmaapMessagePut(TOPIC2, lst2); + + // should only invoke this once for each topic + verify(prov).makeTopicData(TOPIC1); + verify(prov).makeTopicData(TOPIC2); + + // should process all writes + verify(data1).write(lst); + verify(data1).write(lst2); + + verify(data2).write(lst2); + } + + @Test + public void testProcessDmaapMessagePut_Single() throws CoderException { + prov = spy(new MyProvider(params)); + + // force topics to exist + prov.processDmaapMessageGet(TOPIC1, CONSUMER1, CONSUMER_ID1, 1, 0); + prov.processDmaapMessageGet(TOPIC2, CONSUMER1, CONSUMER_ID1, 1, 0); + + final String value1 = "abc"; + Response resp = prov.processDmaapMessagePut(TOPIC1, value1); + assertEquals(Status.OK.getStatusCode(), resp.getStatus()); + + // ensure that the response can be decoded + new StandardCoder().decode(resp.getEntity().toString(), StandardCoderObject.class); + + final String value2 = "def"; + prov.processDmaapMessagePut(TOPIC1, value2); + prov.processDmaapMessagePut(TOPIC2, value2); + + // should only invoke this once for each topic + verify(prov).makeTopicData(TOPIC1); + verify(prov).makeTopicData(TOPIC2); + + // should process all writes as singleton lists + listCaptor.getAllValues().clear(); + verify(data1, times(2)).write(listCaptor.capture()); + assertEquals(Collections.singletonList(value1), listCaptor.getAllValues().get(0)); + assertEquals(Collections.singletonList(value2), listCaptor.getAllValues().get(1)); + + listCaptor.getAllValues().clear(); + verify(data2).write(listCaptor.capture()); + assertEquals(Collections.singletonList(value2), listCaptor.getAllValues().get(0)); + } + + @Test + public void testProcessDmaapMessageGet() throws InterruptedException { + List<String> msgs = Arrays.asList("400", "500"); + when(data1.read(any(), anyInt(), anyLong())).thenReturn(msgs); + + Response resp = prov.processDmaapMessageGet(TOPIC1, CONSUMER1, CONSUMER_ID1, 4, 400L); + assertEquals(Status.OK.getStatusCode(), resp.getStatus()); + assertEquals(msgs.toString(), resp.getEntity().toString()); + } + + @Test + public void testProcessDmaapMessageGet_Timeout() throws InterruptedException { + when(data1.read(any(), anyInt(), anyLong())).thenReturn(Collections.emptyList()); + + Response resp = prov.processDmaapMessageGet(TOPIC1, CONSUMER1, CONSUMER_ID1, 3, 300L); + assertEquals(Status.REQUEST_TIMEOUT.getStatusCode(), resp.getStatus()); + assertEquals("[]", resp.getEntity().toString()); + } + + @Test + public void testProcessDmaapMessageGet_Ex() throws InterruptedException { + BlockingQueue<Response> respQueue = new LinkedBlockingQueue<>(); + + // put in a background thread so it doesn't interrupt the tester thread + new Thread(() -> { + try { + when(data1.read(any(), anyInt(), anyLong())).thenThrow(new InterruptedException(EXPECTED_EXCEPTION)); + respQueue.offer(prov.processDmaapMessageGet(TOPIC1, CONSUMER1, CONSUMER_ID1, 3, 300L)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }).start(); + + Response resp = respQueue.poll(3, TimeUnit.SECONDS); + assertNotNull(resp); + + assertEquals(Status.GONE.getStatusCode(), resp.getStatus()); + assertEquals("[]", resp.getEntity().toString()); + } + + @Test + public void testSweepTopicTaskRun() { + prov.start(); + prov.processDmaapMessageGet(TOPIC1, CONSUMER1, CONSUMER_ID1, 0, 0); + prov.processDmaapMessageGet(TOPIC2, CONSUMER1, CONSUMER_ID1, 0, 0); + + ArgumentCaptor<Runnable> captor = ArgumentCaptor.forClass(Runnable.class); + verify(timer).scheduleWithFixedDelay(captor.capture(), anyLong(), anyLong(), any(TimeUnit.class)); + + captor.getValue().run(); + verify(data1).removeIdleConsumers(); + verify(data2).removeIdleConsumers(); + + // run it again + captor.getValue().run(); + verify(data1, times(2)).removeIdleConsumers(); + verify(data2, times(2)).removeIdleConsumers(); + } + + @Test + public void testMakeTimerPool() { + // use a real provider so we can test the real makeTimer() method + DmaapSimProvider prov2 = new DmaapSimProvider(params); + prov2.start(); + prov2.stop(); + } + + @Test + public void testMakeTopicData() { + // use a real provider so we can test the real makeTopicData() method + DmaapSimProvider prov2 = new DmaapSimProvider(params); + prov2.processDmaapMessageGet(TOPIC1, CONSUMER1, CONSUMER_ID1, 0, 0); + } + + @Test + public void testGetInstance_testSetInstance() { + DmaapSimProvider.setInstance(prov); + assertSame(prov, DmaapSimProvider.getInstance()); + + DmaapSimProvider.setInstance(null); + assertNull(DmaapSimProvider.getInstance()); + } + + + public class MyProvider extends DmaapSimProvider { + + public MyProvider(DmaapSimParameterGroup params) { + super(params); + } + + @Override + protected ScheduledExecutorService makeTimerPool() { + return timer; + } + + @Override + protected TopicData makeTopicData(String topicName) { + switch (topicName) { + case TOPIC1: + return data1; + case TOPIC2: + return data2; + default: + throw new IllegalArgumentException("unknown topic name: " + topicName); + } + } + } +} diff --git a/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/provider/TopicDataTest.java b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/provider/TopicDataTest.java new file mode 100644 index 000000000..f7e1f5e6c --- /dev/null +++ b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/provider/TopicDataTest.java @@ -0,0 +1,213 @@ +/*- + * ============LICENSE_START======================================================= + * ONAP Policy Models + * ================================================================================ + * Copyright (C) 2019 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.models.sim.dmaap.provider; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyInt; +import static org.mockito.Matchers.anyLong; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; +import java.util.TreeSet; +import java.util.stream.Collectors; +import org.junit.Before; +import org.junit.Test; +import org.mockito.internal.util.reflection.Whitebox; +import org.onap.policy.common.utils.coder.Coder; +import org.onap.policy.common.utils.coder.CoderException; +import org.onap.policy.common.utils.coder.StandardCoder; + +public class TopicDataTest { + private static final String EXPECTED_EXCEPTION = "expected exception"; + private static final String GROUP1 = "group-A"; + private static final String GROUP2 = "group-B"; + private static final String GROUP3 = "group-C"; + + private TopicData data; + private ConsumerGroupData consgrp1; + private ConsumerGroupData consgrp2; + private ConsumerGroupData consgrp3; + private List<ConsumerGroupData> groups; + + /** + * Sets up mocks and the initial data object. + * + * @throws Exception if an error occurs + */ + @Before + public void setUp() throws Exception { + consgrp1 = mock(ConsumerGroupData.class); + consgrp2 = mock(ConsumerGroupData.class); + consgrp3 = mock(ConsumerGroupData.class); + + when(consgrp1.read(anyInt(), anyLong())).thenReturn(Collections.emptyList()); + when(consgrp2.read(anyInt(), anyLong())).thenReturn(Collections.emptyList()); + when(consgrp3.read(anyInt(), anyLong())).thenReturn(Collections.emptyList()); + + groups = new LinkedList<>(Arrays.asList(consgrp1, consgrp2, consgrp3)); + + data = new TopicData("my-topic") { + @Override + protected ConsumerGroupData makeData(String consumerGroup) { + return groups.remove(0); + } + }; + } + + @Test + public void testRemoveIdleConsumers() throws Exception { + // force two consumers into the map + data.read(GROUP1, 0, 0); + data.read(GROUP2, 0, 0); + data.read(GROUP3, 0, 0); + + // indicate that one should be removed + when(consgrp1.shouldRemove()).thenReturn(true); + + // sweep + data.removeIdleConsumers(); + + assertEquals("[group-B, group-C]", new TreeSet<>(getGroups().keySet()).toString()); + + // indicate that the others should be removed + when(consgrp2.shouldRemove()).thenReturn(true); + when(consgrp3.shouldRemove()).thenReturn(true); + + // sweep + data.removeIdleConsumers(); + + assertTrue(getGroups().isEmpty()); + } + + @Test + public void testRead() throws Exception { + List<String> lst = Collections.emptyList(); + + when(consgrp1.read(anyInt(), anyLong())).thenReturn(ConsumerGroupData.UNREADABLE_LIST) + .thenReturn(ConsumerGroupData.UNREADABLE_LIST).thenReturn(lst); + + assertSame(lst, data.read(GROUP1, 10, 20)); + + // should have invoked three times + verify(consgrp1, times(3)).read(anyInt(), anyLong()); + + // should have used the given values + verify(consgrp1, times(3)).read(10, 20); + + // should not have allocated more than one group + assertEquals(2, groups.size()); + } + + @Test + public void testRead_MultipleGroups() throws Exception { + List<String> lst1 = Collections.emptyList(); + when(consgrp1.read(anyInt(), anyLong())).thenReturn(lst1); + + List<String> lst2 = Collections.emptyList(); + when(consgrp2.read(anyInt(), anyLong())).thenReturn(lst2); + + // one from each group + assertSame(lst1, data.read(GROUP1, 0, 0)); + assertSame(lst2, data.read(GROUP2, 0, 0)); + + // repeat + assertSame(lst1, data.read(GROUP1, 0, 0)); + assertSame(lst2, data.read(GROUP2, 0, 0)); + + // again + assertSame(lst1, data.read(GROUP1, 0, 0)); + assertSame(lst2, data.read(GROUP2, 0, 0)); + + // should still have group3 in the list + assertEquals(1, groups.size()); + } + + @Test + public void testWrite() throws Exception { + // no groups yet + List<Object> messages = Arrays.asList("hello", "world"); + data.write(messages); + + // add two groups + data.read(GROUP1, 0, 0); + data.read(GROUP2, 0, 0); + + data.write(messages); + + // should have been written to both groups + List<String> strings = messages.stream().map(Object::toString).collect(Collectors.toList()); + verify(consgrp1).write(strings); + verify(consgrp2).write(strings); + } + + @Test + public void testConvertMessagesToStrings() { + assertEquals("[abc, 200]", data.convertMessagesToStrings(Arrays.asList("abc", null, 200)).toString()); + } + + @Test + public void testConvertMessageToString() throws CoderException { + Coder coder = new StandardCoder(); + + assertNull(data.convertMessageToString(null, coder)); + assertEquals("text-msg", data.convertMessageToString("text-msg", coder)); + assertEquals("100", data.convertMessageToString(100, coder)); + + coder = mock(Coder.class); + when(coder.encode(any())).thenThrow(new CoderException(EXPECTED_EXCEPTION)); + assertNull(data.convertMessageToString(new TreeMap<String,Object>(), coder)); + } + + @Test + public void testMakeData() throws Exception { + // use real objects instead of mocks + TopicData data2 = new TopicData("real-data-topic"); + + // force a group into the topic + data2.read(GROUP1, 0, 0); + + data2.write(Arrays.asList("abc", "def", "ghi")); + + assertEquals("[abc, def]", data2.read(GROUP1, 2, 0).toString()); + } + + /** + * Gets the consumer group map from the topic data object. + * + * @return the topic's consumer group map + */ + @SuppressWarnings("unchecked") + private Map<String, ConsumerGroupData> getGroups() { + return (Map<String, ConsumerGroupData>) Whitebox.getInternalState(data, "group2data"); + } +} diff --git a/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/rest/BaseRestControllerV1Test.java b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/rest/BaseRestControllerV1Test.java new file mode 100644 index 000000000..5cba78355 --- /dev/null +++ b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/rest/BaseRestControllerV1Test.java @@ -0,0 +1,63 @@ +/* + * ============LICENSE_START======================================================= + * ONAP PAP + * ================================================================================ + * Copyright (C) 2019 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.models.sim.dmaap.rest; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import java.util.UUID; +import javax.ws.rs.core.Response; +import javax.ws.rs.core.Response.ResponseBuilder; +import org.junit.Before; +import org.junit.Test; + +public class BaseRestControllerV1Test { + + private BaseRestControllerV1 ctlr; + private ResponseBuilder bldr; + + @Before + public void setUp() { + ctlr = new BaseRestControllerV1(); + bldr = Response.status(Response.Status.OK); + } + + @Test + public void testAddVersionControlHeaders() { + Response resp = ctlr.addVersionControlHeaders(bldr).build(); + assertEquals("0", resp.getHeaderString(BaseRestControllerV1.VERSION_MINOR_NAME)); + assertEquals("0", resp.getHeaderString(BaseRestControllerV1.VERSION_PATCH_NAME)); + assertEquals("1.0.0", resp.getHeaderString(BaseRestControllerV1.VERSION_LATEST_NAME)); + } + + @Test + public void testAddLoggingHeaders_Null() { + Response resp = ctlr.addLoggingHeaders(bldr, null).build(); + assertNotNull(resp.getHeaderString(BaseRestControllerV1.REQUEST_ID_NAME)); + } + + @Test + public void testAddLoggingHeaders_NonNull() { + UUID uuid = UUID.randomUUID(); + Response resp = ctlr.addLoggingHeaders(bldr, uuid).build(); + assertEquals(uuid.toString(), resp.getHeaderString(BaseRestControllerV1.REQUEST_ID_NAME)); + } +} diff --git a/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/rest/CambriaMessageBodyHandlerTest.java b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/rest/CambriaMessageBodyHandlerTest.java new file mode 100644 index 000000000..5d9186c75 --- /dev/null +++ b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/rest/CambriaMessageBodyHandlerTest.java @@ -0,0 +1,145 @@ +/*- + * ============LICENSE_START======================================================= + * ONAP Policy Models + * ================================================================================ + * Copyright (C) 2019 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.models.sim.dmaap.rest; + +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.io.ByteArrayInputStream; +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.List; +import javax.ws.rs.core.MediaType; +import org.junit.Before; +import org.junit.Test; + +public class CambriaMessageBodyHandlerTest { + private static final String STD_INPUT = "1.3.XAbc"; + private static final String EXPECTED_OUTPUT = "[Abc]"; + + private CambriaMessageBodyHandler hdlr; + + @Before + public void setUp() { + hdlr = new CambriaMessageBodyHandler(); + } + + @Test + public void testIsReadable() { + assertTrue(hdlr.isReadable(null, null, null, MediaType.valueOf("application/cambria"))); + + assertFalse(hdlr.isReadable(null, null, null, null)); + assertFalse(hdlr.isReadable(null, null, null, MediaType.valueOf("application/other"))); + assertFalse(hdlr.isReadable(null, null, null, MediaType.valueOf("other/cambria"))); + } + + @Test + public void testReadFrom() throws IOException { + List<Object> lst = readStream("1.11.AMessageBody", "3.3.123Foo3.3.123Bar", "0.16.You can do that..8.Or that."); + assertEquals("[MessageBody, Foo, Bar, You can do that., Or that.]", lst.toString()); + + // empty stream + lst = readStream(); + assertEquals("[]", lst.toString()); + } + + @Test + public void testReadMessage_InvalidPartitionLength() { + assertThatThrownBy(() -> readStream("100000000.3.")).isInstanceOf(IOException.class) + .hasMessage("invalid partition length"); + } + + @Test + public void testReadMessage_InvalidMessageLength() { + assertThatThrownBy(() -> readStream("3.100000000.ABC")).isInstanceOf(IOException.class) + .hasMessage("invalid message length"); + } + + @Test + public void testSkipWhitespace() throws IOException { + // no white space + assertEquals(EXPECTED_OUTPUT, readStream(STD_INPUT).toString()); + + // single white space + assertEquals(EXPECTED_OUTPUT, readStream(" " + STD_INPUT).toString()); + + // multiple white spaces + assertEquals(EXPECTED_OUTPUT, readStream("\n\n\t" + STD_INPUT).toString()); + } + + @Test + public void testReadLength_NoDigits() throws IOException { + assertEquals("[]", readStream("..").toString()); + } + + @Test + public void testReadLength_NoDot() { + assertThatThrownBy(() -> readStream("3.2")).isInstanceOf(EOFException.class) + .hasMessage("missing '.' in 'length' field"); + } + + @Test + public void testReadLength_NonDigit() { + assertThatThrownBy(() -> readStream("3.2x.ABCde")).isInstanceOf(IOException.class) + .hasMessage("invalid character in 'length' field"); + } + + @Test + public void testReadLength_TooManyDigits() { + assertThatThrownBy(() -> readStream("3.12345678901234567890.ABCde")).isInstanceOf(IOException.class) + .hasMessage("too many digits in 'length' field"); + } + + @Test + public void testReadString_ZeroLength() throws IOException { + assertEquals("[]", readStream("1..X").toString()); + } + + @Test + public void testReadString_TooShort() { + assertThatThrownBy(() -> readStream(".5.me")).isInstanceOf(EOFException.class).hasMessageContaining("actual"); + } + + /** + * Reads a stream via the handler. + * + * @param text lines of text to be read + * @return the list of objects that were decoded from the stream + * @throws IOException if an error occurs + */ + private List<Object> readStream(String... text) throws IOException { + return hdlr.readFrom(null, null, null, null, null, makeStream(text)); + } + + /** + * Creates an input stream from lines of text. + * + * @param text lines of text + * @return an input stream + */ + private InputStream makeStream(String... text) { + return new ByteArrayInputStream(String.join("\n", text).getBytes(StandardCharsets.UTF_8)); + } +} diff --git a/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/rest/CommonRestServer.java b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/rest/CommonRestServer.java new file mode 100644 index 000000000..7e30c5a4c --- /dev/null +++ b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/rest/CommonRestServer.java @@ -0,0 +1,181 @@ +/* + * ============LICENSE_START======================================================= + * Modifications Copyright (C) 2019 AT&T Intellectual Property. + * ================================================================================ + * 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. + * + * SPDX-License-Identifier: Apache-2.0 + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.models.sim.dmaap.rest; + +import java.io.File; +import java.io.FileOutputStream; +import java.nio.charset.StandardCharsets; +import javax.ws.rs.client.Client; +import javax.ws.rs.client.ClientBuilder; +import javax.ws.rs.client.Invocation; +import javax.ws.rs.client.WebTarget; +import javax.ws.rs.core.MediaType; +import lombok.Getter; +import org.glassfish.jersey.client.ClientProperties; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.onap.policy.common.endpoints.http.server.HttpServletServerFactoryInstance; +import org.onap.policy.common.gson.GsonMessageBodyHandler; +import org.onap.policy.common.utils.network.NetworkUtil; +import org.onap.policy.common.utils.services.Registry; +import org.onap.policy.models.sim.dmaap.DmaapSimException; +import org.onap.policy.models.sim.dmaap.startstop.Main; +import org.onap.policy.sim.dmaap.parameters.CommonTestData; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Common base class for rest server tests. + */ +public class CommonRestServer { + + private static final Logger LOGGER = LoggerFactory.getLogger(CommonRestServer.class); + + public static final String NOT_ALIVE = "not alive"; + public static final String ALIVE = "alive"; + public static final String SELF = "self"; + public static final String NAME = "DMaaP Simulator"; + public static final String ENDPOINT_PREFIX = "events/"; + + @Getter + private static int port; + + protected static String httpPrefix; + + private static Main main; + + /** + * Allocates a port for the server, writes a config file, and then starts Main. + * + * @throws Exception if an error occurs + */ + @BeforeClass + public static void setUpBeforeClass() throws Exception { + port = NetworkUtil.allocPort(); + + httpPrefix = "http://localhost:" + port + "/"; + + String json = new CommonTestData().getParameterGroupAsString(port); + makeConfigFile("src/test/resources/parameters/TestConfigParams.json", json); + + HttpServletServerFactoryInstance.getServerFactory().destroy(); + + startMain(); + } + + /** + * Stops Main. + */ + @AfterClass + public static void teardownAfterClass() { + try { + if (main != null) { + Main main2 = main; + main = null; + + main2.shutdown(); + } + + } catch (DmaapSimException exp) { + LOGGER.error("cannot stop main", exp); + } + } + + /** + * Set up. + * + * @throws Exception if an error occurs + */ + @Before + public void setUp() throws Exception { + // restart, if not currently running + if (main == null) { + startMain(); + } + } + + /** + * Makes a parameter configuration file. + * @param fileName name of the config file to be created + * @param json json to be written to the file + * + * @throws Exception if an error occurs + */ + protected static void makeConfigFile(String fileName, String json) throws Exception { + File file = new File(fileName); + file.deleteOnExit(); + + try (FileOutputStream output = new FileOutputStream(file)) { + output.write(json.getBytes(StandardCharsets.UTF_8)); + } + } + + /** + * Starts the "Main". + * + * @throws Exception if an error occurs + */ + private static void startMain() throws Exception { + Registry.newRegistry(); + + // make sure port is available + if (NetworkUtil.isTcpPortOpen("localhost", port, 1, 1L)) { + throw new IllegalStateException("port " + port + " is still in use"); + } + + final String[] simConfigParameters = {"-c", "src/test/resources/parameters/TestConfigParams.json"}; + + main = new Main(simConfigParameters); + + if (!NetworkUtil.isTcpPortOpen("localhost", port, 60, 1000L)) { + throw new IllegalStateException("server is not listening on port " + port); + } + } + + /** + * Sends a request to an endpoint. + * + * @param endpoint the target endpoint + * @return a request builder + * @throws Exception if an error occurs + */ + protected Invocation.Builder sendRequest(final String endpoint) throws Exception { + return sendFqeRequest(httpPrefix + ENDPOINT_PREFIX + endpoint); + } + + /** + * Sends a request to a fully qualified endpoint. + * + * @param fullyQualifiedEndpoint the fully qualified target endpoint + * @return a request builder + */ + protected Invocation.Builder sendFqeRequest(final String fullyQualifiedEndpoint) { + final Client client = ClientBuilder.newBuilder().build(); + + client.property(ClientProperties.METAINF_SERVICES_LOOKUP_DISABLE, "true"); + client.register(GsonMessageBodyHandler.class); + + final WebTarget webTarget = client.target(fullyQualifiedEndpoint); + + return webTarget.request(MediaType.APPLICATION_JSON); + } +} diff --git a/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/rest/DmaapSimRestControllerV1Test.java b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/rest/DmaapSimRestControllerV1Test.java new file mode 100644 index 000000000..7b84d543d --- /dev/null +++ b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/rest/DmaapSimRestControllerV1Test.java @@ -0,0 +1,94 @@ +/*- + * ============LICENSE_START======================================================= + * Copyright (C) 2019 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.models.sim.dmaap.rest; + +import static org.junit.Assert.assertEquals; + +import java.io.File; +import java.util.Arrays; +import java.util.Map; +import javax.ws.rs.core.Response; +import org.junit.Before; +import org.junit.Test; +import org.onap.policy.common.utils.coder.Coder; +import org.onap.policy.common.utils.coder.CoderException; +import org.onap.policy.common.utils.coder.StandardCoder; +import org.onap.policy.models.sim.dmaap.parameters.DmaapSimParameterGroup; +import org.onap.policy.models.sim.dmaap.provider.DmaapSimProvider; + +public class DmaapSimRestControllerV1Test { + private static final int LIMIT = 5; + private static final String TOPIC = "my-topic"; + private static final String TOPIC2 = "my-topic-B"; + private static final String MESSAGE = "my-message"; + private static final String MESSAGE2 = "my-message-B"; + private static final String CONSUMER = "my-consumer"; + private static final String CONSUMER_ID = "my-id"; + + private static Coder coder = new StandardCoder(); + + private DmaapSimRestControllerV1 rest; + + /** + * Creates the controller. + * + * @throws CoderException if the parameters cannot be loaded + */ + @Before + public void setUp() throws CoderException { + DmaapSimParameterGroup params = coder.decode(new File("src/test/resources/parameters/NormalParameters.json"), + DmaapSimParameterGroup.class); + DmaapSimProvider.setInstance(new DmaapSimProvider(params)); + rest = new DmaapSimRestControllerV1(); + } + + @Test + public void test() { + Response resp = rest.getDmaapMessage(TOPIC, CONSUMER, CONSUMER_ID, LIMIT, 0); + assertEquals(Response.Status.OK.getStatusCode(), resp.getStatus()); + assertEquals("[]", resp.getEntity().toString()); + + // add some messages + resp = rest.postDmaapMessage(TOPIC, Arrays.asList(MESSAGE, MESSAGE2)); + assertEquals(Response.Status.OK.getStatusCode(), resp.getStatus()); + assertEquals(2, getCount(resp)); + + resp = rest.postDmaapMessage(TOPIC2, Arrays.asList(MESSAGE, MESSAGE2, MESSAGE)); + assertEquals(Response.Status.OK.getStatusCode(), resp.getStatus()); + assertEquals(3, getCount(resp)); + + // hadn't registered with topic 2 so nothing expected from there + resp = rest.getDmaapMessage(TOPIC2, CONSUMER, CONSUMER_ID, LIMIT, 0); + assertEquals(Response.Status.OK.getStatusCode(), resp.getStatus()); + assertEquals("[]", resp.getEntity().toString()); + + // now read from topic 1 + resp = rest.getDmaapMessage(TOPIC, CONSUMER, CONSUMER_ID, LIMIT, 0); + assertEquals(Response.Status.OK.getStatusCode(), resp.getStatus()); + assertEquals("[my-message, my-message-B]", resp.getEntity().toString()); + } + + private int getCount(Response resp) { + @SuppressWarnings("unchecked") + Map<String, Object> map = (Map<String, Object>) resp.getEntity(); + + return (int) map.get("count"); + } + +} diff --git a/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/rest/TextMessageBodyHandlerTest.java b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/rest/TextMessageBodyHandlerTest.java new file mode 100644 index 000000000..2dfbae980 --- /dev/null +++ b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/models/sim/dmaap/rest/TextMessageBodyHandlerTest.java @@ -0,0 +1,81 @@ +/*- + * ============LICENSE_START======================================================= + * Copyright (C) 2019 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.models.sim.dmaap.rest; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.List; +import javax.ws.rs.core.MediaType; +import org.junit.Before; +import org.junit.Test; + +public class TextMessageBodyHandlerTest { + private TextMessageBodyHandler hdlr; + + @Before + public void setUp() { + hdlr = new TextMessageBodyHandler(); + } + + @Test + public void testIsReadable() { + assertTrue(hdlr.isReadable(null, null, null, MediaType.valueOf("text/plain"))); + + assertFalse(hdlr.isReadable(null, null, null, null)); + assertFalse(hdlr.isReadable(null, null, null, MediaType.valueOf("text/other"))); + assertFalse(hdlr.isReadable(null, null, null, MediaType.valueOf("other/plain"))); + } + + @Test + public void testReadFrom() throws IOException { + List<Object> lst = readStream("hello", "world"); + assertEquals("[hello, world]", lst.toString()); + + // empty stream + lst = readStream(); + assertEquals("[]", lst.toString()); + } + + /** + * Reads a stream via the handler. + * + * @param text lines of text to be read + * @return the list of objects that were decoded from the stream + * @throws IOException if an error occurs + */ + private List<Object> readStream(String... text) throws IOException { + return hdlr.readFrom(null, null, null, null, null, makeStream(text)); + } + + /** + * Creates an input stream from lines of text. + * + * @param text lines of text + * @return an input stream + */ + private InputStream makeStream(String... text) { + return new ByteArrayInputStream(String.join("\n", text).getBytes(StandardCharsets.UTF_8)); + } +} diff --git a/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/e2e/EndToEndTest.java b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/e2e/EndToEndTest.java new file mode 100644 index 000000000..8c35de64e --- /dev/null +++ b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/e2e/EndToEndTest.java @@ -0,0 +1,199 @@ +/*- + * ============LICENSE_START======================================================= + * Copyright (C) 2019 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.sim.dmaap.e2e; + +import static org.junit.Assert.assertEquals; + +import java.io.File; +import java.io.PrintWriter; +import java.net.HttpURLConnection; +import java.net.URL; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.function.BiConsumer; +import org.junit.After; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.onap.policy.common.endpoints.event.comm.TopicEndpointManager; +import org.onap.policy.common.endpoints.event.comm.TopicSink; +import org.onap.policy.common.endpoints.parameters.TopicParameterGroup; +import org.onap.policy.common.endpoints.parameters.TopicParameters; +import org.onap.policy.common.utils.coder.CoderException; +import org.onap.policy.common.utils.coder.StandardCoder; +import org.onap.policy.models.sim.dmaap.rest.CommonRestServer; + +/** + * This tests the simulator using dmaap endpoints to verify that it works from publisher + * to subscriber. + */ +public class EndToEndTest extends CommonRestServer { + private static final int MAX_WAIT_SEC = 5; + private static final String ORIG_TOPIC = "MY-TOPIC"; + private static final String ORIG_TOPIC2 = "MY-TOPIC-B"; + private static final int MAX_MSG = 200; + + private static int ntests = 0; + private static String topicJson; + + private TopicParameterGroup topicConfig; + + private String topic = "MY-TOPIC"; + private String topic2 = "MY-TOPIC-B"; + + /** + * Messages from the topic are placed here by the endpoint. + */ + private BlockingQueue<String> queue; + + /** + * Messages from topic 2 are placed here by the endpoint. + */ + private BlockingQueue<String> queue2; + + /** + * Starts the rest server. + * + * @throws Exception if an error occurs + */ + @BeforeClass + public static void setUpBeforeClass() throws Exception { + TopicEndpointManager.getManager().shutdown(); + + CommonRestServer.setUpBeforeClass(); + + topicJson = new String( + Files.readAllBytes(new File("src/test/resources/parameters/TopicParameters.json").toPath()), + StandardCharsets.UTF_8); + topicJson = topicJson.replace("${port}", String.valueOf(getPort())); + } + + /** + * Starts the topics. + * + * @throws CoderException if the parameters cannot be decoded + */ + @Before + @Override + public void setUp() throws CoderException { + queue = new LinkedBlockingQueue<>(); + queue2 = new LinkedBlockingQueue<>(); + + /* + * change topic names for each test so any listeners that may still exist will not + * grab new messages + */ + + ++ntests; + topic = "my-topic-" + ntests; + topic2 = "my-topic-b" + ntests; + + String json = topicJson.replace(ORIG_TOPIC2, topic2).replace(ORIG_TOPIC, topic); + + topicConfig = new StandardCoder().decode(json, TopicParameterGroup.class); + + TopicEndpointManager.getManager().addTopics(topicConfig); + TopicEndpointManager.getManager().start(); + } + + @After + public void tearDown() { + TopicEndpointManager.getManager().shutdown(); + } + + @Test + public void testWithTopicEndpointAtEachEnd() throws InterruptedException { + // register listeners to add events to appropriate queue + TopicEndpointManager.getManager().getDmaapTopicSource(topic) + .register((infra, topic, event) -> queue.add(event)); + TopicEndpointManager.getManager().getDmaapTopicSource(topic2) + .register((infra, topic, event) -> queue2.add(event)); + + // publish events + TopicSink sink = TopicEndpointManager.getManager().getDmaapTopicSink(topic); + TopicSink sink2 = TopicEndpointManager.getManager().getDmaapTopicSink(topic2); + for (int x = 0; x < MAX_MSG; ++x) { + sink.send("hello-" + x); + sink2.send("world-" + x); + } + + // verify events where received + for (int x = 0; x < MAX_MSG; ++x) { + assertEquals("message " + x, "hello-" + x, queue.poll(MAX_WAIT_SEC, TimeUnit.SECONDS)); + assertEquals("message " + x, "world-" + x, queue2.poll(MAX_WAIT_SEC, TimeUnit.SECONDS)); + } + } + + @Test + public void testCambriaFormat() throws Exception { + test("testCambriaFormat", "application/cambria", + (wtr, messages) -> messages.forEach(msg -> wtr.write("0." + msg.length() + "." + msg + "\n"))); + } + + @Test + public void testJson() throws Exception { + test("testJson", "application/json", (wtr, messages) -> wtr.write("[" + String.join(", ", messages) + "]")); + } + + @Test + public void testText() throws Exception { + test("testText", "text/plain", (wtr, messages) -> messages.forEach(wtr::println)); + } + + /** + * Uses a raw URL connection to ensure the server can process messages of the given + * media type. + * + * @param testName name of the test + * @param mediaType media type + * @param writeMessages function that writes messages to a PrintWriter + * @throws Exception if an error occurs + */ + private void test(String testName, String mediaType, BiConsumer<PrintWriter, List<String>> writeMessages) + throws Exception { + String msg1 = "{'abc':10.0}".replace('\'', '"'); + String msg2 = "{'def':20.0}".replace('\'', '"'); + + TopicEndpointManager.getManager().getDmaapTopicSource(topic) + .register((infra, topic, event) -> queue.add(event)); + + TopicParameters sinkcfg = topicConfig.getTopicSinks().get(0); + URL url = new URL(httpPrefix + "events/" + sinkcfg.getTopic()); + + HttpURLConnection conn = (HttpURLConnection) url.openConnection(); + conn.setRequestMethod("POST"); + conn.setRequestProperty("Content-type", mediaType); + conn.setDoOutput(true); + conn.connect(); + + try (PrintWriter wtr = new PrintWriter(conn.getOutputStream())) { + writeMessages.accept(wtr, Arrays.asList(msg1, msg2)); + } + + assertEquals(testName + " response code", HttpURLConnection.HTTP_OK, conn.getResponseCode()); + + assertEquals(testName + " message 1", msg1, queue.poll(MAX_WAIT_SEC, TimeUnit.SECONDS)); + assertEquals(testName + " message 2", msg2, queue.poll(MAX_WAIT_SEC, TimeUnit.SECONDS)); + } +} diff --git a/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/parameters/CommonTestData.java b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/parameters/CommonTestData.java new file mode 100644 index 000000000..21d9ed604 --- /dev/null +++ b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/parameters/CommonTestData.java @@ -0,0 +1,89 @@ +/*- + * ============LICENSE_START======================================================= + * Modifications Copyright (C) 2019 AT&T Intellectual Property. + * ================================================================================ + * 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. + * + * SPDX-License-Identifier: Apache-2.0 + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.sim.dmaap.parameters; + +import java.io.File; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import org.onap.policy.common.utils.coder.Coder; +import org.onap.policy.common.utils.coder.CoderException; +import org.onap.policy.common.utils.coder.StandardCoder; +import org.onap.policy.models.sim.dmaap.DmaapSimRuntimeException; +import org.onap.policy.models.sim.dmaap.parameters.DmaapSimParameterGroup; + +/** + * Class to hold/create all parameters for test cases. + */ +public class CommonTestData { + public static final String SIM_GROUP_NAME = "DMaapSim"; + + private static final Coder coder = new StandardCoder(); + + /** + * Gets the standard simulator parameters. + * + * @param port port to be inserted into the parameters + * @return the standard simulator parameters + */ + public DmaapSimParameterGroup getParameterGroup(int port) { + try { + return coder.decode(getParameterGroupAsString(port), DmaapSimParameterGroup.class); + + } catch (CoderException e) { + throw new DmaapSimRuntimeException("cannot read simulator parameters", e); + } + } + + /** + * Gets the standard simulator parameters, as a String. + * + * @param port port to be inserted into the parameters + * @return the standard simulator parameters + */ + public String getParameterGroupAsString(int port) { + + try { + File file = new File("src/test/resources/parameters/NormalParameters.json"); + String json = new String(Files.readAllBytes(file.toPath()), StandardCharsets.UTF_8); + + json = json.replace("6845", String.valueOf(port)); + + return json; + + } catch (IOException e) { + throw new DmaapSimRuntimeException("cannot read simulator parameters", e); + } + } + + /** + * Nulls out a field within a JSON string. It does it by adding a field with the same + * name, having a null value, and then prefixing the original field name with "Xxx", + * thus causing the original field and value to be ignored. + * + * @param json JSON string + * @param field field to be nulled out + * @return a new JSON string with the field nulled out + */ + public String nullifyField(String json, String field) { + return json.replace(field + "\"", field + "\":null, \"" + field + "Xxx\""); + } +} diff --git a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/DmaapSimConstants.java b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/parameters/DmaapSimParameterGroupTest.java index 1682eb991..828cd89b0 100644 --- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/DmaapSimConstants.java +++ b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/parameters/DmaapSimParameterGroupTest.java @@ -1,6 +1,6 @@ -/* +/*- * ============LICENSE_START======================================================= - * Copyright (C) 2019 Nordix Foundation. + * Copyright (C) 2019 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. @@ -16,17 +16,19 @@ * ============LICENSE_END========================================================= */ -package org.onap.policy.models.sim.dmaap; +package org.onap.policy.sim.dmaap.parameters; -/** - * Names of various items contained in the Registry. - */ -public class DmaapSimConstants { +import static org.junit.Assert.assertEquals; + +import org.junit.Test; +import org.onap.policy.models.sim.dmaap.parameters.DmaapSimParameterGroup; - // Registry keys - public static final String REG_DMAAP_SIM_ACTIVATOR = "object:activator/dmaap-sim"; +public class DmaapSimParameterGroupTest { + private static final String MY_NAME = "my-name"; - private DmaapSimConstants() { - super(); + @Test + public void testDmaapSimParameterGroup() { + DmaapSimParameterGroup params = new DmaapSimParameterGroup(MY_NAME); + assertEquals(MY_NAME, params.getName()); } } diff --git a/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/parameters/DmaapSimParameterHandlerTest.java b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/parameters/DmaapSimParameterHandlerTest.java new file mode 100644 index 000000000..8f053d219 --- /dev/null +++ b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/parameters/DmaapSimParameterHandlerTest.java @@ -0,0 +1,70 @@ +/*- + * ============LICENSE_START======================================================= + * Copyright (C) 2019 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.sim.dmaap.parameters; + +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import org.junit.Before; +import org.junit.Test; +import org.onap.policy.models.sim.dmaap.DmaapSimException; +import org.onap.policy.models.sim.dmaap.parameters.DmaapSimParameterGroup; +import org.onap.policy.models.sim.dmaap.parameters.DmaapSimParameterHandler; +import org.onap.policy.models.sim.dmaap.startstop.DmaapSimCommandLineArguments; + +public class DmaapSimParameterHandlerTest { + + private static final String RESOURCE_DIR = "src/test/resources/parameters/"; + + private DmaapSimParameterHandler handler; + + @Before + public void setUp() { + handler = new DmaapSimParameterHandler(); + } + + @Test + public void testGetParameters() throws DmaapSimException { + final DmaapSimCommandLineArguments args = new DmaapSimCommandLineArguments(); + + args.parse(new String[] {"-c", RESOURCE_DIR + "NormalParameters.json"}); + DmaapSimParameterGroup params = handler.getParameters(args); + assertNotNull(params); + assertEquals("DMaapSim", params.getName()); + assertEquals(300L, params.getTopicSweepSec()); + assertEquals(6845, params.getRestServerParameters().getPort()); + + + args.parse(new String[] {"-c", "FileNotFound.json"}); + assertThatThrownBy(() -> handler.getParameters(args)).isInstanceOf(DmaapSimException.class) + .hasMessageStartingWith("error reading parameters"); + + + args.parse(new String[] {"-c", RESOURCE_DIR + "EmptyParameterFile.json"}); + assertThatThrownBy(() -> handler.getParameters(args)).isInstanceOf(DmaapSimException.class) + .hasMessageStartingWith("no parameters found"); + + + args.parse(new String[] {"-c", RESOURCE_DIR + "Parameters_InvalidName.json"}); + assertThatThrownBy(() -> handler.getParameters(args)).isInstanceOf(DmaapSimException.class) + .hasMessageContaining("validation error"); + } + +} diff --git a/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/startstop/DmaapSimActivatorTest.java b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/startstop/DmaapSimActivatorTest.java new file mode 100644 index 000000000..380a72423 --- /dev/null +++ b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/startstop/DmaapSimActivatorTest.java @@ -0,0 +1,95 @@ +/*- + * ============LICENSE_START======================================================= + * Copyright (C) 2019 AT&T Intellectual Property. + * ================================================================================ + * 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. + * + * SPDX-License-Identifier: Apache-2.0 + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.sim.dmaap.startstop; + +import static org.assertj.core.api.Assertions.assertThatIllegalStateException; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.onap.policy.common.endpoints.http.server.HttpServletServerFactoryInstance; +import org.onap.policy.common.utils.services.Registry; +import org.onap.policy.models.sim.dmaap.parameters.DmaapSimParameterGroup; +import org.onap.policy.models.sim.dmaap.parameters.DmaapSimParameterHandler; +import org.onap.policy.models.sim.dmaap.startstop.DmaapSimActivator; +import org.onap.policy.models.sim.dmaap.startstop.DmaapSimCommandLineArguments; + + +/** + * Class to perform unit test of {@link DmaapSimActivator}}. + */ +public class DmaapSimActivatorTest { + + private DmaapSimActivator activator; + + /** + * Initializes an activator. + * + * @throws Exception if an error occurs + */ + @Before + public void setUp() throws Exception { + Registry.newRegistry(); + HttpServletServerFactoryInstance.getServerFactory().destroy(); + + final String[] papConfigParameters = {"-c", "parameters/NormalParameters.json"}; + final DmaapSimCommandLineArguments arguments = new DmaapSimCommandLineArguments(papConfigParameters); + final DmaapSimParameterGroup parGroup = new DmaapSimParameterHandler().getParameters(arguments); + + activator = new DmaapSimActivator(parGroup); + } + + /** + * Method for cleanup after each test. + * + * @throws Exception if an error occurs + */ + @After + public void teardown() throws Exception { + if (activator != null && activator.isAlive()) { + activator.stop(); + } + } + + @Test + public void testDmaapSimActivator() { + assertFalse(activator.isAlive()); + activator.start(); + assertTrue(activator.isAlive()); + + // repeat - should throw an exception + assertThatIllegalStateException().isThrownBy(() -> activator.start()); + assertTrue(activator.isAlive()); + } + + @Test + public void testTerminate() { + activator.start(); + activator.stop(); + assertFalse(activator.isAlive()); + + // repeat - should throw an exception + assertThatIllegalStateException().isThrownBy(() -> activator.stop()); + assertFalse(activator.isAlive()); + } +} diff --git a/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/startstop/MainTest.java b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/startstop/MainTest.java new file mode 100644 index 000000000..b8e285a99 --- /dev/null +++ b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/startstop/MainTest.java @@ -0,0 +1,100 @@ +/* + * ============LICENSE_START======================================================= + * Modifications Copyright (C) 2019 AT&T Intellectual Property. + * ================================================================================ + * 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. + * + * SPDX-License-Identifier: Apache-2.0 + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.sim.dmaap.startstop; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.onap.policy.common.endpoints.http.server.HttpServletServerFactoryInstance; +import org.onap.policy.models.sim.dmaap.DmaapSimException; +import org.onap.policy.models.sim.dmaap.startstop.Main; +import org.onap.policy.sim.dmaap.parameters.CommonTestData; + +/** + * Class to perform unit test of {@link Main}}. + * + * @author Ram Krishna Verma (ram.krishna.verma@est.tech) + */ +public class MainTest { + private Main main; + + /** + * Set up. + */ + @Before + public void setUp() { + main = null; + HttpServletServerFactoryInstance.getServerFactory().destroy(); + } + + /** + * Shuts "main" down. + * + * @throws Exception if an error occurs + */ + @After + public void tearDown() throws Exception { + if (main != null) { + main.shutdown(); + } + } + + @Test + public void testMain() throws DmaapSimException { + final String[] NormalParameters = {"-c", "parameters/NormalParameters.json"}; + main = new Main(NormalParameters); + assertTrue(main.getParameters().isValid()); + assertEquals(CommonTestData.SIM_GROUP_NAME, main.getParameters().getName()); + + main.shutdown(); + } + + @Test + public void testMain_NoArguments() { + final String[] NormalParameters = {}; + main = new Main(NormalParameters); + assertTrue(main.getParameters() == null); + } + + @Test + public void testMain_InvalidArguments() { + // note: this is missing the "-c" argument, thus the ARGUMENTS are invalid + final String[] NormalParameters = {"parameters/NormalParameters.json"}; + main = new Main(NormalParameters); + assertTrue(main.getParameters() == null); + } + + @Test + public void testMain_Help() { + final String[] NormalParameters = {"-h"}; + Main.main(NormalParameters); + } + + @Test + public void testMain_InvalidParameters() { + final String[] NormalParameters = {"-c", "parameters/InvalidParameters.json"}; + main = new Main(NormalParameters); + assertTrue(main.getParameters() == null); + } +} diff --git a/models-sim/models-sim-dmaap/src/test/resources/parameters/EmptyParameterFile.json b/models-sim/models-sim-dmaap/src/test/resources/parameters/EmptyParameterFile.json new file mode 100644 index 000000000..e69de29bb --- /dev/null +++ b/models-sim/models-sim-dmaap/src/test/resources/parameters/EmptyParameterFile.json diff --git a/models-sim/models-sim-dmaap/src/test/resources/parameters/MinimumParameters.json b/models-sim/models-sim-dmaap/src/test/resources/parameters/MinimumParameters.json index aeedf9d6e..a1c98a5b1 100644 --- a/models-sim/models-sim-dmaap/src/test/resources/parameters/MinimumParameters.json +++ b/models-sim/models-sim-dmaap/src/test/resources/parameters/MinimumParameters.json @@ -1,5 +1,6 @@ -{ +{ "name":"DMaapSim", + "topicSweepSec": 1, "restServerParameters":{ "host":"0.0.0.0", "port":6845 diff --git a/models-sim/models-sim-dmaap/src/test/resources/parameters/NormalParameters.json b/models-sim/models-sim-dmaap/src/test/resources/parameters/NormalParameters.json index a2a036645..deec966e8 100644 --- a/models-sim/models-sim-dmaap/src/test/resources/parameters/NormalParameters.json +++ b/models-sim/models-sim-dmaap/src/test/resources/parameters/NormalParameters.json @@ -1,5 +1,6 @@ { "name": "DMaapSim", + "topicSweepSec": 300, "restServerParameters": { "host": "0.0.0.0", "port": 6845 diff --git a/models-sim/models-sim-dmaap/src/test/resources/parameters/Parameters_InvalidName.json b/models-sim/models-sim-dmaap/src/test/resources/parameters/Parameters_InvalidName.json index fba033e52..51e9458b0 100644 --- a/models-sim/models-sim-dmaap/src/test/resources/parameters/Parameters_InvalidName.json +++ b/models-sim/models-sim-dmaap/src/test/resources/parameters/Parameters_InvalidName.json @@ -1,5 +1,6 @@ { "name":" ", + "topicSweepSec": 1, "restServerParameters":{ "host":"0.0.0.0", "port":6969, diff --git a/models-sim/models-sim-dmaap/src/test/resources/parameters/TopicParameters.json b/models-sim/models-sim-dmaap/src/test/resources/parameters/TopicParameters.json new file mode 100644 index 000000000..77a320f6d --- /dev/null +++ b/models-sim/models-sim-dmaap/src/test/resources/parameters/TopicParameters.json @@ -0,0 +1,36 @@ +{ + "topicSources": [ + { + "topic": "MY-TOPIC", + "servers": [ + "localhost:${port}" + ], + "topicCommInfrastructure": "dmaap", + "fetchTimeout": 100 + }, + { + "topic": "MY-TOPIC-B", + "servers": [ + "localhost:${port}" + ], + "topicCommInfrastructure": "dmaap", + "fetchTimeout": 100 + } + ], + "topicSinks": [ + { + "topic": "MY-TOPIC", + "servers": [ + "localhost:${port}" + ], + "topicCommInfrastructure": "dmaap" + }, + { + "topic": "MY-TOPIC-B", + "servers": [ + "localhost:${port}" + ], + "topicCommInfrastructure": "dmaap" + } + ] +}
\ No newline at end of file |