diff options
Diffstat (limited to 'message-router/publisher/provider')
3 files changed, 202 insertions, 0 deletions
diff --git a/message-router/publisher/provider/pom.xml b/message-router/publisher/provider/pom.xml new file mode 100755 index 000000000..ca51aea42 --- /dev/null +++ b/message-router/publisher/provider/pom.xml @@ -0,0 +1,39 @@ +<?xml version="1.0" encoding="UTF-8"?> +<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> + <modelVersion>4.0.0</modelVersion> + + <parent> + <groupId>org.onap.ccsdk.messagerouter</groupId> + <artifactId>publisher.aggregate</artifactId> + <version>0.4.1-SNAPSHOT</version> + </parent> + + <artifactId>publisher.provider</artifactId> + <packaging>bundle</packaging> + + <dependencies> + <dependency> + <groupId>org.onap.ccsdk.messagerouter</groupId> + <artifactId>publisher.api</artifactId> + <version>${project.version}</version> + </dependency> + <dependency> + <groupId>org.slf4j</groupId> + <artifactId>slf4j-api</artifactId> + </dependency> + </dependencies> + + <build> + <plugins> + <plugin> + <groupId>org.apache.felix</groupId> + <artifactId>maven-bundle-plugin</artifactId> + <configuration> + <instructions> + <Private-Package>org.onap.ccsdk.messagerouter.publisher.provider.impl</Private-Package> + </instructions> + </configuration> + </plugin> + </plugins> + </build> +</project> diff --git a/message-router/publisher/provider/src/main/java/org/onap/ccsdk/messagerouter/publisher/provider/impl/PublisherApiImpl.java b/message-router/publisher/provider/src/main/java/org/onap/ccsdk/messagerouter/publisher/provider/impl/PublisherApiImpl.java new file mode 100755 index 000000000..3e8ab3333 --- /dev/null +++ b/message-router/publisher/provider/src/main/java/org/onap/ccsdk/messagerouter/publisher/provider/impl/PublisherApiImpl.java @@ -0,0 +1,149 @@ +package org.onap.ccsdk.messagerouter.publisher.provider.impl;
+
+import java.io.BufferedReader;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.net.HttpURLConnection;
+import java.net.SocketException;
+import java.net.URL;
+import java.util.Base64;
+
+import org.onap.ccsdk.messagerouter.publisher.api.PublisherApi;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PublisherApiImpl implements PublisherApi {
+ private static final Logger logger = LoggerFactory.getLogger(PublisherApiImpl.class);
+ protected final Integer DEFAULT_CONNECT_TIMEOUT = 30000; // will be treated as 30 seconds
+ protected final Integer DEFAULT_READ_TIMEOUT = 180000; // will be treated as 3 minutes
+ private String authorizationString;
+ protected Integer connectTimeout;
+ protected Integer readTimeout;
+ protected String baseUrl;
+ protected String username;
+ protected String[] hosts;
+ private String password;
+
+ public void setUsername(String username) {
+ this.username = username;
+ buildAuthorizationString();
+ }
+
+ public void setPassword(String password) {
+ this.password = password;
+ buildAuthorizationString();
+ }
+
+ public void setHost(String hostString) {
+ // a comma separated list of hosts can be passed in or a single host may be used
+ if (!hostString.contains(",")) {
+ this.hosts = new String[] { hostString };
+ } else {
+ this.hosts = hostString.split(",");
+ }
+ }
+
+ public PublisherApiImpl() {
+ connectTimeout = DEFAULT_CONNECT_TIMEOUT;
+ readTimeout = DEFAULT_READ_TIMEOUT;
+ }
+
+ public void init() {
+ buildAuthorizationString();
+ }
+
+ private String buildUrlString(Integer hostIndex, String topic) {
+ return hosts[hostIndex] + "/events/" + topic;
+ }
+
+ protected void configureHttpURLConnection(HttpURLConnection httpUrlConnection) {
+ httpUrlConnection.setRequestProperty("Content-Type", "application/json");
+ }
+
+ public Boolean publish(String topic, String body) {
+ for (int hostIndex = 0; hostIndex < hosts.length; hostIndex++) {
+ HttpURLConnection httpUrlConnection = null;
+ URL url = null;
+ try {
+ url = new URL(buildUrlString(hostIndex, topic));
+ logger.info("Publishing body to topic {} using the url {}", topic, url);
+ logger.info("Message to publish is\n{}", body);
+ httpUrlConnection = buildHttpURLConnection(url);
+ httpUrlConnection.setDoInput(true);
+ httpUrlConnection.setDoOutput(true);
+ httpUrlConnection.setUseCaches(false);
+ httpUrlConnection.setRequestMethod("POST");
+
+ // Write message
+ httpUrlConnection.setRequestProperty("Content-Length", Integer.toString(body.length()));
+ DataOutputStream outStr = new DataOutputStream(httpUrlConnection.getOutputStream());
+ outStr.write(body.getBytes());
+ outStr.close();
+
+ int status = httpUrlConnection.getResponseCode();
+ logger.info("Publishing body for topic {} using url {} returned status {}.", topic, url, status);
+ if (status < 300) {
+ String responseFromDMaaP = readFromStream(httpUrlConnection.getInputStream());
+ logger.info("Message router response is\n{}", responseFromDMaaP);
+ return true;
+ } else {
+ if (httpUrlConnection.getErrorStream() != null) {
+ String responseFromDMaaP = readFromStream(httpUrlConnection.getErrorStream());
+ logger.warn("Publishing body for topic {} using url {} failed." + " Error message is\n{}",
+ topic, url, responseFromDMaaP);
+ }
+ return false;
+ }
+
+ } catch (SocketException socketException) {
+ logger.error("SocketException was thrown during publishing message to DMaaP on url {}.", url,
+ socketException);
+ if (hostIndex < hosts.length) {
+ logger.info("Message sent to {} failed with a SocketException, but will be tried on {}",
+ hosts[hostIndex], hosts[hostIndex + 1]);
+ }
+ } catch (Exception e) {
+ logger.warn("Exception was thrown during publishing message to DMaaP on url {}.", url, e);
+ return false;
+ } finally {
+ if (httpUrlConnection != null) {
+ httpUrlConnection.disconnect();
+ }
+ }
+ }
+ return false;
+ }
+
+ private void buildAuthorizationString() {
+ String basicAuthString = username + ":" + password;
+ basicAuthString = Base64.getEncoder().encodeToString(basicAuthString.getBytes());
+ this.authorizationString = "Basic " + basicAuthString;
+ }
+
+ protected HttpURLConnection buildHttpURLConnection(URL url) throws IOException {
+ HttpURLConnection httpUrlConnection = (HttpURLConnection) url.openConnection();
+ if (authorizationString != null) {
+ httpUrlConnection.setRequestProperty("Authorization", authorizationString);
+ }
+ httpUrlConnection.setRequestProperty("Accept", "application/json");
+ httpUrlConnection.setUseCaches(false);
+ httpUrlConnection.setConnectTimeout(connectTimeout);
+ httpUrlConnection.setReadTimeout(readTimeout);
+ configureHttpURLConnection(httpUrlConnection);
+ return httpUrlConnection;
+ }
+
+ protected String readFromStream(InputStream stream) throws IOException {
+ BufferedReader br = new BufferedReader(new InputStreamReader(stream));
+ StringBuilder sb = new StringBuilder();
+ String line;
+ while ((line = br.readLine()) != null) {
+ sb.append(line);
+ }
+ br.close();
+ return sb.toString();
+ }
+
+}
\ No newline at end of file diff --git a/message-router/publisher/provider/src/main/resources/OSGI-INF/blueprint/blueprint.xml b/message-router/publisher/provider/src/main/resources/OSGI-INF/blueprint/blueprint.xml new file mode 100755 index 000000000..da25fd23d --- /dev/null +++ b/message-router/publisher/provider/src/main/resources/OSGI-INF/blueprint/blueprint.xml @@ -0,0 +1,14 @@ +<?xml version="1.0" encoding="UTF-8"?> +<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0" + xmlns:cm="http://aries.apache.org/blueprint/xmlns/blueprint-cm/v1.2.0"> + + <bean id="publisher" + class="org.onap.ccsdk.messagerouter.publisher.provider.impl.PublisherApiImpl"> + <cm:managed-properties + persistent-id="org.onap.ccsdk.messagerouter.publisher.provider" + update-strategy="container-managed" /> + </bean> + + <service ref="publisher" interface="org.onap.ccsdk.messagerouter.publisher.api.PublisherApi" /> + +</blueprint>
\ No newline at end of file |