* ============LICENSE_START=======================================================
* PolicyEngineUtils
* ================================================================================
- * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2017, 2019 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.
import java.util.List;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
+
import org.onap.dmaap.mr.client.impl.MRSimplerBatchPublisher;
import org.onap.dmaap.mr.test.clients.ProtocolTypeConstants;
import org.onap.policy.common.logging.eelf.PolicyLogger;
import org.onap.policy.common.logging.flexlogger.Logger;
public interface BusPublisher {
-
- /**
- * sends a message
- *
- * @param partition id
- * @param message the message
- * @return true if success, false otherwise
- * @throws IllegalArgumentException if no message provided
- */
- public boolean send(String partitionId, String message);
-
- /**
- * closes the publisher
- */
- public void close();
-
- /**
- * DmaapClient library wrapper
- */
- public static class DmaapPublisherWrapper implements BusPublisher {
- private static Logger logger = FlexLogger.getLogger(DmaapPublisherWrapper.class);
- /**
- * MR based Publisher
- */
- protected MRSimplerBatchPublisher publisher;
-
- public DmaapPublisherWrapper(List<String> servers, String topic,
- String aafLogin,
- String aafPassword) {
-
- ArrayList<String> dmaapServers = new ArrayList<>();
- for (String server: servers) {
- dmaapServers.add(server + ":3904");
- }
-
- this.publisher =
- new MRSimplerBatchPublisher.Builder().
- againstUrls(dmaapServers).
- onTopic(topic).
- build();
-
- this.publisher.setProtocolFlag(ProtocolTypeConstants.AAF_AUTH.getValue());
-
- this.publisher.setUsername(aafLogin);
- this.publisher.setPassword(aafPassword);
-
- Properties props = new Properties();
- props.setProperty("Protocol", "http");
- props.setProperty("contenttype", "application/json");
-
- this.publisher.setProps(props);
-
- this.publisher.setHost(servers.get(0));
-
- if (PolicyLogger.isInfoEnabled())
- PolicyLogger.info(DmaapPublisherWrapper.class.getName(),
- "CREATION: " + this);
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public void close() {
- if (logger.isInfoEnabled())
- logger.info(DmaapPublisherWrapper.class.getName()+
- "CREATION: " + this);
-
- try {
- this.publisher.close(1, TimeUnit.SECONDS);
- } catch (Exception e) {
- logger.warn(DmaapPublisherWrapper.class.getName()+
- "CLOSE: " + this + " because of " +
- e.getMessage(), e);
- }
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public boolean send(String partitionId, String message){
- if (message == null)
- throw new IllegalArgumentException("No message provided");
-
- this.publisher.send(partitionId, message);
- return true;
-
- }
-
- @Override
- public String toString() {
- StringBuilder builder = new StringBuilder();
- builder.append("DmaapPublisherWrapper [").
- append("publisher.getAuthDate()=").append(publisher.getAuthDate()).
- append(", publisher.getAuthKey()=").append(publisher.getAuthKey()).
- append(", publisher.getHost()=").append(publisher.getHost()).
- append(", publisher.getProtocolFlag()=").append(publisher.getProtocolFlag()).
- append(", publisher.getUsername()=").append(publisher.getUsername()).
- append(", publisher.getPendingMessageCount()=").append(publisher.getPendingMessageCount()).
- append("]");
- return builder.toString();
- }
- }
+
+ /**
+ * sends a message
+ * .
+ * @param partitionId id
+ * @param message the message
+ * @return true if success, false otherwise
+ * @throws IllegalArgumentException if no message provided
+ */
+ public boolean send(String partitionId, String message);
+
+ /**
+ * closes the publisher.
+ */
+ public void close();
+
+ /**
+ * DmaapClient library wrapper.
+ */
+ public static class DmaapPublisherWrapper implements BusPublisher {
+ private static Logger logger = FlexLogger.getLogger(DmaapPublisherWrapper.class);
+ /**
+ * MR based Publisher.
+ */
+ protected MRSimplerBatchPublisher publisher;
+
+ /**
+ * DmaapPublisherWrapper constructor.
+ *
+ * @param servers list of servers
+ * @param topic topic
+ * @param aafLogin login
+ * @param aafPassword password
+ */
+ public DmaapPublisherWrapper(List<String> servers, String topic, String aafLogin, String aafPassword) {
+
+ ArrayList<String> dmaapServers = new ArrayList<>();
+ for (String server : servers) {
+ dmaapServers.add(server + ":3904");
+ }
+
+ this.publisher = new MRSimplerBatchPublisher.Builder().againstUrls(dmaapServers).onTopic(topic).build();
+
+ this.publisher.setProtocolFlag(ProtocolTypeConstants.AAF_AUTH.getValue());
+
+ this.publisher.setUsername(aafLogin);
+ this.publisher.setPassword(aafPassword);
+
+ Properties props = new Properties();
+ props.setProperty("Protocol", "http");
+ props.setProperty("contenttype", "application/json");
+
+ this.publisher.setProps(props);
+
+ this.publisher.setHost(servers.get(0));
+
+ if (PolicyLogger.isInfoEnabled()) {
+ PolicyLogger.info(DmaapPublisherWrapper.class.getName(), "CREATION: " + this);
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void close() {
+ if (logger.isInfoEnabled()) {
+ logger.info(DmaapPublisherWrapper.class.getName() + "CREATION: " + this);
+ }
+
+ try {
+ this.publisher.close(1, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ logger.warn(DmaapPublisherWrapper.class.getName() + "CLOSE: " + this + " because of " + e.getMessage(),
+ e);
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean send(String partitionId, String message) {
+ if (message == null) {
+ throw new IllegalArgumentException("No message provided");
+ }
+
+ this.publisher.send(partitionId, message);
+ return true;
+
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder builder = new StringBuilder();
+ builder.append("DmaapPublisherWrapper [").append("publisher.getAuthDate()=").append(publisher.getAuthDate())
+ .append(", publisher.getAuthKey()=").append(publisher.getAuthKey()).append(", publisher.getHost()=")
+ .append(publisher.getHost()).append(", publisher.getProtocolFlag()=")
+ .append(publisher.getProtocolFlag()).append(", publisher.getUsername()=")
+ .append(publisher.getUsername()).append(", publisher.getPendingMessageCount()=")
+ .append(publisher.getPendingMessageCount()).append("]");
+ return builder.toString();
+ }
+ }
}
Code Review / policy / engine.git / blobdiff