diff options
Diffstat (limited to 'models-sim/models-sim-dmaap/src/main/java')
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; - } - } -} |