2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright (C) 2017, 2019 AT&T Intellectual Property. All rights reserved.
6 * ================================================================================
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 * ============LICENSE_END=========================================================
21 package org.onap.policy.utils;
23 import java.util.ArrayList;
24 import java.util.List;
25 import java.util.Properties;
26 import java.util.concurrent.TimeUnit;
28 import org.onap.dmaap.mr.client.impl.MRSimplerBatchPublisher;
29 import org.onap.dmaap.mr.test.clients.ProtocolTypeConstants;
30 import org.onap.policy.common.logging.eelf.PolicyLogger;
31 import org.onap.policy.common.logging.flexlogger.FlexLogger;
32 import org.onap.policy.common.logging.flexlogger.Logger;
34 public interface BusPublisher {
39 * @param partitionId id
40 * @param message the message
41 * @return true if success, false otherwise
42 * @throws IllegalArgumentException if no message provided
44 public boolean send(String partitionId, String message);
47 * closes the publisher.
52 * DmaapClient library wrapper.
54 public static class DmaapPublisherWrapper implements BusPublisher {
55 private static Logger logger = FlexLogger.getLogger(DmaapPublisherWrapper.class);
59 protected MRSimplerBatchPublisher publisher;
62 * DmaapPublisherWrapper constructor.
64 * @param servers list of servers
66 * @param aafLogin login
67 * @param aafPassword password
69 public DmaapPublisherWrapper(List<String> servers, String topic, String aafLogin, String aafPassword) {
71 ArrayList<String> dmaapServers = new ArrayList<>();
72 for (String server : servers) {
73 dmaapServers.add(server + ":3904");
76 this.publisher = new MRSimplerBatchPublisher.Builder().againstUrls(dmaapServers).onTopic(topic).build();
78 this.publisher.setProtocolFlag(ProtocolTypeConstants.AAF_AUTH.getValue());
80 this.publisher.setUsername(aafLogin);
81 this.publisher.setPassword(aafPassword);
83 Properties props = new Properties();
84 props.setProperty("Protocol", "http");
85 props.setProperty("contenttype", "application/json");
87 this.publisher.setProps(props);
89 this.publisher.setHost(servers.get(0));
91 if (PolicyLogger.isInfoEnabled()) {
92 PolicyLogger.info(DmaapPublisherWrapper.class.getName(), "CREATION: " + this);
100 public void close() {
101 if (logger.isInfoEnabled()) {
102 logger.info(DmaapPublisherWrapper.class.getName() + "CREATION: " + this);
106 this.publisher.close(1, TimeUnit.SECONDS);
107 } catch (Exception e) {
108 logger.warn(DmaapPublisherWrapper.class.getName() + "CLOSE: " + this + " because of " + e.getMessage(),
117 public boolean send(String partitionId, String message) {
118 if (message == null) {
119 throw new IllegalArgumentException("No message provided");
122 this.publisher.send(partitionId, message);
128 public String toString() {
129 StringBuilder builder = new StringBuilder();
130 builder.append("DmaapPublisherWrapper [").append("publisher.getAuthDate()=").append(publisher.getAuthDate())
131 .append(", publisher.getAuthKey()=").append(publisher.getAuthKey()).append(", publisher.getHost()=")
132 .append(publisher.getHost()).append(", publisher.getProtocolFlag()=")
133 .append(publisher.getProtocolFlag()).append(", publisher.getUsername()=")
134 .append(publisher.getUsername()).append(", publisher.getPendingMessageCount()=")
135 .append(publisher.getPendingMessageCount()).append("]");
136 return builder.toString();