aboutsummaryrefslogtreecommitdiffstats
path: root/models-sim/models-sim-dmaap/src/main/java/org
diff options
context:
space:
mode:
Diffstat (limited to 'models-sim/models-sim-dmaap/src/main/java/org')
-rw-r--r--models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/parameters/DmaapSimParameterGroup.java56
-rw-r--r--models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/parameters/RestServerParameters.java95
-rw-r--r--models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/ConsumerGroupData.java190
-rw-r--r--models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/DmaapGetTopicResponse.java37
-rw-r--r--models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/DmaapSimProvider.java189
-rw-r--r--models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/TopicData.java200
-rw-r--r--models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/BaseRestControllerV1.java94
-rw-r--r--models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/CambriaMessageBodyHandler.java177
-rw-r--r--models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/DmaapSimRestControllerV1.java91
-rw-r--r--models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/TextMessageBodyHandler.java66
10 files changed, 0 insertions, 1195 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
deleted file mode 100644
index af5e4fd46..000000000
--- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/parameters/DmaapSimParameterGroup.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * Copyright (C) 2019 Nordix Foundation.
- * Modifications Copyright (C) 2019-2021 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.
- *
- * SPDX-License-Identifier: Apache-2.0
- * ============LICENSE_END=========================================================
- */
-
-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;
-import org.onap.policy.common.parameters.annotations.Valid;
-
-/**
- * Class to hold all parameters needed for the DMaaP simulator component.
- */
-@NotNull
-@NotBlank
-@Getter
-public class DmaapSimParameterGroup extends ParameterGroupImpl {
- private @Valid 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
- */
- public DmaapSimParameterGroup(final String name) {
- super(name);
- }
-}
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
deleted file mode 100644
index 4a7b12cbf..000000000
--- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/parameters/RestServerParameters.java
+++ /dev/null
@@ -1,95 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * Copyright (C) 2019,2023 Nordix Foundation.
- * Modifications Copyright (C) 2019, 2021 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.
- *
- * SPDX-License-Identifier: Apache-2.0
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.policy.models.sim.dmaap.parameters;
-
-import java.util.Properties;
-import lombok.Getter;
-import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
-import org.onap.policy.common.gson.GsonMessageBodyHandler;
-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;
-import org.onap.policy.models.sim.dmaap.rest.CambriaMessageBodyHandler;
-import org.onap.policy.models.sim.dmaap.rest.DmaapSimRestControllerV1;
-import org.onap.policy.models.sim.dmaap.rest.TextMessageBodyHandler;
-
-/**
- * Class to hold all parameters needed for rest server.
- */
-@NotNull
-@NotBlank
-@Getter
-public class RestServerParameters extends ParameterGroupImpl {
- private String host;
-
- @Min(value = 1)
- private int port;
-
- private String userName;
-
- private String password;
-
- private boolean useHttps;
-
- private boolean sniHostCheck;
-
- public RestServerParameters() {
- super(RestServerParameters.class.getSimpleName());
- }
-
- /**
- * Creates a set of properties, suitable for building a REST server, from the
- * parameters.
- *
- * @return a set of properties representing the given parameters
- */
- public Properties getServerProperties() {
- final var props = new Properties();
- props.setProperty(PolicyEndPointProperties.PROPERTY_HTTP_SERVER_SERVICES, getName());
-
- final String svcpfx =
- PolicyEndPointProperties.PROPERTY_HTTP_SERVER_SERVICES + "." + getName();
-
- props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_HOST_SUFFIX, getHost());
- props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_PORT_SUFFIX,
- Integer.toString(getPort()));
- props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_REST_CLASSES_SUFFIX,
- DmaapSimRestControllerV1.class.getName());
- props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX, "false");
- props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_SWAGGER_SUFFIX, "false");
- props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX, Boolean.toString(isUseHttps()));
- props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_SNI_HOST_CHECK_SUFFIX,
- Boolean.toString(isSniHostCheck()));
-
- if (getUserName() != null && getPassword() != null) {
- props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_AUTH_USERNAME_SUFFIX, getUserName());
- props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_AUTH_PASSWORD_SUFFIX, getPassword());
- }
-
- props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_SERIALIZATION_PROVIDER,
- String.join(",", CambriaMessageBodyHandler.class.getName(),
- GsonMessageBodyHandler.class.getName(),
- TextMessageBodyHandler.class.getName()));
- return props;
- }
-}
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
deleted file mode 100644
index 3acaf0888..000000000
--- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/ConsumerGroupData.java
+++ /dev/null
@@ -1,190 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * ONAP Policy Models
- * ================================================================================
- * Copyright (C) 2019, 2021 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 (var 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/DmaapGetTopicResponse.java b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/DmaapGetTopicResponse.java
deleted file mode 100644
index 1f05976f7..000000000
--- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/DmaapGetTopicResponse.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * Copyright (C) 2021 Bell Canada. 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.
- *
- * SPDX-License-Identifier: Apache-2.0
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.policy.models.sim.dmaap.provider;
-
-import java.util.List;
-import lombok.Getter;
-import lombok.Setter;
-import lombok.ToString;
-
-/**
- * Class to capture get topic response from dmaap simulator.
- */
-@Getter
-@Setter
-@ToString
-public class DmaapGetTopicResponse {
-
- private List<String> topics;
-}
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
deleted file mode 100644
index afbe10f51..000000000
--- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/DmaapSimProvider.java
+++ /dev/null
@@ -1,189 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * Copyright (C) 2019, 2023 Nordix Foundation.
- * Modifications Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
- * Modifications Copyright (C) 2021 Bell Canada. 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.
- *
- * SPDX-License-Identifier: Apache-2.0
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.policy.models.sim.dmaap.provider;
-
-import jakarta.ws.rs.core.Response;
-import jakarta.ws.rs.core.Response.Status;
-import java.util.Collections;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-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;
-
-/**
- * Provider to simulate DMaaP.
- *
- * @author Liam Fallon (liam.fallon@est.tech)
- */
-public class DmaapSimProvider extends ServiceManagerContainer {
- private static final Logger LOGGER = LoggerFactory.getLogger(DmaapSimProvider.class);
-
- @Getter
- @Setter
- private static DmaapSimProvider instance;
-
- /**
- * Maps a topic name to its data.
- */
- private final Map<String, TopicData> topic2data = new ConcurrentHashMap<>();
-
- /**
- * Thread used to remove idle consumers from the topics.
- */
- private ScheduledExecutorService timerPool;
-
-
- /**
- * 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 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: {}, Received DMaaP message(s): {}", topicName, dmaapMessage);
-
- List<Object> lst;
-
- if (dmaapMessage instanceof List) {
- lst = (List<Object>) dmaapMessage;
- } else {
- lst = Collections.singletonList(dmaapMessage);
- }
-
- 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();
- }
-
- /**
- * Wait for and return a DMaaP message.
- *
- * @param topicName The topic to wait on
- * @param consumerGroup the consumer group that is waiting
- * @param consumerId the consumer ID that is waiting
- * @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 limit, final long timeoutMs) {
-
- LOGGER.debug("Topic: {}, Request for DMaaP message: {}: {} with limit={} timeout={}", topicName, consumerGroup,
- consumerId, limit, timeoutMs);
-
- try {
- List<String> lst = topic2data.computeIfAbsent(topicName, this::makeTopicData).read(consumerGroup, limit,
- timeoutMs);
-
- LOGGER.debug("Topic: {}, Retrieved {} messages for: {}: {}", topicName, lst.size(), consumerGroup,
- consumerId);
- return Response.status(Status.OK).entity(lst).build();
-
- } 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();
- }
- }
-
- /**
- * Returns the list of default topics.
- *
- * @return the topic list
- */
- public Response processDmaapTopicsGet() {
-
- LOGGER.debug("Request for listing DMaaP topics");
- var response = new DmaapGetTopicResponse();
- response.setTopics(List.of("POLICY-PDP-PAP", "POLICY-NOTIFICATION", "unauthenticated.DCAE_CL_OUTPUT",
- "POLICY-CL-MGT"));
- return Response.status(Status.OK).entity(response).build();
- }
-
- /**
- * Task to remove idle consumers from each topic.
- */
- private class SweeperTask implements Runnable {
- @Override
- public void run() {
- topic2data.values().forEach(TopicData::removeIdleConsumers);
- }
- }
-
- // the following methods may be overridden by junit tests
-
- /**
- * Makes a new timer pool.
- *
- * @return a new timer pool
- */
- protected ScheduledExecutorService makeTimerPool() {
- return Executors.newScheduledThreadPool(1);
- }
-
- /**
- * 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
deleted file mode 100644
index 8e5cf24cc..000000000
--- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/provider/TopicData.java
+++ /dev/null
@@ -1,200 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * ONAP Policy Models
- * ================================================================================
- * Copyright (C) 2019, 2021 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) {
- var 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/BaseRestControllerV1.java b/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/BaseRestControllerV1.java
deleted file mode 100644
index 84b610a8f..000000000
--- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/BaseRestControllerV1.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * Copyright (C) 2019, 2023 Nordix Foundation.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * 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 jakarta.ws.rs.Consumes;
-import jakarta.ws.rs.Produces;
-import jakarta.ws.rs.core.Response.ResponseBuilder;
-import java.net.HttpURLConnection;
-import java.util.UUID;
-
-/**
- * Version v1 common superclass to provide DMaaP endpoints for the DMaaP simulator component.
- */
-@Produces("application/json")
-@Consumes({"application/cambria", "application/json"})
-public class BaseRestControllerV1 {
- public static final String EXTENSION_NAME = "interface info";
-
- public static final String API_VERSION_NAME = "api-version";
- public static final String API_VERSION = "1.0.0";
-
- public static final String LAST_MOD_NAME = "last-mod-release";
- public static final String LAST_MOD_RELEASE = "Dublin";
-
- public static final String VERSION_MINOR_NAME = "X-MinorVersion";
- public static final String VERSION_MINOR_DESCRIPTION =
- "Used to request or communicate a MINOR version back from the client"
- + " to the server, and from the server back to the client";
-
- public static final String VERSION_PATCH_NAME = "X-PatchVersion";
- public static final String VERSION_PATCH_DESCRIPTION = "Used only to communicate a PATCH version in a response for"
- + " troubleshooting purposes only, and will not be provided by" + " the client on request";
-
- public static final String VERSION_LATEST_NAME = "X-LatestVersion";
- public static final String VERSION_LATEST_DESCRIPTION = "Used only to communicate an API's latest version";
-
- public static final String REQUEST_ID_NAME = "X-ONAP-RequestID";
- public static final String REQUEST_ID_HDR_DESCRIPTION = "Used to track REST transactions for logging purpose";
- public static final String REQUEST_ID_PARAM_DESCRIPTION = "RequestID for http transaction";
-
- public static final String AUTHORIZATION_TYPE = "basicAuth";
-
- public static final int AUTHENTICATION_ERROR_CODE = HttpURLConnection.HTTP_UNAUTHORIZED;
- public static final int AUTHORIZATION_ERROR_CODE = HttpURLConnection.HTTP_FORBIDDEN;
- public static final int SERVER_ERROR_CODE = HttpURLConnection.HTTP_INTERNAL_ERROR;
-
- public static final String AUTHENTICATION_ERROR_MESSAGE = "Authentication Error";
- public static final String AUTHORIZATION_ERROR_MESSAGE = "Authorization Error";
- public static final String SERVER_ERROR_MESSAGE = "Internal Server Error";
-
- /**
- * Adds version headers to the response.
- *
- * @param respBuilder response builder
- * @return the response builder, with version headers
- */
- public ResponseBuilder addVersionControlHeaders(ResponseBuilder respBuilder) {
- return respBuilder.header(VERSION_MINOR_NAME, "0").header(VERSION_PATCH_NAME, "0").header(VERSION_LATEST_NAME,
- API_VERSION);
- }
-
- /**
- * Adds logging headers to the response.
- *
- * @param respBuilder response builder
- * @return the response builder, with version logging
- */
- public ResponseBuilder addLoggingHeaders(ResponseBuilder respBuilder, UUID requestId) {
- if (requestId == null) {
- // Generate a random uuid if client does not embed requestId in rest request
- return respBuilder.header(REQUEST_ID_NAME, UUID.randomUUID());
- }
-
- return respBuilder.header(REQUEST_ID_NAME, requestId);
- }
-}
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
deleted file mode 100644
index 0efab160b..000000000
--- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/CambriaMessageBodyHandler.java
+++ /dev/null
@@ -1,177 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * ONAP Policy Models
- * ================================================================================
- * Copyright (C) 2019, 2021 AT&T Intellectual Property. All rights reserved.
- * Modifications Copyright (C) 2023 Nordix Foundation.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * 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 jakarta.ws.rs.Consumes;
-import jakarta.ws.rs.core.MediaType;
-import jakarta.ws.rs.core.MultivaluedMap;
-import jakarta.ws.rs.ext.MessageBodyReader;
-import jakarta.ws.rs.ext.Provider;
-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 org.apache.commons.io.IOUtils;
-
-/**
- * Provider that decodes "application/cambria" messages.
- */
-@Provider
-@Consumes(CambriaMessageBodyHandler.MEDIA_TYPE_APPLICATION_CAMBRIA)
-public class CambriaMessageBodyHandler implements MessageBodyReader<Object> {
- public static final String MEDIA_TYPE_APPLICATION_CAMBRIA = "application/cambria";
-
- /**
- * 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 (mediaType != null && MEDIA_TYPE_APPLICATION_CAMBRIA.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 (var 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 {
- var bldr = new StringBuilder(MAX_DIGITS);
-
- int chr;
- for (var x = 0; x < MAX_DIGITS; ++x) {
- if ((chr = reader.read()) < 0) {
- throw new EOFException("missing '.' in 'length' field");
- }
-
- if (chr == '.') {
- String text = bldr.toString().trim();
- return (text.isEmpty() ? 0 : Integer.parseInt(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 {
- var 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
deleted file mode 100644
index 5ed04f1a3..000000000
--- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/DmaapSimRestControllerV1.java
+++ /dev/null
@@ -1,91 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * Copyright (C) 2019, 2021, 2023 Nordix Foundation.
- * Modifications Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
- * Modifications Copyright (C) 2021 Bell Canada. 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.
- *
- * SPDX-License-Identifier: Apache-2.0
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.policy.models.sim.dmaap.rest;
-
-import jakarta.ws.rs.Consumes;
-import jakarta.ws.rs.DefaultValue;
-import jakarta.ws.rs.GET;
-import jakarta.ws.rs.POST;
-import jakarta.ws.rs.Path;
-import jakarta.ws.rs.PathParam;
-import jakarta.ws.rs.Produces;
-import jakarta.ws.rs.QueryParam;
-import jakarta.ws.rs.core.Response;
-import org.onap.policy.models.sim.dmaap.provider.DmaapSimProvider;
-
-/**
- * Class to provide REST endpoints for DMaaP simulator component statistics.
- */
-@Path("/")
-@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.
- *
- * @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 timeoutMs timeout for the message
- * @return the message
- */
- @GET
- @Path("events/{topicName}/{consumerGroup}/{consumerId}")
- 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 DmaapSimProvider.getInstance().processDmaapMessageGet(topicName, consumerGroup, consumerId, limit,
- timeoutMs);
- }
-
- /**
- * Post a DMaaP message.
- *
- * @param topicName topic to get message from
- * @return the response to the post
- */
- @POST
- @Path("events/{topicName}")
- @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 DmaapSimProvider.getInstance().processDmaapMessagePut(topicName, dmaapMessage);
- }
-
- /**
- * Get the list of topics configured.
- *
- * @return the message
- */
- @GET
- @Path("topics")
- public Response getDmaapTopics() {
-
- return DmaapSimProvider.getInstance().processDmaapTopicsGet();
- }
-}
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
deleted file mode 100644
index 6d3bd730e..000000000
--- a/models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/TextMessageBodyHandler.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * ONAP Policy Models
- * ================================================================================
- * Copyright (C) 2019, 2021 AT&T Intellectual Property. All rights reserved.
- * Modifications Copyright (C) 2023 Nordix Foundation.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * 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 jakarta.ws.rs.Consumes;
-import jakarta.ws.rs.core.MediaType;
-import jakarta.ws.rs.core.MultivaluedMap;
-import jakarta.ws.rs.ext.MessageBodyReader;
-import jakarta.ws.rs.ext.Provider;
-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;
-
-/**
- * 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 (var 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;
- }
- }
-}