054eaf7b5118f0e8fd686927de667133631ea9ed
[policy/clamp.git] /
1 /*-
2  * ============LICENSE_START=======================================================
3  *  Copyright (C) 2021 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.runtime.supervision.comm;
22
23 import java.util.List;
24 import javax.ws.rs.core.Response.Status;
25 import org.onap.policy.clamp.acm.runtime.config.messaging.Publisher;
26 import org.onap.policy.clamp.common.acm.exception.AutomationCompositionRuntimeException;
27 import org.onap.policy.clamp.models.acm.messages.dmaap.participant.ParticipantMessage;
28 import org.onap.policy.common.endpoints.event.comm.TopicSink;
29 import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClient;
30
31 public abstract class AbstractParticipantPublisher<E extends ParticipantMessage> implements Publisher {
32
33     private TopicSinkClient topicSinkClient;
34     private boolean active = false;
35
36     /**
37      * Method to send Participant message to participants on demand.
38      *
39      * @param participantMessage the Participant message
40      */
41     public void send(final E participantMessage) {
42         if (!active) {
43             throw new AutomationCompositionRuntimeException(Status.NOT_ACCEPTABLE, "Not Active!");
44         }
45         topicSinkClient.send(participantMessage);
46     }
47
48
49     @Override
50     public void active(List<TopicSink> topicSinks) {
51         if (topicSinks.size() != 1) {
52             throw new IllegalArgumentException("Topic Sink must be one");
53         }
54         this.topicSinkClient = new TopicSinkClient(topicSinks.get(0));
55         active = true;
56     }
57
58     @Override
59     public void stop() {
60         active = false;
61     }
62 }