ab733d77704511adb03df27a7910a4bf38cfc670
[policy/clamp.git] /
1 /*-
2  * ============LICENSE_START=======================================================
3  * Copyright (C) 2022,2024 Nordix Foundation.
4  * ================================================================================
5  * Licensed under the Apache License, Version 2.0 (the "License");
6  * you may not use this file except in compliance with the License.
7  * You may obtain a copy of the License at
8  *
9  *      http://www.apache.org/licenses/LICENSE-2.0
10  *
11  * Unless required by applicable law or agreed to in writing, software
12  * distributed under the License is distributed on an "AS IS" BASIS,
13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14  * See the License for the specific language governing permissions and
15  * limitations under the License.
16  *
17  * SPDX-License-Identifier: Apache-2.0
18  * ============LICENSE_END=========================================================
19  */
20
21 package org.onap.policy.clamp.acm.element.handler;
22
23 import io.micrometer.core.annotation.Timed;
24 import jakarta.ws.rs.core.Response;
25 import java.util.List;
26 import org.onap.policy.clamp.acm.element.handler.messages.ElementMessage;
27 import org.onap.policy.clamp.common.acm.exception.AutomationCompositionRuntimeException;
28 import org.onap.policy.common.endpoints.event.comm.TopicSink;
29 import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClient;
30 import org.slf4j.Logger;
31 import org.slf4j.LoggerFactory;
32 import org.springframework.stereotype.Component;
33
34 @Component
35 public class MessagePublisher {
36     private static final Logger LOGGER = LoggerFactory.getLogger(MessagePublisher.class);
37     private static final String NOT_ACTIVE_TEXT = "Not Active!";
38
39     private boolean active = false;
40     private TopicSinkClient topicSinkClient;
41
42     /**
43      * Constructor for instantiating MessagePublisher.
44      *
45      * @param topicSinks the topic sinks
46      */
47     public void active(List<TopicSink> topicSinks) {
48         if (topicSinks.size() != 1) {
49             throw new IllegalArgumentException("Configuration unsupported, Topic sinks greater than 1");
50         }
51         this.topicSinkClient = new TopicSinkClient(topicSinks.get(0));
52         active = true;
53     }
54
55     /**
56      * Method to send message.
57      *
58      * @param msg the acknowledgement message
59      */
60     @Timed(value = "publisher.status", description = "STATUS messages published")
61     public void publishMsg(final ElementMessage msg) {
62         if (!active) {
63             throw new AutomationCompositionRuntimeException(Response.Status.CONFLICT, NOT_ACTIVE_TEXT);
64         }
65
66         topicSinkClient.send(msg);
67         LOGGER.debug("Sent message {}", msg);
68     }
69
70     public void stop() {
71         active = false;
72     }
73 }