2 * ============LICENSE_START=======================================================
3 * Copyright (C) 2019 Nordix Foundation.
4 * Modifications Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
5 * ================================================================================
6 * Licensed under the Apache License, Version 2.0 (the "License");
7 * you may not use this file except in compliance with the License.
8 * You may obtain a copy of the License at
10 * http://www.apache.org/licenses/LICENSE-2.0
12 * Unless required by applicable law or agreed to in writing, software
13 * distributed under the License is distributed on an "AS IS" BASIS,
14 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15 * See the License for the specific language governing permissions and
16 * limitations under the License.
18 * SPDX-License-Identifier: Apache-2.0
19 * ============LICENSE_END=========================================================
22 package org.onap.policy.models.sim.dmaap.provider;
24 import java.util.Collections;
25 import java.util.LinkedHashMap;
26 import java.util.List;
28 import java.util.concurrent.ConcurrentHashMap;
29 import java.util.concurrent.Executors;
30 import java.util.concurrent.ScheduledExecutorService;
31 import java.util.concurrent.TimeUnit;
32 import javax.ws.rs.core.Response;
33 import javax.ws.rs.core.Response.Status;
36 import org.onap.policy.common.utils.services.ServiceManagerContainer;
37 import org.onap.policy.models.sim.dmaap.parameters.DmaapSimParameterGroup;
38 import org.slf4j.Logger;
39 import org.slf4j.LoggerFactory;
42 * Provider to simulate DMaaP.
44 * @author Liam Fallon (liam.fallon@est.tech)
46 public class DmaapSimProvider extends ServiceManagerContainer {
47 private static final Logger LOGGER = LoggerFactory.getLogger(DmaapSimProvider.class);
51 private static DmaapSimProvider instance;
54 * Maps a topic name to its data.
56 private final Map<String, TopicData> topic2data = new ConcurrentHashMap<>();
59 * Thread used to remove idle consumers from the topics.
61 private ScheduledExecutorService timerPool;
65 * Constructs the object.
67 * @param params parameters
69 public DmaapSimProvider(DmaapSimParameterGroup params) {
70 addAction("Topic Sweeper", () -> {
71 timerPool = makeTimerPool();
72 timerPool.scheduleWithFixedDelay(new SweeperTask(), params.getTopicSweepSec(), params.getTopicSweepSec(),
74 }, () -> timerPool.shutdown());
78 * Process a DMaaP message.
80 * @param topicName the topic name
81 * @param dmaapMessage the message to process
82 * @return a response to the message
84 @SuppressWarnings("unchecked")
85 public Response processDmaapMessagePut(final String topicName, final Object dmaapMessage) {
86 LOGGER.debug("Topic: {}, Received DMaaP message(s): {}", topicName, dmaapMessage);
90 if (dmaapMessage instanceof List) {
91 lst = (List<Object>) dmaapMessage;
93 lst = Collections.singletonList(dmaapMessage);
96 TopicData topic = topic2data.get(topicName);
99 * Write all messages and return the count. If the topic doesn't exist yet, then
100 * there are no subscribers to receive the messages, thus treat it as if all
101 * messages were published.
103 int nmessages = (topic != null ? topic.write(lst) : lst.size());
105 Map<String, Object> map = new LinkedHashMap<>();
106 map.put("serverTimeMs", 0);
107 map.put("count", nmessages);
109 return Response.status(Response.Status.OK).entity(map).build();
113 * Wait for and return a DMaaP message.
115 * @param topicName The topic to wait on
116 * @param consumerGroup the consumer group that is waiting
117 * @param consumerId the consumer ID that is waiting
118 * @param limit the maximum number of messages to get
119 * @param timeoutMs the length of time to wait for
120 * @return the DMaaP message or
122 public Response processDmaapMessageGet(final String topicName, final String consumerGroup, final String consumerId,
123 final int limit, final long timeoutMs) {
125 LOGGER.debug("Topic: {}, Request for DMaaP message: {}: {} with limit={} timeout={}", topicName, consumerGroup,
126 consumerId, limit, timeoutMs);
129 List<String> lst = topic2data.computeIfAbsent(topicName, this::makeTopicData).read(consumerGroup, limit,
132 LOGGER.debug("Topic: {}, Retrieved {} messages for: {}: {}", topicName, lst.size(), consumerGroup,
134 return Response.status(Status.OK).entity(lst).build();
136 } catch (InterruptedException e) {
137 LOGGER.warn("Topic: {}, Request for DMaaP message interrupted: {}: {}", topicName, consumerGroup,
139 Thread.currentThread().interrupt();
140 return Response.status(Status.GONE).entity(Collections.emptyList()).build();
145 * Task to remove idle consumers from each topic.
147 private class SweeperTask implements Runnable {
150 topic2data.values().forEach(TopicData::removeIdleConsumers);
154 // the following methods may be overridden by junit tests
157 * Makes a new timer pool.
159 * @return a new timer pool
161 protected ScheduledExecutorService makeTimerPool() {
162 return Executors.newScheduledThreadPool(1);
168 * @param topicName topic name
169 * @return a new topic
171 protected TopicData makeTopicData(String topicName) {
172 return new TopicData(topicName);